chore: Add a new sqlite based queue package

hoarder-app · Jul 14, 2024 · f77a41a · f77a41a
1 parent aa3dce0
commit f77a41a
Show file tree

Hide file tree

Showing 16 changed files with 1,062 additions and 0 deletions.
diff --git a/packages/queue/db.ts b/packages/queue/db.ts
@@ -0,0 +1,19 @@
+import path from "node:path";
+import Database from "better-sqlite3";
+import { BetterSQLite3Database, drizzle } from "drizzle-orm/better-sqlite3";
+import { migrate } from "drizzle-orm/better-sqlite3/migrator";
+
+import * as schema from "./schema";
+
+export function buildDBClient(dbPath: string, runMigrations = false) {
+  const sqlite = new Database(dbPath);
+  const db = drizzle(sqlite, { schema });
+  if (runMigrations) {
+    migrateDB(db);
+  }
+  return db;
+}
+
+export function migrateDB(db: BetterSQLite3Database<typeof schema>) {
+  migrate(db, { migrationsFolder: path.join(__dirname, "drizzle") });
+}
diff --git a/packages/queue/drizzle.config.ts b/packages/queue/drizzle.config.ts
@@ -0,0 +1,10 @@
+import type { Config } from "drizzle-kit";
+
+export default {
+  schema: "./schema.ts",
+  out: "./drizzle",
+  driver: "better-sqlite",
+  dbCredentials: {
+    url: "data.db",
+  },
+} satisfies Config;
diff --git a/packages/queue/drizzle/0000_wonderful_talisman.sql b/packages/queue/drizzle/0000_wonderful_talisman.sql
@@ -0,0 +1,18 @@
+CREATE TABLE `tasks` (
+	`id` integer PRIMARY KEY AUTOINCREMENT NOT NULL,
+	`queue` text NOT NULL,
+	`payload` text NOT NULL,
+	`createdAt` integer NOT NULL,
+	`status` text DEFAULT 'pending' NOT NULL,
+	`expireAt` integer,
+	`allocationId` text NOT NULL,
+	`numRunsLeft` integer NOT NULL,
+	`maxNumRuns` integer NOT NULL
+);
+--> statement-breakpoint
+CREATE INDEX `tasks_queue_idx` ON `tasks` (`queue`);--> statement-breakpoint
+CREATE INDEX `tasks_status_idx` ON `tasks` (`status`);--> statement-breakpoint
+CREATE INDEX `tasks_expire_at_idx` ON `tasks` (`expireAt`);--> statement-breakpoint
+CREATE INDEX `tasks_num_runs_left_idx` ON `tasks` (`numRunsLeft`);--> statement-breakpoint
+CREATE INDEX `tasks_max_num_runs_idx` ON `tasks` (`maxNumRuns`);--> statement-breakpoint
+CREATE INDEX `tasks_allocation_id_idx` ON `tasks` (`allocationId`);
diff --git a/packages/queue/drizzle/meta/0000_snapshot.json b/packages/queue/drizzle/meta/0000_snapshot.json
@@ -0,0 +1,130 @@
+{
+  "version": "5",
+  "dialect": "sqlite",
+  "id": "3094773c-0138-46b2-b617-4b10093b0f53",
+  "prevId": "00000000-0000-0000-0000-000000000000",
+  "tables": {
+    "tasks": {
+      "name": "tasks",
+      "columns": {
+        "id": {
+          "name": "id",
+          "type": "integer",
+          "primaryKey": true,
+          "notNull": true,
+          "autoincrement": true
+        },
+        "queue": {
+          "name": "queue",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true,
+          "autoincrement": false
+        },
+        "payload": {
+          "name": "payload",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true,
+          "autoincrement": false
+        },
+        "createdAt": {
+          "name": "createdAt",
+          "type": "integer",
+          "primaryKey": false,
+          "notNull": true,
+          "autoincrement": false
+        },
+        "status": {
+          "name": "status",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true,
+          "autoincrement": false,
+          "default": "'pending'"
+        },
+        "expireAt": {
+          "name": "expireAt",
+          "type": "integer",
+          "primaryKey": false,
+          "notNull": false,
+          "autoincrement": false
+        },
+        "allocationId": {
+          "name": "allocationId",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true,
+          "autoincrement": false
+        },
+        "numRunsLeft": {
+          "name": "numRunsLeft",
+          "type": "integer",
+          "primaryKey": false,
+          "notNull": true,
+          "autoincrement": false
+        },
+        "maxNumRuns": {
+          "name": "maxNumRuns",
+          "type": "integer",
+          "primaryKey": false,
+          "notNull": true,
+          "autoincrement": false
+        }
+      },
+      "indexes": {
+        "tasks_queue_idx": {
+          "name": "tasks_queue_idx",
+          "columns": [
+            "queue"
+          ],
+          "isUnique": false
+        },
+        "tasks_status_idx": {
+          "name": "tasks_status_idx",
+          "columns": [
+            "status"
+          ],
+          "isUnique": false
+        },
+        "tasks_expire_at_idx": {
+          "name": "tasks_expire_at_idx",
+          "columns": [
+            "expireAt"
+          ],
+          "isUnique": false
+        },
+        "tasks_num_runs_left_idx": {
+          "name": "tasks_num_runs_left_idx",
+          "columns": [
+            "numRunsLeft"
+          ],
+          "isUnique": false
+        },
+        "tasks_max_num_runs_idx": {
+          "name": "tasks_max_num_runs_idx",
+          "columns": [
+            "maxNumRuns"
+          ],
+          "isUnique": false
+        },
+        "tasks_allocation_id_idx": {
+          "name": "tasks_allocation_id_idx",
+          "columns": [
+            "allocationId"
+          ],
+          "isUnique": false
+        }
+      },
+      "foreignKeys": {},
+      "compositePrimaryKeys": {},
+      "uniqueConstraints": {}
+    }
+  },
+  "enums": {},
+  "_meta": {
+    "schemas": {},
+    "tables": {},
+    "columns": {}
+  }
+}
diff --git a/packages/queue/drizzle/meta/_journal.json b/packages/queue/drizzle/meta/_journal.json
@@ -0,0 +1,13 @@
+{
+  "version": "5",
+  "dialect": "sqlite",
+  "entries": [
+    {
+      "idx": 0,
+      "version": "5",
+      "when": 1720992922192,
+      "tag": "0000_wonderful_talisman",
+      "breakpoints": true
+    }
+  ]
+}
diff --git a/packages/queue/index.ts b/packages/queue/index.ts
@@ -0,0 +1,6 @@
+export { SqliteQueue } from "./queue";
+export { buildDBClient, migrateDB } from "./db";
+export type { SqliteQueueOptions, RunnerOptions, RunnerFuncs } from "./options";
+export { Runner } from "./runner";
+
+export type { DequeuedJob, DequeuedJobError } from "./types";
diff --git a/packages/queue/options.ts b/packages/queue/options.ts
@@ -0,0 +1,22 @@
+import { ZodType } from "zod";
+
+import { DequeuedJob, DequeuedJobError } from "./types";
+
+export interface SqliteQueueOptions {
+  defaultJobArgs: {
+    numRetries: number;
+  };
+}
+
+export interface RunnerFuncs<T> {
+  run: (job: DequeuedJob<T>) => Promise<void>;
+  onComplete?: (job: DequeuedJob<T>) => Promise<void>;
+  onError?: (job: DequeuedJobError<T>) => Promise<void>;
+}
+
+export interface RunnerOptions<T> {
+  pollIntervalMs: number;
+  timeoutSecs: number;
+  concurrency: number;
+  validator?: ZodType<T>;
+}
diff --git a/packages/queue/package.json b/packages/queue/package.json
@@ -0,0 +1,35 @@
+{
+  "$schema": "https://json.schemastore.org/package.json",
+  "name": "@hoarder/queue",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "dependencies": {
+    "better-sqlite3": "^9.4.3",
+    "drizzle-orm": "^0.29.4",
+    "zod": "^3.22.4",
+    "async-mutex": "^0.4.1"
+  },
+  "devDependencies": {
+    "@hoarder/eslint-config": "workspace:^0.2.0",
+    "@hoarder/prettier-config": "workspace:^0.1.0",
+    "@hoarder/tsconfig": "workspace:^0.1.0",
+    "@types/better-sqlite3": "^7.6.9",
+    "drizzle-kit": "^0.20.14",
+    "vitest": "^1.3.1"
+  },
+  "scripts": {
+    "typecheck": "tsc --noEmit",
+    "test": "vitest",
+    "format": "prettier . --ignore-path ../../.prettierignore",
+    "lint": "eslint ."
+  },
+  "main": "index.ts",
+  "eslintConfig": {
+    "root": true,
+    "extends": [
+      "@hoarder/eslint-config/base"
+    ]
+  },
+  "prettier": "@hoarder/prettier-config"
+}