feat(queue): add typed application queue lifecycle

2026-08-23 09:22:05 +05:30
parent 9332522614
commit 9377a69d88
17 changed files with 667 additions and 13 deletions
@@ -0,0 +1,167 @@
+import type { Job } from "./index.ts";
+import type { QueueJobRecord, QueueStore } from "./durable.ts";
+
+export interface SqliteQueueClient {
+  exec(sql: string, parameters?: unknown[]): Promise<{ changes: number }>;
+  one<T>(sql: string, parameters?: unknown[]): Promise<T | null>;
+  all<T>(sql: string, parameters?: unknown[]): Promise<T[]>;
+}
+
+interface QueueJobRow {
+  id: string;
+  name: string;
+  payload: string;
+  run_at: number;
+  priority: number;
+  idempotency_key: string | null;
+}
+
+const SAFE_TABLE = /^[a-z_][a-z0-9_]*$/i;
+
+export function sqliteQueueSchema(table = "wrnexus_jobs"): string {
+  if (!SAFE_TABLE.test(table)) throw new Error("Invalid queue table name");
+  return `CREATE TABLE IF NOT EXISTS ${table} (
+    id TEXT PRIMARY KEY,
+    name TEXT NOT NULL,
+    payload TEXT NOT NULL,
+    run_at INTEGER NOT NULL,
+    priority INTEGER NOT NULL DEFAULT 0,
+    idempotency_key TEXT,
+    lease_owner TEXT,
+    lease_until INTEGER
+  );
+  CREATE INDEX IF NOT EXISTS ${table}_due ON ${table} (run_at, priority DESC);
+  CREATE UNIQUE INDEX IF NOT EXISTS ${table}_idempotency ON ${table} (name, idempotency_key)
+    WHERE idempotency_key IS NOT NULL;
+  CREATE TABLE IF NOT EXISTS ${table}_history (
+    id TEXT PRIMARY KEY,
+    name TEXT NOT NULL,
+    state TEXT NOT NULL,
+    payload TEXT NOT NULL,
+    error TEXT,
+    finished_at INTEGER NOT NULL
+  );
+  CREATE INDEX IF NOT EXISTS ${table}_history_finished ON ${table}_history (finished_at DESC);`;
+}
+
+export async function installSqliteQueueSchema(
+  db: SqliteQueueClient,
+  table = "wrnexus_jobs",
+): Promise<void> {
+  for (const statement of sqliteQueueSchema(table)
+    .split(";")
+    .map((value) => value.trim())) {
+    if (statement) await db.exec(statement);
+  }
+}
+
+export function sqliteQueueStore(db: SqliteQueueClient, table = "wrnexus_jobs"): QueueStore {
+  if (!SAFE_TABLE.test(table)) throw new Error("Invalid queue table name");
+  const decode = (row: QueueJobRow): Job => JSON.parse(row.payload) as Job;
+  return {
+    async put(job) {
+      await db.exec(
+        `INSERT INTO ${table} (id,name,payload,run_at,priority,idempotency_key,lease_owner,lease_until)
+         VALUES (?,?,?,?,?,?,NULL,NULL)
+         ON CONFLICT(id) DO UPDATE SET name=excluded.name,payload=excluded.payload,
+         run_at=excluded.run_at,priority=excluded.priority,idempotency_key=excluded.idempotency_key,
+         lease_owner=NULL,lease_until=NULL`,
+        [
+          job.id,
+          job.name,
+          JSON.stringify(job),
+          job.runAt,
+          job.priority,
+          job.idempotencyKey ?? null,
+        ],
+      );
+    },
+    async get(id) {
+      const row = await db.one<QueueJobRow>(`SELECT * FROM ${table} WHERE id = ?`, [id]);
+      return row ? decode(row) : null;
+    },
+    async remove(id) {
+      await db.exec(`DELETE FROM ${table} WHERE id = ?`, [id]);
+    },
+    async due(now, limit) {
+      const rows = await db.all<QueueJobRow>(
+        `SELECT * FROM ${table} WHERE run_at <= ? AND (lease_until IS NULL OR lease_until < ?)
+         ORDER BY priority DESC,run_at ASC LIMIT ?`,
+        [now, now, limit],
+      );
+      return rows.map(decode);
+    },
+    async list(name) {
+      const rows = name
+        ? await db.all<QueueJobRow>(`SELECT * FROM ${table} WHERE name = ? ORDER BY run_at`, [name])
+        : await db.all<QueueJobRow>(`SELECT * FROM ${table} ORDER BY run_at`);
+      return rows.map(decode);
+    },
+    async findByIdempotencyKey(name, key) {
+      const row = await db.one<QueueJobRow>(
+        `SELECT * FROM ${table} WHERE name = ? AND idempotency_key = ? LIMIT 1`,
+        [name, key],
+      );
+      return row ? decode(row) : null;
+    },
+    async size() {
+      const row = await db.one<{ total: number }>(`SELECT COUNT(*) AS total FROM ${table}`);
+      return Number(row?.total ?? 0);
+    },
+    async claim(id, worker, leaseUntil, now = Date.now()) {
+      const result = await db.exec(
+        `UPDATE ${table} SET lease_owner = ?,lease_until = ?
+         WHERE id = ? AND (lease_until IS NULL OR lease_until < ?)`,
+        [worker, leaseUntil, id, now],
+      );
+      return result.changes > 0;
+    },
+    async release(id, worker) {
+      await db.exec(
+        `UPDATE ${table} SET lease_owner = NULL,lease_until = NULL WHERE id = ? AND lease_owner = ?`,
+        [id, worker],
+      );
+    },
+    async archive(record) {
+      await db.exec(
+        `INSERT INTO ${table}_history (id,name,state,payload,error,finished_at) VALUES (?,?,?,?,?,?)
+         ON CONFLICT(id) DO UPDATE SET state=excluded.state,payload=excluded.payload,
+         error=excluded.error,finished_at=excluded.finished_at`,
+        [
+          record.job.id,
+          record.job.name,
+          record.state,
+          JSON.stringify(record.job),
+          record.error ?? null,
+          record.finishedAt,
+        ],
+      );
+    },
+    async history(id) {
+      const rows = id
+        ? await db.all<{
+            state: QueueJobRecord["state"];
+            payload: string;
+            error: string | null;
+            finished_at: number;
+          }>(
+            `SELECT state,payload,error,finished_at FROM ${table}_history WHERE id = ? ORDER BY finished_at DESC`,
+            [id],
+          )
+        : await db.all<{
+            state: QueueJobRecord["state"];
+            payload: string;
+            error: string | null;
+            finished_at: number;
+          }>(
+            `SELECT state,payload,error,finished_at FROM ${table}_history ORDER BY finished_at DESC`,
+          );
+      return rows.map((row) => ({
+        job: JSON.parse(row.payload) as Job,
+        state: row.state,
+        finishedAt: row.finished_at,
+        error: row.error ?? undefined,
+      }));
+    },
+  };
+}