73 lines
2.3 KiB
TypeScript
73 lines
2.3 KiB
TypeScript
import { PgBoss } from "pg-boss";
|
|
|
|
export type JobHandler = (data: Record<string, unknown>) => Promise<void>;
|
|
|
|
export interface JobDefinition {
|
|
name: string;
|
|
handler: JobHandler;
|
|
/** Cron expression for a recurring job (UTC). */
|
|
cron?: string;
|
|
}
|
|
|
|
/** Background work. The real one is pg-boss; tests can pass a recording fake. */
|
|
export interface JobQueue {
|
|
start(): Promise<void>;
|
|
stop(): Promise<void>;
|
|
/** Enqueues a job, optionally not before `startAfter` (Unix ms). */
|
|
send(name: string, data: Record<string, unknown>, startAfter?: number): Promise<string | null>;
|
|
}
|
|
|
|
/**
|
|
* pg-boss in the same Postgres database (its own `pgboss` schema). Queues are
|
|
* always created so the app can enqueue; workers and cron schedules run only
|
|
* when `workers` is true (Step 7.1).
|
|
*/
|
|
export function createPgBossQueue(
|
|
databaseUrl: string,
|
|
options: { workers: boolean; jobs: JobDefinition[]; onError?: (error: Error) => void },
|
|
): JobQueue & { boss: PgBoss } {
|
|
const boss = new PgBoss({ connectionString: databaseUrl, schedule: options.workers });
|
|
if (options.onError) boss.on("error", options.onError);
|
|
|
|
return {
|
|
boss,
|
|
async start() {
|
|
await boss.start();
|
|
for (const job of options.jobs) {
|
|
await boss.createQueue(job.name);
|
|
if (!options.workers) continue;
|
|
await boss.work<Record<string, unknown>>(job.name, async (batch) => {
|
|
for (const item of batch) await job.handler(item.data);
|
|
});
|
|
if (job.cron) await boss.schedule(job.name, job.cron);
|
|
}
|
|
},
|
|
async stop() {
|
|
await boss.stop({ graceful: true });
|
|
},
|
|
send(name, data, startAfter) {
|
|
return boss.send(
|
|
name,
|
|
data,
|
|
startAfter === undefined ? {} : { startAfter: new Date(startAfter) },
|
|
);
|
|
},
|
|
};
|
|
}
|
|
|
|
/** A queue that only records what was sent; for tests that call handlers directly. */
|
|
export function createRecordingQueue(): JobQueue & {
|
|
sent: { name: string; data: Record<string, unknown>; startAfter?: number }[];
|
|
} {
|
|
const sent: { name: string; data: Record<string, unknown>; startAfter?: number }[] = [];
|
|
return {
|
|
sent,
|
|
async start() {},
|
|
async stop() {},
|
|
async send(name, data, startAfter) {
|
|
sent.push({ name, data, ...(startAfter === undefined ? {} : { startAfter }) });
|
|
return `recorded-${sent.length}`;
|
|
},
|
|
};
|
|
}
|