automations.processor.ts

automations.processor.tsapps/worker/src/processors/automations.processor.tsraw ← processors6,126 B← Back to the app
1/**
2 * apps/worker — automation worker + trigger matcher consumer (TASK-0113).
3 *
4 * Two BullMQ/stream surfaces, both thin over the `automations/` package:
5 *
6 *  1. `createAutomationsWorker` — consumes the `automations` queue.
7 *     Two job kinds share the queue:
8 *      - `run-automation` jobs (`{ runId, automationId, workspaceId,
9 *        triggerEvent, eventId }`) → `handleRunAutomation` (the TASK-0113
10 *        pipeline: PENDING → RUNNING → terminal);
11 *      - schedule repeatable fires (job NAME `automations:run:{automationId}`,
12 *        data `{ automationId, event: "schedule", cron }`) →
13 *        `executeScheduleFire` (creates the PENDING run + enqueues a
14 *        `run-automation` job).
15 *     NO BullMQ attempts on this queue (`attempts: 1`): the pipeline is the
16 *     retry manager (worker-side retryCount + backoff, §7). BullMQ's
17 *     `lockDuration` 60 s / `stalledInterval` 30 s still reprocess in-process
18 *     stalls (§9); the watchdog is the cross-restart backstop.
19 *
20 *  2. `createAutomationsConsumer` — a StreamConsumer on `factory:events`,
21 *     group **`automations-matcher`** (renamed from the old
22 *     `worker-automations` — spec §5; a first deploy with the old group
23 *     simply gets a second group reading from "0", which the dedupe key
24 *     makes idempotent). Maps stream events → trigger types
25 *     (`record.updated` fans out to RECORD_UPDATED + PROPERTY_CHANGED) and
26 *     delegates to `matchAndEnqueue` (scope match + dedupe + PENDING run +
27 *     enqueue).
28 *
29 * The old processor's `handleAutomationFinalFailure` (BullMQ `failed` hook)
30 * is GONE: with attempts:1 the pipeline handles every failure itself
31 * (FAILED / RETRYING / DEAD_LETTER are the handler's return values, not
32 * BullMQ retries). index.ts drops the `.on("failed")` attachment.
33 */
34import type { PrismaClient } from "@prisma/client";
35import type { Redis } from "ioredis";
36import { Worker, type Job } from "bullmq";
37import { withJobOtel } from "../lib/job-otel.js";
38import { getLogger } from "../logger.js";
39import type { WorkerEnv } from "../config.js";
40import { QUEUES } from "../queues.js";
41import { StreamConsumer } from "../lib/redis-stream.js";
42import { tryParseEvent } from "../events.js";
43import { handleRunAutomation, runAutomationJobSchema } from "./automations/pipeline.js";
44import { matchAndEnqueue } from "./automations/matcher.js";
45import { executeScheduleFire, isScheduleFireJobName } from "./automations/schedule.js";
46 
47const log = getLogger("automations");
48 
49/**
50 * Create the `automations` BullMQ worker. `concurrency` comes from
51 * `AUTOMATIONS_WORKER_CONCURRENCY` (default 4 — spec §1; schedule fires and
52 * run jobs share the pool).
53 */
54export function createAutomationsWorker(opts: {
55  connection: Redis;
56  prisma: PrismaClient;
57  env: WorkerEnv;
58  concurrency?: number;
59}): Worker {
60  const { connection, prisma, env } = opts;
61  return new Worker(
62    QUEUES.automations,
63    withJobOtel(QUEUES.automations, async (job: Job) => {
64      // 1) Schedule repeatable fire (job NAME carries the identity).
65      if (isScheduleFireJobName(job.name)) {
66        return executeScheduleFire(connection, prisma, env, (job.data ?? {}) as {
67          automationId?: string;
68          cron?: string;
69          timeZone?: string;
70        });
71      }
72      // 2) run-automation job.
73      const data = runAutomationJobSchema.safeParse(job.data);
74      if (!data.success) {
75        log.error({ jobId: job.id, err: data.error.message }, "malformed run-automation job (dropped)");
76        return { status: "FAILED" };
77      }
78      const status = await handleRunAutomation({ prisma, redis: connection, env }, { data: job.data });
79      log.info({ runId: data.data.runId, jobId: job.id, status }, "run-automation job settled");
80      return { status };
81    }),
82    {
83      connection,
84      concurrency: opts.concurrency ?? env.AUTOMATIONS_WORKER_CONCURRENCY,
85      // §9: the in-process stall backstop (a wedged worker is reprocessed by
86      // a peer after 60 s; the watchdog covers the cross-restart case).
87      lockDuration: 60_000,
88      stalledInterval: 30_000
89    }
90  );
91}
92 
93/**
94 * Create the trigger-matcher stream consumer (group `automations-matcher`).
95 * Every event the engine can react to is handled here; unknown events are
96 * ACKed as not-mine (StreamConsumer semantics: a resolved handler = ACK).
97 *
98 * `record.updated` fans out to BOTH RECORD_UPDATED and PROPERTY_CHANGED
99 * trigger types (PROPERTY_CHANGED is not in EVENT_TO_TRIGGER — it is a
100 * property-scoped view of the same event).
101 */
102export function createAutomationsConsumer(opts: {
103  redis: Redis;
104  prisma: PrismaClient;
105  consumerId: string;
106}): StreamConsumer {
107  const { redis, prisma, consumerId } = opts;
108  return new StreamConsumer(
109    redis,
110    "factory:events",
111    "automations-matcher",
112    consumerId,
113    async (entry) => {
114      const parsed = tryParseEvent(entry);
115      if (!parsed) return; // poison entry — StreamConsumer logs + drops
116      const { envelope } = parsed;
117      const triggerTypes = triggerTypesFor(envelope.event);
118      if (triggerTypes.length === 0) return; // not mine — ACK
119      const payload = (envelope.payload ?? {}) as Record<string, unknown>;
120      for (const tt of triggerTypes) {
121        try {
122          await matchAndEnqueue({ prisma, redis }, tt, { id: parsed.id, event: envelope.event, payload });
123        } catch (err) {
124          // A matcher failure for one trigger type must not wedge the group:
125          // log + ACK (dedupe keys mean a manual re-drive is safe).
126          log.error({ err: (err as Error).message, triggerType: tt, entryId: parsed.id }, "matcher pass failed");
127        }
128      }
129    }
130  );
131}
132 
133/** Stream event → the trigger types it can satisfy. */
134function triggerTypesFor(eventName: string): string[] {
135  switch (eventName) {
136    case "record.created":
137      return ["RECORD_CREATED"];
138    case "record.updated":
139      return ["RECORD_UPDATED", "PROPERTY_CHANGED"];
140    case "page.created":
141      return ["PAGE_CREATED"];
142    case "form.submitted":
143      return ["FORM_SUBMITTED"];
144    default:
145      return [];
146  }
147}
148