`packages/shared/src/queue.ts`

43 lines
/**
 * Thin queue facade over pg-boss so a later broker swap (BullMQ etc.)
 * touches this file only. Pipeline stages are pure functions — the queue
 * merely transports work; tests and the CLI invoke stages directly.
 */
import PgBoss from "pg-boss";

let boss: PgBoss | null = null;

export async function getQueue(): Promise<PgBoss> {
  if (!boss) {
    const url = process.env.DATABASE_URL;
    if (!url) throw new Error("DATABASE_URL is required for the job queue");
    boss = new PgBoss(url);
    await boss.start();
  }
  return boss;
}

export async function enqueue<T extends object>(jobName: string, payload: T): Promise<string | null> {
  const queue = await getQueue();
  await queue.createQueue(jobName);
  return queue.send(jobName, payload);
}

export async function work<T extends object>(
  jobName: string,
  handler: (payload: T) => Promise<void>,
): Promise<void> {
  const queue = await getQueue();
  await queue.createQueue(jobName);
  await queue.work<T>(jobName, async (jobs) => {
    for (const job of jobs) await handler(job.data);
  });
}

export async function stopQueue(): Promise<void> {
  if (boss) {
    await boss.stop();
    boss = null;
  }
}