WRNexusJS/packages/queue/src/patterns.ts

import type { Db } from "@wrnexus/db";
import type { AddOptions, Job } from "./index.ts";
import type { DefinedJob } from "./defined.ts";
import type { DurableQueue } from "./durable.ts";

const SAFE = /^[A-Za-z_][A-Za-z0-9_]*$/;
const ident = (value: string) => {
  if (!SAFE.test(value)) throw new TypeError(`Unsafe queue identifier: ${value}`);
  return value;
};

export interface OutboxEntry<T = unknown> {
  id: number;
  queue: string;
  payload: T;
  options?: AddOptions;
}

export interface OutboxWriter {
  enqueue<T>(job: Pick<DefinedJob<T>, "name">, data: T, options?: AddOptions): Promise<void>;
}

export async function installOutboxSchema(db: Db, table = "wrnexus_outbox"): Promise<void> {
  const target = ident(table);
  const id =
    db.driver.dialect === "postgres"
      ? "BIGSERIAL PRIMARY KEY"
      : "INTEGER PRIMARY KEY AUTOINCREMENT";
  await db.exec(`CREATE TABLE IF NOT EXISTS ${target} (
    id ${id}, queue_name TEXT NOT NULL, payload TEXT NOT NULL,
    options TEXT, created_at BIGINT NOT NULL, dispatched_at BIGINT
  )`);
  await db.exec(`CREATE INDEX IF NOT EXISTS ${target}_pending ON ${target} (dispatched_at, id)`);
}

/** Commit business writes and durable enqueue intents in the same database transaction. */
export async function withOutbox<T>(
  db: Db,
  operation: (context: { db: Db; enqueue: OutboxWriter["enqueue"] }) => Promise<T>,
  table = "wrnexus_outbox",
): Promise<T> {
  await installOutboxSchema(db, table);
  const target = ident(table);
  return db.tx((transaction) =>
    operation({
      db: transaction,
      enqueue: async (job, data, options) => {
        await transaction.exec(
          `INSERT INTO ${target} (queue_name,payload,options,created_at,dispatched_at) VALUES (?,?,?,?,NULL)`,
          [job.name, JSON.stringify(data), options ? JSON.stringify(options) : null, Date.now()],
        );
      },
    }),
  );
}

/** Dispatch committed outbox rows. A crash before marking simply retries idempotently. */
export async function drainOutbox(
  db: Db,
  resolve: (
    queueName: string,
  ) => { add(data: unknown, options?: AddOptions): Promise<Job> } | undefined,
  options: { table?: string; limit?: number } = {},
): Promise<number> {
  const target = ident(options.table ?? "wrnexus_outbox");
  await installOutboxSchema(db, target);
  const rows = await db.all<{
    id: number;
    queue_name: string;
    payload: string;
    options: string | null;
  }>(
    `SELECT id,queue_name,payload,options FROM ${target} WHERE dispatched_at IS NULL ORDER BY id LIMIT ?`,
    [options.limit ?? 100],
  );
  let dispatched = 0;
  for (const row of rows) {
    const producer = resolve(row.queue_name);
    if (!producer) continue;
    const addOptions = row.options ? (JSON.parse(row.options) as AddOptions) : {};
    await producer.add(JSON.parse(row.payload), {
      idempotencyKey: `outbox:${row.id}`,
      ...addOptions,
    });
    await db.exec(`UPDATE ${target} SET dispatched_at = ? WHERE id = ? AND dispatched_at IS NULL`, [
      Date.now(),
      row.id,
    ]);
    dispatched++;
  }
  return dispatched;
}

export interface JobStateMachineOptions {
  db: () => Db;
  table: string;
  idColumn?: string;
  stateColumn?: string;
  states: { queued: string; running: string; success: string; failed: string };
}

export function defineJobStateMachine(options: JobStateMachineOptions) {
  const table = ident(options.table);
  const id = ident(options.idColumn ?? "id");
  const state = ident(options.stateColumn ?? "status");
  const transition = async (recordId: string | number, from: string, to: string) =>
    (
      await options
        .db()
        .exec(`UPDATE ${table} SET ${state} = ? WHERE ${id} = ? AND ${state} = ?`, [
          to,
          recordId,
          from,
        ])
    ).changes > 0;
  return {
    queue: (recordId: string | number) =>
      transition(recordId, options.states.failed, options.states.queued),
    claim: (recordId: string | number) =>
      transition(recordId, options.states.queued, options.states.running),
    complete: (recordId: string | number) =>
      transition(recordId, options.states.running, options.states.success),
    fail: (recordId: string | number) =>
      transition(recordId, options.states.running, options.states.failed),
    transition,
  };
}

export function testQueue(queue: DurableQueue) {
  return {
    add: queue.add.bind(queue),
    runNext: () => queue.drain(),
    async runAll(limit = 1_000) {
      let total = 0;
      while (total < limit) {
        const count = await queue.drain();
        if (!count) break;
        total += count;
      }
      return total;
    },
    get: queue.get.bind(queue),
    status: queue.status.bind(queue),
    history: queue.history.bind(queue),
    shutdown: queue.shutdown.bind(queue),
  };
}

export interface SagaStep<T> {
  name: string;
  run(value: T): void | Promise<void>;
  compensate?(value: T, cause: unknown): void | Promise<void>;
}

export function defineSaga<T>(definition: { name: string; steps: readonly SagaStep<T>[] }) {
  return {
    name: definition.name,
    async run(value: T): Promise<void> {
      const completed: SagaStep<T>[] = [];
      try {
        for (const step of definition.steps) {
          await step.run(value);
          completed.push(step);
        }
      } catch (cause) {
        for (const step of completed.reverse()) await step.compensate?.(value, cause);
        throw cause;
      }
    },
  };
}