index.ts · Source

index.tsapps/worker/src/index.tsraw ← src24,336 B← Back to the app
1/**
2 * apps/worker — main entry (headless: no business HTTP surface).
3 *
4 * The worker listens on exactly ONE port: the HTTP health server
5 * (TASK-0009 item 4, default 4003) — /health + /health/ready. No other
6 * routes are exposed.
7 *
8 * Startup order:
9 *   1. config (zod, dev-safe defaults) + logger
10 *   2. Redis (shared connection: BullMQ + streams + heartbeat)
11 *   3. Prisma connect (surfaces bad DATABASE_URL immediately)
12 *   4. OpenSearch indices (idempotent create)
13 *   5. seven BullMQ Workers (concurrency = WORKER_CONCURRENCY)
14 *   6. four StreamConsumers on factory:events (one consumer group per queue:
15 *      worker-search-index / worker-webhooks / worker-automations /
16 *      worker-email — the TASK-0094 notification→email decision)
17 *   7. repeatable maintenance jobs + schedule-automation reconciliation
18 *      (re-run every 5 min so DB edits to schedules take effect)
19 *   8. heartbeat: SET worker:heartbeat:<WORKER_ID> EX 90 every 30s (deep
20 *      liveness for the api) + HTTP health server on HEALTH_PORT (4003)
21 *
22 * Graceful shutdown (SIGTERM/SIGINT):
23 *   close workers (finish active jobs, 30s cap) → stop consumers →
24 *   disconnect prisma + redis → exit 0.
25 */
26import { setTimeout as delay } from "node:timers/promises";
27import { Queue, Worker } from "bullmq";
28import { initOtel, shutdownOtel, m, renderPrometheus } from "@factory/observability";
29import { loadEnv } from "./config.js";
30import { getLogger } from "./logger.js";
31import { getRedis, closeRedis, createRedis } from "./lib/redis.js";
32import { createHealthServer } from "./health/server.js";
33import { connectPrisma, closePrisma } from "./lib/prisma.js";
34import { S3Client } from "./lib/s3.js";
35import { SearchClient } from "@factory/search";
36import { QUEUES, enqueue } from "./queues.js";
37import { createFilesWorker } from "./processors/files/processor.js";
38import { assertScannerConfigured } from "./processors/files/steps/scan.js";
39import { cleanupTempFiles } from "./processors/files/temp.js";
40import {
41  createSearchIndexWorker,
42  createSearchIndexConsumer,
43  ensureContentIndex,
44  recordSearchDlq
45} from "./processors/search-index/index.js";
46import {
47  createWebhooksWorker,
48  createWebhooksConsumer,
49  handleWebhookFinalFailure
50} from "./processors/webhooks.processor.js";
51import {
52  createAutomationsWorker,
53  createAutomationsConsumer
54} from "./processors/automations.processor.js";
55import { reconcileScheduleAutomations } from "./processors/automations/schedule.js";
56import { runWatchdog } from "./processors/automations/watchdog.js";
57import { createEmailWorker } from "./processors/email.processor.js";
58import { createNotificationEmailConsumer } from "./processors/email-decision.processor.js";
59import { createAiWorker } from "./processors/ai.processor.js";
60import {
61  createMaintenanceWorker,
62  MAINTENANCE_REPEATABLES
63} from "./processors/maintenance.processor.js";
64 
65const log = getLogger("main");
66 
67const SHUTDOWN_CAP_MS = 30_000;
68const HEARTBEAT_INTERVAL_MS = 30_000;
69const HEARTBEAT_TTL_SEC = 90;
70 
71async function main(): Promise<void> {
72  const env = loadEnv();
73  const started = Date.now();
74  // TASK-0135: OTel FIRST (same contract as the API — the lazy metric
75  // instruments bind to the SDK provider at first touch; disabled → no-op).
76  // OTEL_* come straight from process.env (NOT in config.ts's zod schema —
77  // keeps env.ts/config.ts untouched per the coordinator boundary).
78  const otelEndpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
79  await initOtel({
80    serviceName: "factory-worker",
81    environment: env.NODE_ENV,
82    ...(process.env.OTEL_ENABLED === "true" && otelEndpoint ? { endpoint: otelEndpoint } : {})
83  });
84  log.info(
85    {
86      workerId: env.WORKER_ID,
87      concurrency: env.WORKER_CONCURRENCY,
88      env: env.NODE_ENV,
89      smtp: env.SMTP_HOST ? "configured" : "not configured (email no-op)",
90      clamav: env.CLAMD_SOCKET
91        ? `unix ${env.CLAMD_SOCKET}`
92        : env.CLAMAV_HOST
93          ? `tcp ${env.CLAMAV_HOST}:${env.CLAMAV_PORT}`
94          : "not configured (scans skipped, dev mode)",
95      ai: env.AI_API_KEY ? "configured" : "not configured (ai no-op)",
96      search: env.SEARCH_URL
97    },
98    "factory worker starting"
99  );
100 
101  // --- infrastructure ------------------------------------------------------
102  const redis = getRedis(env.REDIS_URL);
103  const prisma = await connectPrisma();
104  const s3 = new S3Client({
105    endpoint: env.S3_ENDPOINT,
106    region: env.S3_REGION,
107    // TASK-0103 fix (mirrors apps/api/src/modules/files/s3.ts): the files
108    // bucket is `S3_BUCKET_FILES` (with `S3_BUCKET` kept as a legacy alias
109    // for older .env files). Previously this read the raw legacy `S3_BUCKET`,
110    // so a dev .env carrying the stale pre-Phase-9 `S3_BUCKET=factory-uploads`
111    // sent the pipeline's downloadTo/putObject to a bucket that no longer
112    // exists (NoSuchKey 404) even though the API presigned into
113    // `factory-files`. See docs/tasks/phase-09-files.md (TASK-0103 notes).
114    bucket: env.S3_BUCKET_FILES || env.S3_BUCKET,
115    accessKey: env.S3_ACCESS_KEY,
116    secretKey: env.S3_SECRET_KEY,
117    publicBase: env.S3_PUBLIC_BASE
118  });
119  // TASK-0102: the thumbs bucket client (same creds, `factory-thumbs`). The
120  // files pipeline writes generated thumbnails here.
121  const s3Thumbs = new S3Client({
122    endpoint: env.S3_ENDPOINT,
123    region: env.S3_REGION,
124    bucket: env.S3_BUCKET_THUMBS,
125    accessKey: env.S3_ACCESS_KEY,
126    secretKey: env.S3_SECRET_KEY
127  });
128  // TASK-0102 §3: fail closed — production refuses to boot without a ClamAV
129  // scanner (a silently-unscanned prod file pipeline is a security hole).
130  assertScannerConfigured(env);
131  // The single OpenSearch writer (TASK-0096): the @factory/search client over
132  // the `factory-content` alias. Basic auth when the password is set
133  // (OpenSearch security on); dev/CI run security-off.
134  const search = new SearchClient({
135    baseUrl: env.SEARCH_URL,
136    ...(env.SEARCH_PASSWORD ? { auth: { username: env.SEARCH_USERNAME, password: env.SEARCH_PASSWORD } } : {})
137  });
138  const searchOps = {
139    search,
140    prisma,
141    connection: redis,
142    env,
143    // TASK-0125 — incremental RAG embedding: after a page/record is
144    // (re)indexed, its vectors are refreshed on the `ai` queue (the embed
145    // job re-derives the chunk set from the source). Fire-and-forget by the
146    // hook's own try/catch (indexDocs never fails on an enqueue error).
147    enqueueEmbed: (job: { workspaceId: string; docType: "page" | "record"; refId: string }) =>
148      enqueue(redis, QUEUES.ai, { type: "embed:chunks", workspaceId: job.workspaceId, docType: job.docType, refId: job.refId } as Record<string, unknown>)
149  };
150  await ensureContentIndex(searchOps).catch((err) => {
151    log.warn(
152      { err: (err as Error).message },
153      "OpenSearch unavailable at startup — index will be created lazily by the first (re)index; continuing"
154    );
155  });
156 
157  // --- workers -------------------------------------------------------------
158  const workers: Worker[] = [];
159  const push = (w: Worker): void => {
160    w.on("failed", (job, err) => {
161      log.error(
162        { queue: w.name, jobId: job?.id, err: err.message },
163        "job failed (BullMQ will retry per job options)"
164      );
165    });
166    workers.push(w);
167  };
168 
169  push(
170    createFilesWorker({
171      connection: redis,
172      prisma,
173      env,
174      s3,
175      s3Thumbs,
176      // TASK-0102 §7: the files pipeline is IO-bound (scan/render) but uses
177      // disk + CPU (sharp/ffmpeg) — its own lower concurrency, not the
178      // general WORKER_CONCURRENCY.
179      concurrency: env.FILES_WORKER_CONCURRENCY
180    })
181  );
182  const searchIndexWorker = createSearchIndexWorker({
183    connection: redis,
184    ops: searchOps,
185    concurrency: env.WORKER_CONCURRENCY
186  });
187  searchIndexWorker.on("failed", (job, err) => {
188    if (job && job.attemptsMade >= 7) {
189      // 8th attempt exhausted → dead-letter record (TASK-0096 §8).
190      void recordSearchDlq(redis, job, err);
191    }
192  });
193  push(searchIndexWorker);
194  const webhooksWorker = createWebhooksWorker({
195    connection: redis,
196    prisma,
197    env,
198    concurrency: env.WORKER_CONCURRENCY
199  });
200  webhooksWorker.on("failed", (job, err) => {
201    // TASK-0130: the worker owns retry scheduling (attempts:1 on the queue —
202    // every FAILED attempt re-enqueues itself as a delayed job), so a
203    // BullMQ "failed" here means an UNEXPECTED job crash: dead-letter the
204    // delivery row + notify admins (handleWebhookFinalFailure).
205    if (job) void handleWebhookFinalFailure(prisma, redis, env)(job, err);
206  });
207  push(webhooksWorker);
208  // TASK-0113: the pipeline is the retry manager (attempts:1 on the queue) —
209  // the old BullMQ `failed` hook (attemptsMade >= 7) is gone; every failure
210  // is handled inside handleRunAutomation (FAILED / RETRYING / DEAD_LETTER).
211  const automationsWorker = createAutomationsWorker({
212    connection: redis,
213    prisma,
214    env,
215    concurrency: env.AUTOMATIONS_WORKER_CONCURRENCY
216  });
217  push(automationsWorker);
218  push(
219    createEmailWorker({
220      connection: redis,
221      prisma,
222      env,
223      concurrency: env.WORKER_CONCURRENCY
224    })
225  );
226  push(
227    createAiWorker({
228      connection: redis,
229      prisma,
230      env,
231      search,
232      concurrency: env.WORKER_CONCURRENCY
233    })
234  );
235  const maintenanceWorker = createMaintenanceWorker({
236    connection: redis,
237    prisma,
238    env,
239    concurrency: Math.max(1, Math.min(2, env.WORKER_CONCURRENCY))
240  });
241  push(maintenanceWorker);
242  log.info({ count: workers.length }, "bullmq workers created");
243 
244  // TASK-0135 — queue_depth{queue} gauge: the spec's "queue-depth per queue".
245  // One short-lived Queue handle per queue every 15 s (waiting + delayed =
246  // the backlog the Workers/Queues dashboard shows + the >10k alert tests).
247  // Fail-soft: a transient Redis blip logs at debug level, never crashes the
248  // worker. m.queue.setDepth is a no-op when OTel is disabled.
249  const queueDepthTick = async (): Promise<void> => {
250    for (const name of Object.values(QUEUES)) {
251      const q = new Queue(name, { connection: redis });
252      try {
253        const c = await q.getJobCounts("waiting", "delayed");
254        m.queue.setDepth(name, (c.waiting ?? 0) + (c.delayed ?? 0));
255      } catch (err) {
256        log.debug({ queue: name, err: (err as Error).message }, "queue depth probe failed");
257      } finally {
258        await q.close();
259      }
260    }
261  };
262  void queueDepthTick();
263  const queueDepthTimer = setInterval(() => void queueDepthTick(), 15_000);
264  queueDepthTimer.unref();
265 
266  // --- stream consumers (one group per queue) -------------------------------
267  // Each consumer gets its OWN ioredis connection. A StreamConsumer's read
268  // loop issues XREADGROUP ... BLOCK 5000 back-to-back, and an ioredis client
269  // is a single ordered-response socket: if two or more consumers share a
270  // connection, their blocking reads serialize on that one socket and every
271  // non-blocking command queued behind them (heartbeat SET, BullMQ lock-renew
272  // PEXPIRE, boot queue ops, the search-index:lag probe) is starved. With the
273  // four TASK-0094/0096/0100 consumers on the shared client, boot deadlocked
274  // before the health server could start (TASK-0100 S5 diagnosis). BullMQ's
275  // workers already run on dedicated named clients for the same reason.
276  const searchStreamRedis = createRedis(env.REDIS_URL);
277  const webhooksStreamRedis = createRedis(env.REDIS_URL);
278  const automationsStreamRedis = createRedis(env.REDIS_URL);
279  const emailStreamRedis = createRedis(env.REDIS_URL);
280  const consumers = [
281    createSearchIndexConsumer({
282      redis: searchStreamRedis,
283      ops: searchOps,
284      consumerId: `${env.WORKER_ID}-search`
285    }),
286    createWebhooksConsumer({
287      redis: webhooksStreamRedis,
288      prisma,
289      consumerId: `${env.WORKER_ID}-webhooks`
290    }),
291    createAutomationsConsumer({
292      redis: automationsStreamRedis,
293      prisma,
294      consumerId: `${env.WORKER_ID}-automations`
295    }),
296    // TASK-0094 R6: the notification→email decision (its own group, so email
297    // lag can't block the other consumers). The hourly digest is a repeatable
298    // maintenance job (registered above via MAINTENANCE_REPEATABLES). `redis`
299    // is the stream read connection (dedicated); `connection` stays the shared
300    // client (only for occasional non-blocking enqueue).
301    createNotificationEmailConsumer({
302      redis: emailStreamRedis,
303      connection: redis,
304      prisma,
305      env,
306      consumerId: `${env.WORKER_ID}-email`
307    })
308  ];
309  for (const c of consumers) await c.start();
310  log.info({ count: consumers.length }, "stream consumers started");
311 
312  // --- repeatable jobs -------------------------------------------------------
313  const mq = new Queue(QUEUES.maintenance, { connection: redis });
314  for (const r of MAINTENANCE_REPEATABLES) {
315    await mq.add(r.id, r.data as Record<string, unknown>, {
316      repeat: { pattern: r.pattern },
317      jobId: r.id,
318      removeOnComplete: true,
319      removeOnFail: false
320    });
321  }
322  await mq.close();
323  log.info({ jobs: MAINTENANCE_REPEATABLES.map((r) => r.id) }, "repeatable maintenance jobs registered");
324 
325  // TASK-0121 — nightly AI usage rollup (the ai queue owns it: it reconciles
326  // the per-workspace month-to-date token totals for dashboards and sweeps
327  // for workspaces that crossed their budget but have since gone quiet — the
328  // API's submit-time guard only notifies on the NEXT call). Registered on a
329  // SEPARATE repeatable queue instance (BullMQ scopes a repeatable's key to
330  // the queue it was added on), offset from the 03:00 maintenance window.
331  const aiQ = new Queue(QUEUES.ai, { connection: redis });
332  await aiQ.add("ai", { type: "usage.rollup" }, {
333    repeat: { pattern: "15 2 * * *" },
334    jobId: "ai:usage-rollup",
335    removeOnComplete: true,
336    removeOnFail: false
337  });
338  await aiQ.close();
339  log.info({ job: "ai:usage-rollup" }, "repeatable ai usage rollup registered");
340 
341  // TASK-0125 — nightly FULL re-embed (03:30 UTC, offset from the 02:15
342  // usage rollup): re-embeds every ai_embeddings row, verifies count parity
343  // (drift > 2% → ai.reembed_required) and is the model-change re-index rule
344  // (a new embedding model simply re-embeds everything overnight). Same
345  // separate-instance pattern as the usage rollup.
346  const aiReembedQ = new Queue(QUEUES.ai, { connection: redis });
347  await aiReembedQ.add("ai", { type: "embed:reembed" }, {
348    repeat: { pattern: "30 3 * * *" },
349    jobId: "ai:reembed",
350    removeOnComplete: true,
351    removeOnFail: false
352  });
353  await aiReembedQ.close();
354  log.info({ job: "ai:reembed" }, "repeatable ai nightly re-embed registered");
355 
356  // Schedule automations: register now + reconcile every 5 minutes.
357  const scheduleReconcile = async (): Promise<void> => {
358    try {
359      await reconcileScheduleAutomations(redis, prisma);
360    } catch (err) {
361      log.warn({ err: (err as Error).message }, "schedule reconcile failed");
362    }
363  };
364  await scheduleReconcile();
365  const scheduleTimer = setInterval(scheduleReconcile, 5 * 60_000);
366  scheduleTimer.unref();
367 
368  // TASK-0113 §9: watchdog — reclaim stale runs (RUNNING > 15 min or
369  // PENDING > 15 min) every 5 min. The in-process interval mirrors the
370  // schedule reconcile (a BullMQ repeatable would just re-enter the same
371  // pool; an interval keeps the watchdog off the queue and trivially
372  // observable).
373  const watchdogTick = async (): Promise<void> => {
374    try {
375      await runWatchdog(prisma, redis, env);
376    } catch (err) {
377      log.warn({ err: (err as Error).message }, "watchdog pass failed");
378    }
379  };
380  const watchdogTimer = setInterval(watchdogTick, 5 * 60_000);
381  watchdogTimer.unref();
382 
383  // TASK-0102 §3/§7: temp-file cleanup cron (every 5 min). The processor
384  // deletes its own temp file in a `finally`; this reaps the crash
385  // leftovers (a SIGKILL mid-pipeline) older than FILES_TEMP_MAX_AGE_MS so
386  // an object "never stays on disk > scan duration + 5 min".
387  const tempCleanup = async (): Promise<void> => {
388    try {
389      await cleanupTempFiles(env);
390    } catch (err) {
391      log.warn({ err: (err as Error).message }, "temp cleanup cron failed");
392    }
393  };
394  await tempCleanup();
395  const tempTimer = setInterval(tempCleanup, 5 * 60_000);
396  tempTimer.unref();
397 
398  // --- heartbeat (headless liveness for api /health/deep) --------------------
399  const heartbeat = async (): Promise<void> => {
400    try {
401      await redis.set(
402        `worker:heartbeat:${env.WORKER_ID}`,
403        JSON.stringify({ at: new Date().toISOString(), uptimeMs: Date.now() - started }),
404        "EX",
405        HEARTBEAT_TTL_SEC
406      );
407    } catch (err) {
408      log.warn({ err: (err as Error).message }, "heartbeat write failed");
409    }
410  };
411  await heartbeat();
412  const heartbeatTimer = setInterval(heartbeat, HEARTBEAT_INTERVAL_MS);
413  heartbeatTimer.unref();
414 
415  // --- HTTP health (TASK-0009 item 4): /health + /health/ready on 4003 --------
416  // The worker is otherwise headless; this is its only port. BullMQ rides on the
417  // same Redis backend, so the queue check is an independent Redis PING.
418  //
419  // TASK-0100 req 1 (Backend Changes): `search-index:lag` — the consumer-lag
420  // metric the lag E2E polls while asserting the worker consumes a 1k-event
421  // burst without backlog growth > 30 s.
422  //   pending    = XPENDING factory:events worker-search-index (delivered,
423  //                not yet ACKed — in-flight or awaiting XAUTOCLAIM redelivery)
424  //   lagEntries = entries in the stream NOT yet delivered to the group:
425  //                the exclusive tail after the group's last-delivered-id
426  //                (XINFO GROUPS). backlog = pending + lagEntries.
427  // Every Redis read fails soft → the endpoint reports available:false
428  // (observed as zeroed counts) instead of 5xx; the probe never gates
429  // readiness and never throws out of boot.
430  // Shared probe: the /health/search-index:lag endpoint AND the 30 s OTel
431  // gauge tick both read the same stream-group state (TASK-0135 added the
432  // seconds dimension — the SearchIndexLagHigh alert needs an elapsed-time
433  // gauge, not just the entry counts).
434  const probeSearchIndexLag = async (): Promise<{
435    pending: number;
436    lagEntries: number;
437    lagSeconds: number;
438  }> => {
439    const STREAM = "factory:events";
440    const GROUP = "worker-search-index";
441    const [pendingInfo, groups] = await Promise.all([
442      (redis.xpending(STREAM, GROUP) as unknown as Promise<{ count: number }>).catch(() => ({ count: 0 })),
443      (redis.xinfo("GROUPS", STREAM) as Promise<{ name: string; "last-delivered-id": string }[]>)
444        .catch(() => [] as { name: string; "last-delivered-id": string }[])
445    ]);
446    const pending = pendingInfo?.count ?? 0;
447    let lagEntries = 0;
448    const groupInfo = groups.find((g) => g.name === GROUP);
449    let tailId: string | null = null;
450    if (groupInfo) {
451      // Entries with ids strictly greater than last-delivered-id. Redis
452      // exclusive id ranges (Redis ≥ 6.2; dev/CI pin redis:7): prefix the
453      // exact id with "<" to exclude it.
454      const lastId = groupInfo["last-delivered-id"] ?? "0-0";
455      const tail = await redis
456        .xrange(STREAM, `<${lastId}`, "+", "COUNT", 100_000)
457        .catch((): string[][] => []);
458      lagEntries = tail.length;
459      // Oldest undelivered entry = first in the ascending tail (its stream
460      // id embeds the creation milliseconds).
461      tailId = tail[0]?.[0] ?? null;
462    }
463    // Lag in seconds = max(elapsed since the oldest UNDELIVERED event,
464    // elapsed since the oldest PENDING (delivered, un-ACKed) event). Each
465    // leg fails soft (a probe blip reports the other leg / 0, never throws).
466    const nowMs = Date.now();
467    let lagSeconds = 0;
468    if (tailId) {
469      const tailMs = Number(tailId.split("-")[0]);
470      if (Number.isFinite(tailMs) && tailMs > 0) lagSeconds = Math.max(0, (nowMs - tailMs) / 1000);
471    }
472    try {
473      const entries = await redis.xpending(STREAM, GROUP, "-", "+", 1);
474      const first = (Array.isArray(entries) ? entries[0] : undefined) as
475        | { "time-delivered-in-ms"?: number }
476        | undefined;
477      const deliveredMs = first?.["time-delivered-in-ms"];
478      if (typeof deliveredMs === "number" && deliveredMs > 0) {
479        lagSeconds = Math.max(lagSeconds, (nowMs - deliveredMs) / 1000);
480      }
481    } catch {
482      // xpending with range unsupported / stream gone — the tail leg above
483      // still covers undelivered lag.
484    }
485    return { pending, lagEntries, lagSeconds };
486  };
487 
488  const searchIndexLag = async (): Promise<{ pending: number; lagEntries: number }> => {
489    const r = await probeSearchIndexLag();
490    return { pending: r.pending, lagEntries: r.lagEntries };
491  };
492 
493  const healthServer = createHealthServer({
494    pingRedis: async () => (await redis.ping()) === "PONG",
495    checkQueue: async () => (await redis.ping()) === "PONG",
496    checkSearchIndexLag: searchIndexLag,
497    // TASK-0135: OTel Prometheus exposition on the worker's ONLY port (the
498    // scrape config targets :4003/metrics — same pattern as the API).
499    metricsBody: renderPrometheus
500  });
501  await new Promise<void>((resolve, reject) => {
502    healthServer.once("error", reject);
503    healthServer.listen(env.HEALTH_PORT, () => {
504      healthServer.removeListener("error", reject);
505      resolve();
506    });
507  });
508  log.info({ port: env.HEALTH_PORT }, "worker health server listening");
509 
510  log.info({ ms: Date.now() - started }, "factory worker ready");
511 
512  // TASK-0135 — search_index_lag_seconds gauge: same probe as the
513  // /health/search-index:lag endpoint, on its own 30 s cadence (the
514  // SearchIndexLagHigh alert reads this; the 30 s window matches the
515  // alert's own "for" period). No-op when OTel is disabled.
516  const searchLagTick = async (): Promise<void> => {
517    try {
518      const { lagSeconds } = await probeSearchIndexLag();
519      m.search.lag(lagSeconds);
520    } catch (err) {
521      log.debug({ err: (err as Error).message }, "search-index lag probe failed");
522    }
523  };
524  void searchLagTick();
525  const searchLagTimer = setInterval(() => void searchLagTick(), 30_000);
526  searchLagTimer.unref();
527 
528  // --- graceful shutdown -----------------------------------------------------
529  let shuttingDown = false;
530  const shutdown = async (signal: string): Promise<void> => {
531    if (shuttingDown) return;
532    shuttingDown = true;
533    log.info({ signal }, "shutting down (30s cap)");
534    const cap = delay(SHUTDOWN_CAP_MS).then(() => {
535      log.error("shutdown cap reached — forcing exit");
536      process.exit(1);
537    });
538    try {
539      clearInterval(heartbeatTimer);
540      clearInterval(scheduleTimer);
541      clearInterval(tempTimer);
542      clearInterval(queueDepthTimer);
543      clearInterval(searchLagTimer);
544      // Health server first (no in-flight jobs; just stop accepting).
545      await new Promise<void>((resolve) => {
546        healthServer.close(() => resolve());
547        healthServer.closeAllConnections?.();
548      });
549      // Workers first (finish active jobs; BullMQ closes connections after).
550      await Promise.allSettled(workers.map((w) => w.close()));
551      for (const c of consumers) {
552        try {
553          await c.stop();
554        } catch (err) {
555          log.warn({ err: (err as Error).message }, "consumer stop error");
556        }
557      }
558      // Close the dedicated per-consumer stream connections (each runs an
559      // XREADGROUP BLOCK read loop; the shared client is closed below).
560      await Promise.allSettled(
561        [searchStreamRedis, webhooksStreamRedis, automationsStreamRedis, emailStreamRedis].map(
562          (r) => r.quit()
563        )
564      );
565      // TASK-0135: flush the OTel SDK (pending spans + the final metric
566      // collection) before the process dies.
567      await shutdownOtel();
568      await closePrisma();
569      await closeRedis();
570      log.info("shutdown complete");
571      process.exit(0);
572    } catch (err) {
573      log.error({ err: (err as Error).message }, "error during shutdown");
574      process.exit(1);
575    }
576  };
577  process.on("SIGTERM", () => void shutdown("SIGTERM"));
578  process.on("SIGINT", () => void shutdown("SIGINT"));
579}
580 
581main().catch((err) => {
582  // Boot failure: log loud + non-zero exit (orchestrator restarts).
583  // eslint-disable-next-line no-console
584  console.error("factory worker failed to start:", err);
585  process.exit(1);
586});
587