Build and operate queued work

mkdir -p examples
cp -R node_modules/@db3.ai/app/src/queue/examples/. examples/

npx tsx examples/runQueueWorkflows.ts

import { pathToFileURL } from 'node:url';
import { createGeneratedTestDatabase } from '@db3.ai/app/db/test/db';
import { FailedJob, QueuedJob } from '@db3.ai/app/queue';
import { App } from '@db3.ai/app/server';
import { registerGenerateReportJob } from './createAndProcessReportJob';
import { dispatchReportBatch, dispatchReportPipeline } from './dispatchReportWorkflows';

/**
 * Compares sequential chains with immediately queued batches using real SQL.
 *
 * The example job only logs a report identity. This lab proves orchestration,
 * not a reporting provider. It creates and destroys its own test database.
 *
 * @returns Stable queue counts, validation and processing observations.
 */
export async function runQueueWorkflows() {
	const database = await createGeneratedTestDatabase('queue_workflows');
	const application = new App({ db: database.db, queue: { queue: 'reports', queueMonitor: false } });
	try {
		application.log.level = 'silent';
		await application.db.install(QueuedJob, FailedJob);
		registerGenerateReportJob();
		let invalidRejected = false;
		try { await dispatchReportPipeline([]); } catch (error) { invalidRejected = error instanceof Error && error.message.includes('at least one report'); }
		const afterInvalid = await QueuedJob.query().count();
		await dispatchReportPipeline(['first', 'second']);
		const chainInitial = await QueuedJob.query().count();
		const first = await application.queue.workNextJob('reports');
		const chainRemaining = await QueuedJob.query().count();
		const second = await application.queue.workNextJob('reports');
		const batch = await dispatchReportBatch(['third', 'fourth']);
		const batchInitial = await QueuedJob.query().count();
		const third = await application.queue.workNextJob('reports');
		const fourth = await application.queue.workNextJob('reports');
		return { invalidRejected, afterInvalid, chainInitial, chainRemaining, batchInitial, batchIds: batch.length, statuses: [first?.status, second?.status, third?.status, fourth?.status], remaining: await QueuedJob.query().count() };
	} finally {
		try { await application.close(); } finally { await database.destroy(); }
	}
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
	console.log(JSON.stringify(await runQueueWorkflows(), null, 2));
}

import { expect, it } from 'vitest';
import { runQueueWorkflows } from '../../examples/runQueueWorkflows';

it('rejects an empty chain, then processes sequential and independent jobs on real SQL', async () => {
	expect(await runQueueWorkflows()).toEqual({
		invalidRejected: true, afterInvalid: 0, chainInitial: 1, chainRemaining: 1,
		batchInitial: 2, batchIds: 2, statuses: ['succeeded', 'succeeded', 'succeeded', 'succeeded'], remaining: 0,
	});
});

npx vitest run tests/queue/runQueueWorkflows.test.ts
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --skipLibCheck examples/*.ts

import { app } from '@db3.ai/app/server';
import { QueueableJob } from '@db3.ai/app/queue';

/**
 * Durable data required to generate one application report.
 */
export interface GenerateReportJobData extends Record<string, unknown> {
	/** Stable application-owned report identifier. */
	reportId: string;
}

/**
 * Example application job restored from its JSON-safe queue payload.
 */
export class GenerateReportJob extends QueueableJob<GenerateReportJobData> {
	/**
	 * Creates a report job after validating its durable payload.
	 *
	 * @param data - Report identity persisted with the queued job.
	 */
	constructor(data: GenerateReportJobData) {
		if (typeof data.reportId !== 'string' || data.reportId.trim() === '') {
			throw new Error('GenerateReportJob requires a report id.');
		}

		super(data);
	}

	/**
	 * Generates the report through services resolved from the active application.
	 */
	async handle(): Promise<void> {
		app().log.info({
			reportId: this.data.reportId,
		}, 'Generating report');
	}
}

import { app } from '@db3.ai/app/server';
import type { QueueJobId, QueueProcessResult } from '@db3.ai/app/queue';
import { GenerateReportJob } from './GenerateReportJob';

/**
 * Registers the example job in a worker process.
 */
export function registerGenerateReportJob(): void {
	app().queue.registerJob(GenerateReportJob);
}

/**
 * Dispatches one report job through the active application queue.
 *
 * @param reportId - Stable application-owned report identifier.
 * @returns Driver-owned queued job identifier.
 */
export async function dispatchGenerateReportJob(reportId: string): Promise<QueueJobId> {
	return app().queue.dispatch(new GenerateReportJob({ reportId }));
}

/**
 * Claims and processes the next job from the default queue.
 *
 * @returns Queue result, or null when no job is available.
 */
export async function processNextReportJob(): Promise<QueueProcessResult | null> {
	return app().queue.workNextJob();
}

import { app } from '@db3.ai/app/server';
import type { QueueJobId } from '@db3.ai/app/queue';
import { GenerateReportJob } from './GenerateReportJob';

/**
 * Dispatches report jobs as an ordered pipeline.
 *
 * Each job is persisted only after its predecessor succeeds. A terminal failure
 * therefore prevents the remaining reports from being dispatched.
 *
 * @param reportIds - Report identifiers to process in order.
 * @returns Driver-owned identifier for the first queued job.
 */
export async function dispatchReportPipeline(reportIds: string[]): Promise<QueueJobId> {
	if (reportIds.length === 0) {
		throw new Error('A report pipeline requires at least one report.');
	}

	return app().queue.chain(
		reportIds.map(reportId => new GenerateReportJob({ reportId })),
		{
			queue: 'reports',
		},
	);
}

/**
 * Dispatches report jobs independently so workers can process them in parallel.
 *
 * @param reportIds - Report identifiers that do not depend on one another.
 * @returns Driver-owned identifiers for every queued job.
 */
export async function dispatchReportBatch(reportIds: string[]): Promise<QueueJobId[]> {
	return app().queue.batch(
		reportIds.map(reportId => new GenerateReportJob({ reportId })),
		{
			queue: 'reports',
		},
	);
}

import { app } from '@db3.ai/app/server';
import { QueueRetryLaterError, type QueueJobId } from '@db3.ai/app/queue';
import { GenerateReportJob } from './GenerateReportJob';

/**
 * Dispatches one report with an explicit ordinary-failure retry policy.
 *
 * @param reportId - Stable application-owned report identifier.
 * @returns Driver-owned queued job identifier.
 */
export async function dispatchReportWithRetryPolicy(reportId: string): Promise<QueueJobId> {
	return app().queue.dispatch(new GenerateReportJob({ reportId }), {
		queue: 'reports',
		maxTries: 8,
		backoff: {
			strategy: 'exponential',
			initialSeconds: 15,
			maxSeconds: 900,
			jitter: true,
		},
		retryUntilSeconds: 21_600,
	});
}

/**
 * Defers the current job when a provider supplies an explicit retry delay.
 *
 * Throw this from a QueueableJob handler after detecting provider backpressure.
 * Deferral restores the claimed attempt instead of consuming an ordinary retry.
 *
 * @param delaySeconds - Provider-requested delay before another attempt.
 */
export function retryReportAfterBackpressure(delaySeconds: number): never {
	throw new QueueRetryLaterError(delaySeconds, 'The report provider requested backpressure.');
}

/**
 * Replays one terminal report failure while preserving its audit record.
 *
 * @param failedJobId - Driver-owned identifier from failedJobs().
 * @returns Driver-owned identifier for the replacement queue job.
 */
export async function retryFailedReportJob(failedJobId: QueueJobId): Promise<QueueJobId> {
	return app().queue.retryFailed(failedJobId, {
		queue: 'reports',
	});
}