import type { AddOptions, Job } from "./index.ts"; import type { DurableQueue } from "./durable.ts"; export interface ScheduledJob { name: string; data: T; everyMs: number; options?: AddOptions; } export interface QueueScheduler { start(): void; stop(): void; tick(now?: number): Promise; snapshot(): { running: boolean; schedules: number; nextRuns: Record }; } /** Restart-safe scheduler when used with a durable queue and stable idempotency buckets. */ export function createQueueScheduler( queue: DurableQueue, schedules: ScheduledJob[], options: { pollMs?: number; now?: () => number } = {}, ): QueueScheduler { const now = options.now ?? Date.now; const pollMs = options.pollMs ?? 1000; const nextRuns = new Map(schedules.map((schedule) => [schedule.name, now()])); let timer: ReturnType | null = null; const tick = async (at = now()) => { let added = 0; for (const schedule of schedules) { if (!Number.isFinite(schedule.everyMs) || schedule.everyMs < 1) throw new RangeError(`Schedule '${schedule.name}' everyMs must be positive`); const next = nextRuns.get(schedule.name) ?? at; if (next > at) continue; const bucket = Math.floor(at / schedule.everyMs); await queue.add(schedule.name, schedule.data, { ...schedule.options, idempotencyKey: schedule.options?.idempotencyKey ?? `schedule:${schedule.name}:${bucket}`, }); nextRuns.set(schedule.name, (bucket + 1) * schedule.everyMs); added++; } return added; }; return { start() { if (!timer) timer = setInterval(() => void tick(), pollMs); }, stop() { if (timer) clearInterval(timer); timer = null; }, tick, snapshot: () => ({ running: timer !== null, schedules: schedules.length, nextRuns: Object.fromEntries(nextRuns), }), }; } export async function addBatch( queue: DurableQueue, name: string, values: T[], options?: AddOptions, ): Promise[]> { return Promise.all( values.map((value, index) => queue.add(name, value, { ...options, idempotencyKey: options?.idempotencyKey ? `${options.idempotencyKey}:${index}` : undefined, }), ), ); } export interface QueueDashboardSnapshot { generatedAt: number; pending: number; failed: number; byName: Record; oldestRunAt?: number; } export async function queueDashboardSnapshot(queue: DurableQueue): Promise { const jobs = await queue.list(); return { generatedAt: Date.now(), pending: jobs.length, failed: queue.failed().length, byName: jobs.reduce>((counts, job) => { counts[job.name] = (counts[job.name] ?? 0) + 1; return counts; }, {}), oldestRunAt: jobs.length ? Math.min(...jobs.map((job) => job.runAt)) : undefined, }; } export function renderQueueDashboard(snapshot: QueueDashboardSnapshot): string { const rows = Object.entries(snapshot.byName) .sort(([left], [right]) => left.localeCompare(right)) .map( ([name, count]) => `${name.replace(/[&<>"']/g, (c) => ({ "&": "&", "<": "<", ">": ">", '"': """, "'": "'" })[c]!)}${count}`, ) .join(""); return `WRNexus Queue Dashboard

Queue dashboard

Pending: ${snapshot.pending} · Failed: ${snapshot.failed}

${rows}

Queue	Pending

`; } /** Long-running scheduler/worker loop suitable for a dedicated process or container. */ export async function runQueueDaemon( queue: DurableQueue, scheduler: QueueScheduler, options: { signal?: AbortSignal; pollMs?: number; onError?: (error: unknown) => void } = {}, ): Promise { const pollMs = options.pollMs ?? 250; if (!Number.isInteger(pollMs) || pollMs < 10) throw new RangeError("Daemon pollMs must be at least 10ms"); scheduler.start(); try { while (!options.signal?.aborted) { try { await scheduler.tick(); await queue.drain(); } catch (error) { options.onError?.(error); } await new Promise((resolve) => { const timer = setTimeout(resolve, pollMs); options.signal?.addEventListener( "abort", () => { clearTimeout(timer); resolve(); }, { once: true }, ); }); } } finally { scheduler.stop(); await queue.shutdown(); } }