queues.ts · Source

queues.tsapps/worker/src/queues.tsraw ← src2,858 B← Back to the app
1/**
2 * apps/worker — BullMQ queue names + enqueue helper.
3 *
4 * The seven queues and their owners (see README for the event→queue map):
5 *   files         → file pipeline (hash, mime check, scan, metadata)
6 *   search-index  → explicit (re)index ops; stream events are handled by the
7 *                   search-index StreamConsumer instead
8 *   webhooks      → outbound webhook dispatch (retry-safe via BullMQ attempts)
9 *   automations   → automation run jobs (schedule repeatables land here too)
10 *   email         → outbound mail (dev-safe no-op without SMTP_HOST)
11 *   ai            → embeddings / generation (no-op without AI_API_KEY)
12 *   maintenance   → retention, reconciliation, DLQ reports, session cleanup
13 *
14 * Retry policy (every job): 8 attempts, exponential backoff, base 5s, ×2 per
15 * attempt. BullMQ's exponential backoff has no explicit delay cap; with base
16 * 5s and 8 attempts the per-retry delays are 5s → 10s → 20s → 40s → 80s →
17 * 160s → 320s (≈5 min), which satisfies the "base 5s ×2, cap 5min" policy.
18 * removeOnComplete keeps the last 1000 completed jobs for observability;
19 * failed jobs are kept (removeOnFail false) for the DLQ reports.
20 */
21import { Queue, type Job, type JobsOptions } from "bullmq";
22import { randomUUID } from "node:crypto";
23import type { Redis } from "ioredis";
24 
25export const QUEUES = {
26  files: "files",
27  searchIndex: "search-index",
28  webhooks: "webhooks",
29  automations: "automations",
30  email: "email",
31  ai: "ai",
32  maintenance: "maintenance"
33} as const;
34 
35export type QueueName = (typeof QUEUES)[keyof typeof QUEUES];
36 
37/** Default job options — applied to EVERY enqueue unless overridden. */
38export function defaultJobOpts(partial: JobsOptions = {}): JobsOptions {
39  return {
40    attempts: 8,
41    backoff: { type: "exponential", delay: 5_000 },
42    removeOnComplete: { age: 1_000 },
43    removeOnFail: false,
44    jobId: partial.jobId ?? randomUUID(),
45    ...partial
46  };
47}
48 
49/**
50 * Enqueue a job on a queue. Job id defaults to a fresh UUID (idempotency
51 * hooks may pass an explicit jobId — BullMQ dedups identical jobIds).
52 */
53export async function enqueue(
54  connection: Redis,
55  queue: QueueName,
56  data: Record<string, unknown>,
57  opts: JobsOptions = {}
58): Promise<Job> {
59  const q = new Queue(queue, { connection });
60  try {
61    // BullMQ's add(name, data, opts): the JOB NAME is the queue name (a
62    // plain string) and `data` is the payload.
63    return await q.add(queue, data, defaultJobOpts(opts));
64  } finally {
65    await q.close();
66  }
67}
68 
69/** Enqueue with an explicit stable job id (schedule-triggered runs, …). */
70export async function enqueueWithId(
71  connection: Redis,
72  queue: QueueName,
73  jobId: string,
74  data: Record<string, unknown>,
75  opts: JobsOptions = {}
76): Promise<Job> {
77  return enqueue(connection, queue, data, { ...opts, jobId });
78}
79