WRNexusJS/packages/queue/src/sqlite.ts

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,
      }));
    },
  };
}