pipeline.ts · Source

pipeline.tsapps/worker/src/processors/automations/pipeline.tsraw ← automations22,662 B← Back to the app
1/**
2 * apps/worker — automation run pipeline (TASK-0113).
3 *
4 * Executes one `run-automation` job: PENDING → RUNNING →
5 * SUCCEEDED/FAILED (+ retry / DEAD_LETTER). This is the single place where
6 * automation side effects are observed — every run, including condition
7 * skips and loop-guard skips, is recorded (exit criterion: "every run
8 * logged").
9 *
10 * Per-run flow:
11 *  1. load the run row (idempotent re-read — retries carry the same runId)
12 *     + automation + actions (position order) + condition;
13 *  2. PENDING → RUNNING (startedAt = now) — transition asserted; a
14 *     SUCCEEDED/DEAD_LETTER row (duplicate job delivery) is dropped; a stale
15 *     RUNNING row (crash redelivery) is tolerated;
16 *  3. loop guard (shared `loopGuard`): self-trigger / depth ≥ 5 → SUCCEEDED
17 *     with `output = { skipped: "self_trigger" | "depth_limit" }`, logged,
18 *     and `automation.executed` emitted (depth 0 only);
19 *  4. 50-action guard (shared `AUTOMATION_MAX_ACTIONS`) → FAILED
20 *     RUN_ACTION_LIMIT (a corrupt row must not run 51 actions);
21 *  5. condition gate (TASK-0111 `evaluateCondition` over a context built
22 *     from the event + record): a condition-skip → SUCCEEDED
23 *     `{ skipped: "conditions_not_met" }`; a corrupt/over-complex tree →
24 *     FAILED CONDITION_INVALID; an over-budget eval → FAILED
25 *     CONDITION_TIMEOUT;
26 *  6. for each action: log step start (redacted config), execute via the
27 *     executor registry (per-action timeout), log step end;
28 *      - FATAL → run FAILED (`error { code, message, actionIndex, actionType,
29 *        attempts }`), stop;
30 *      - RETRYABLE → retry decision (§7 below).
31 *
32 * Retry policy (the worker manages it — NO BullMQ attempts on this job):
33 *   retryCount = the run's attempts so far. A retryable error with
34 *   retryCount < AUTO_RUN_MAX_RETRIES (8) records FAILED then hops to
35 *   PENDING (retryCount + 1, backoff `runBackoffSeconds(retryCount+1)`
36 *   delay) and re-enqueues the SAME runId; at the cap → DEAD_LETTER (§4).
37 *   A FATAL error never retries (it is a validation/permission/quota/stub/
38 *   timeout condition, which will not heal).
39 *
40 * Idempotency notes (spec §3): the retry re-executes from action 1.
41 * UPDATE_RECORD / ASSIGN_USER / CHANGE_STATUS are idempotent by value-set;
42 * SEND_EMAIL is deduped by the email job key `automation-email:{runId}:{i}`;
43 * CREATE_PAGE is NOT value-keyed (a retry may create a second page — the run
44 * log records both ids, documented per executor).
45 */
46import type { PrismaClient } from "@prisma/client";
47import type { Redis } from "ioredis";
48import { z } from "zod";
49import {
50  AUTOMATION_MAX_ACTIONS,
51  loopGuard,
52  runBackoffSeconds,
53  truncateJson,
54  type InterpolationVars
55} from "@factory/shared";
56import { asJson } from "../../lib/prisma.js";
57import { getLogger } from "../../logger.js";
58import type { WorkerEnv } from "../../config.js";
59import { buildVars, runWithTimeout } from "./executors/helpers.js";
60import type { AutomationRunContext } from "./executors/types.js";
61import { classifyError, FatalError } from "./executors/classify.js";
62import { executeAction, executorFor, validateAction, type ValidatedAction } from "./executors/registry.js";
63import { buildConditionContext, ConditionError, evaluateCondition } from "./condition-evaluator.js";
64import { assertTransition, isTerminal } from "./run-state.js";
65import { runLog } from "./log.js";
66import { deadLetterRun, emitAutomationExecuted } from "./dead-letter.js";
67 
68const log = getLogger("automation-run");
69 
70/** Run `output` storage cap (spec §7: 128 KB, truncated flag preserved). */
71export const RUN_OUTPUT_MAX_BYTES = 128 * 1024;
72 
73/**
74 * §10 metrics. The worker has no prometheus infra, so these are lightweight
75 * Redis counters (documented as the metric surface; dashboards/Phase-15 can
76 * scrape or replace them): `automations:runs:{status}`,
77 * `automations:actions:{type}:{status}`, `automations:matched:{type}`,
78 * `automations:skipped:{reason}`. Best-effort — a Redis hiccup never fails a
79 * run. `automations:run_duration_ms` (histogram) is logged, not stored
80 * (a real histogram is Phase-15).
81 */
82async function bumpMetric(redis: Redis, key: string): Promise<void> {
83  try {
84    await redis.incr(key);
85  } catch {
86    /* best-effort */
87  }
88}
89 
90/** Run-level retry cap (spec §2). `AUTO_RUN_MAX_RETRIES` (dev-only, default 8). */
91export function autoRunMaxRetries(env: WorkerEnv): number {
92  const v = env.AUTO_RUN_MAX_RETRIES;
93  return Number.isInteger(v) && v >= 1 ? v : 8;
94}
95 
96/** Validate the job payload (defense: a corrupt job must not crash the worker). */
97export const runAutomationJobSchema = z.object({
98  runId: z.string().min(1),
99  automationId: z.string().min(1),
100  workspaceId: z.string().min(1),
101  triggerEvent: z.record(z.string(), z.unknown()),
102  eventId: z.string().min(1)
103});
104 
105export interface RunPipelineDeps {
106  prisma: PrismaClient;
107  redis: Redis;
108  env: WorkerEnv;
109}
110 
111function depthOf(triggerEvent: Record<string, unknown>): number {
112  const d = (triggerEvent.automationContext as { depth?: number } | undefined)?.depth;
113  return typeof d === "number" ? d : 0;
114}
115 
116/** The BullMQ worker handler. Returns the terminal status for observability. */
117export async function handleRunAutomation(
118  deps: RunPipelineDeps,
119  job: { data: unknown }
120): Promise<string> {
121  const parsed = runAutomationJobSchema.safeParse(job.data);
122  if (!parsed.success) {
123    log.error({ err: parsed.error.message }, "malformed run-automation job (dropped)");
124    return "FAILED";
125  }
126  const { runId, automationId, workspaceId, triggerEvent } = parsed.data;
127  const startedAt = Date.now();
128 
129  // 1. Load the run row (idempotent re-read — a retry carries the same runId).
130  const run = await deps.prisma.automationRun.findUnique({ where: { id: runId } });
131  if (!run) {
132    log.error({ runId }, "run row missing (dropped — enqueuer should have created it)");
133    return "FAILED";
134  }
135  const automation = await deps.prisma.automation.findUnique({
136    where: { id: automationId },
137    include: {
138      actions: { orderBy: { position: "asc" } },
139      conditions: { orderBy: { position: "asc" } }
140    }
141  });
142  if (!automation) {
143    await runLog(deps.prisma, runId, "error", "automation deleted before run (skipped)");
144    await markFailed(deps, runId, run.status, { code: "AUTOMATION_DELETED", message: "automation no longer exists" }, { durationMs: 0, retries: 0 });
145    return "FAILED";
146  }
147 
148  const depth = depthOf(triggerEvent);
149  const triggerType = typeof triggerEvent.type === "string" ? triggerEvent.type : "unknown";
150  const retryCount = run.retryCount;
151  const maxRetries = autoRunMaxRetries(deps.env);
152  const duration = () => Date.now() - startedAt;
153 
154  // 2. State entry. Terminal rows (SUCCEEDED/DEAD_LETTER) are duplicate job
155  //    deliveries → drop. A stale RUNNING row (crash redelivery) is
156  //    tolerated (executors re-run; idempotency notes above).
157  if (isTerminal(run.status)) {
158    log.warn({ runId, status: run.status }, "duplicate job for a terminal run (dropped)");
159    return run.status;
160  }
161  if (run.status !== "RUNNING") {
162    assertTransition(run.status, "RUNNING");
163    await deps.prisma.automationRun.update({ where: { id: runId }, data: { status: "RUNNING", startedAt: new Date() } });
164    await runLog(deps.prisma, runId, "info", "run started", { triggerType, retryCount, depth });
165  }
166 
167  const ctx = await buildRunContext(deps, runId, automation, triggerEvent);
168 
169  // 3. Loop guard (self-trigger / depth limit) — SUCCEEDED skip.
170  const guard = loopGuard(
171    triggerEvent.automationContext as { automationId?: string; depth?: number } | undefined,
172    automationId
173  );
174  if (guard.blocked) {
175    const output = { skipped: guard.reason };
176    await deps.prisma.automationRun.update({
177      where: { id: runId },
178      data: { status: "SUCCEEDED", finishedAt: new Date(), output: asJson(output) }
179    });
180    await runLog(deps.prisma, runId, "warn", `run skipped: ${guard.reason}`, {
181      depth: guard.depth,
182      reason: guard.reason === "self_trigger" ? "own event re-triggered this automation" : "chain depth limit reached"
183    });
184    await emitAutomationExecuted({
185      redis: deps.redis,
186      workspaceId,
187      automationId,
188      runId,
189      status: "SUCCEEDED",
190      triggerType,
191      durationMs: duration(),
192      depth
193    });
194    await recordTerminal(deps, { runId, automationId, workspaceId, status: "SUCCEEDED", triggerType, durationMs: duration(), retries: retryCount, depth });
195    return "SUCCEEDED";
196  }
197 
198  // 4. 50-action guard (spec: corrupt 60-action automation → FAILED RUN_ACTION_LIMIT).
199  const actions = automation.actions;
200  if (actions.length > AUTOMATION_MAX_ACTIONS) {
201    const code = "RUN_ACTION_LIMIT";
202    const msg = `automation has ${actions.length} actions (max ${AUTOMATION_MAX_ACTIONS})`;
203    await runLog(deps.prisma, runId, "error", msg, { count: actions.length });
204    await markFailed(deps, runId, "RUNNING", { code, message: msg, attempts: retryCount }, { durationMs: duration(), retries: retryCount });
205    return "FAILED";
206  }
207 
208  // 5. Condition gate.
209  const conditionTree = automation.conditions.length > 0 ? (automation.conditions[0]?.expression ?? null) : null;
210  if (conditionTree) {
211    try {
212      const recordId = typeof triggerEvent.recordId === "string" ? triggerEvent.recordId : null;
213      const cctx = await buildConditionContext(deps.prisma, triggerEvent, recordId);
214      const decision = evaluateCondition(conditionTree as never, cctx);
215      if (!decision.result) {
216        await deps.prisma.automationRun.update({
217          where: { id: runId },
218          data: { status: "SUCCEEDED", finishedAt: new Date(), output: asJson({ skipped: "conditions_not_met" }) }
219        });
220        await runLog(deps.prisma, runId, "info", "run skipped: conditions not met", {
221          evaluatedNodes: decision.evaluatedNodes,
222          ms: decision.ms,
223          warnings: decision.warnings.slice(0, 10)
224        });
225        await emitAutomationExecuted({
226          redis: deps.redis,
227          workspaceId,
228          automationId,
229          runId,
230          status: "SUCCEEDED",
231          triggerType,
232          durationMs: duration(),
233          depth
234        });
235        await recordTerminal(deps, { runId, automationId, workspaceId, status: "SUCCEEDED", triggerType, durationMs: duration(), retries: retryCount, depth });
236        return "SUCCEEDED";
237      }
238    } catch (err) {
239      if (err instanceof ConditionError) {
240        const code = err.code;
241        await runLog(deps.prisma, runId, "error", `condition evaluation failed: ${code}`, { message: err.message });
242        await markFailed(deps, runId, "RUNNING", { code, message: err.message, attempts: retryCount }, { durationMs: duration(), retries: retryCount });
243        return "FAILED";
244      }
245      throw err;
246    }
247  }
248 
249  // 6. Execute actions in position order.
250  const output = { actions: [] as Record<string, unknown>[], totalMs: 0 };
251  for (let i = 0; i < actions.length; i++) {
252    const action = actions[i]!;
253    const actionType = String(action.type);
254    await runLog(deps.prisma, runId, "info", `action ${i + 1}/${actions.length} ${actionType} started`, {
255      actionId: action.id,
256      config: (action.config as Record<string, unknown>) ?? {}
257    });
258    const actionStarted = Date.now();
259    try {
260      const validated = normalizeStoredAction(action);
261      const result = await executeActionWithTimeout(deps, validated, ctx, i);
262      const ms = Date.now() - actionStarted;
263      output.actions.push({ index: i, type: actionType, output: result.output, ms, status: "succeeded" });
264      await bumpMetric(deps.redis, `automations:actions:${actionType.toLowerCase()}:succeeded`);
265      await runLog(deps.prisma, runId, "info", `action ${i + 1}/${actions.length} ${actionType} done`, {
266        actionId: action.id,
267        ms,
268        output: result.output,
269        emittedEvents: result.emittedEvents
270      });
271      if (result.audit) {
272        await auditAction(deps, ctx, result.audit, actionType, action.id);
273      }
274    } catch (err) {
275      const cls = classifyError(err);
276      const errorObj = {
277        code: cls.code,
278        message: (err as Error)?.message ?? String(err),
279        actionIndex: i,
280        actionType,
281        attempts: retryCount
282      };
283      await runLog(deps.prisma, runId, "error", `action ${i + 1}/${actions.length} ${actionType} failed`, {
284        actionId: action.id,
285        code: cls.code,
286        retryable: cls.retryable,
287        message: errorObj.message
288      });
289      await bumpMetric(deps.redis, `automations:actions:${actionType.toLowerCase()}:failed`);
290      if (cls.retryable && retryCount < maxRetries) {
291        await scheduleRetry(deps, runId, retryCount, errorObj);
292        return "RETRYING";
293      }
294      if (cls.retryable && retryCount >= maxRetries) {
295        await deadLetterRun(deps.prisma, deps.redis, deps.env, {
296          runId,
297          workspaceId,
298          automationId,
299          automationName: automation.name,
300          fromStatus: "RUNNING",
301          code: cls.code,
302          message: errorObj.message,
303          actionIndex: i,
304          actionType,
305          attempts: retryCount,
306          triggerType,
307          durationMs: duration(),
308          depth
309        });
310        return "DEAD_LETTER";
311      }
312      await markFailed(deps, runId, "RUNNING", errorObj, { durationMs: duration(), retries: retryCount });
313      return "FAILED";
314    }
315  }
316 
317  // 7. All actions completed (or there were none) → SUCCEEDED.
318  output.totalMs = duration();
319  const { value: outputVal, truncated: outputTruncated } = truncateJson(output, RUN_OUTPUT_MAX_BYTES);
320  await deps.prisma.automationRun.update({
321    where: { id: runId },
322    data: { status: "SUCCEEDED", finishedAt: new Date(), output: asJson(outputTruncated ? { ...(outputVal as Record<string, unknown>), truncated: true } : outputVal) }
323  });
324  await runLog(deps.prisma, runId, "info", "run completed", { actions: actions.length, ms: duration() });
325  await emitAutomationExecuted({
326    redis: deps.redis,
327    workspaceId,
328    automationId,
329    runId,
330    status: "SUCCEEDED",
331    triggerType,
332    durationMs: duration(),
333    depth
334  });
335  await recordTerminal(deps, { runId, automationId, workspaceId, status: "SUCCEEDED", triggerType, durationMs: duration(), retries: retryCount, depth });
336  return "SUCCEEDED";
337}
338 
339// ---------------------------------------------------------------------------
340// Helpers
341// ---------------------------------------------------------------------------
342 
343/**
344 * §10 per-run structured log line + status counter. Called once per terminal
345 * (or skip) outcome. Best-effort.
346 */
347async function recordTerminal(
348  deps: RunPipelineDeps,
349  args: {
350    runId: string;
351    automationId: string;
352    workspaceId: string;
353    status: string;
354    triggerType: string;
355    durationMs: number;
356    retries: number;
357    depth: number;
358  }
359): Promise<void> {
360  log.info(
361    {
362      runId: args.runId,
363      automationId: args.automationId,
364      status: args.status,
365      durationMs: args.durationMs,
366      retries: args.retries,
367      triggerType: args.triggerType
368    },
369    "automation.run"
370  );
371  await bumpMetric(deps.redis, `automations:runs:${args.status.toLowerCase()}`);
372  // TASK-0133 audit catalog: EVERY terminal run is an audit row (the
373  // catalog's `automation.executed`; required metadata `automationId`,
374  // `runId`, `status`). This is the single terminal chokepoint (SUCCEEDED +
375  // every markFailed path), so the audit cannot be missed. Best-effort (an
376  // audit failure must not lose the run's bookkeeping). The DOMAIN event is
377  // depth-0 only (event-storm guard) — the audit records all depths (a
378  // complete trail, bounded by the depth guard).
379  try {
380    await deps.prisma.auditLog.create({
381      data: {
382        actorType: "SYSTEM",
383        actorId: `worker:automation`,
384        workspaceId: args.workspaceId || null,
385        resourceType: "automation_run",
386        resourceId: args.runId,
387        action: "automation.executed",
388        metadata: asJson({
389          automationId: args.automationId,
390          runId: args.runId,
391          status: args.status,
392          triggerType: args.triggerType,
393          durationMs: args.durationMs,
394          retries: args.retries,
395          depth: args.depth
396        })
397      }
398    });
399  } catch (err) {
400    log.warn({ err: (err as Error).message, runId: args.runId }, "automation.executed audit write failed");
401  }
402}
403 
404async function buildRunContext(
405  deps: RunPipelineDeps,
406  runId: string,
407  automation: { id: string; name: string; workspaceId: string; databaseId: string | null; createdById: string },
408  triggerEvent: Record<string, unknown>
409): Promise<AutomationRunContext> {
410  const recordId = typeof triggerEvent.recordId === "string" ? triggerEvent.recordId : null;
411  const depth = depthOf(triggerEvent);
412  const base = {
413    runId,
414    automationId: automation.id,
415    automationName: automation.name,
416    workspaceId: automation.workspaceId,
417    databaseId: automation.databaseId,
418    recordId,
419    triggerEvent,
420    depth,
421    input: (triggerEvent.input as Record<string, unknown> | undefined) ?? null,
422    actorUserId: automation.createdById,
423    vars: {}
424  };
425  const vars: InterpolationVars = await buildVars(deps.prisma, base);
426  return { ...base, vars };
427}
428 
429/**
430 * A stored action row may be corrupt (drift, a bad manual edit). Re-validate
431 * through the shared schema; an invalid row → a FATAL INVALID_CONFIG (no
432 * retry) rather than a cryptic executor crash.
433 */
434function normalizeStoredAction(action: { type: string; config: unknown; retries: number; timeoutMs: number }): ValidatedAction {
435  return validateAction({ type: action.type, config: action.config, retries: action.retries, timeoutMs: action.timeoutMs });
436}
437 
438async function executeActionWithTimeout(
439  deps: RunPipelineDeps,
440  validated: ValidatedAction,
441  ctx: AutomationRunContext,
442  index: number
443): Promise<Awaited<ReturnType<typeof executeAction>>> {
444  const executor = executorFor(validated.type);
445  if (!executor) throw new FatalError("INVALID_CONFIG", `no executor for ${validated.type}`);
446  return runWithTimeout(validated.timeoutMs, () => executor.execute(validated.config, ctx, deps, index));
447}
448 
449async function auditAction(
450  deps: RunPipelineDeps,
451  ctx: AutomationRunContext,
452  entry: { action: string; resourceType: string; resourceId: string; metadata?: Record<string, unknown> },
453  actionType: string,
454  actionId: string
455): Promise<void> {
456  try {
457    await deps.prisma.auditLog.create({
458      data: {
459        actorType: "SYSTEM",
460        actorId: `automation:${ctx.automationId}`,
461        workspaceId: ctx.workspaceId,
462        resourceType: entry.resourceType,
463        resourceId: entry.resourceId,
464        action: entry.action,
465        metadata: asJson({
466          ...(entry.metadata ?? {}),
467          onBehalfOf: "automation",
468          automationId: ctx.automationId,
469          runId: ctx.runId,
470          actionType,
471          actionId
472        })
473      }
474    });
475  } catch (err) {
476    log.warn({ err: (err as Error).message, runId: ctx.runId }, "action audit write failed");
477  }
478}
479 
480/**
481 * Retryable error below the cap: record the FAILED hop (the error is visible
482 * immediately) then hop to PENDING with retryCount + 1 and re-enqueue the
483 * SAME runId with the backoff delay. State path asserted: RUNNING → FAILED
484 * → PENDING (the §2 state machine); a single atomic write lands on PENDING.
485 */
486async function scheduleRetry(
487  deps: RunPipelineDeps,
488  runId: string,
489  retryCount: number,
490  errorObj: Record<string, unknown>
491): Promise<void> {
492  const nextRetry = retryCount + 1;
493  const backoffSec = runBackoffSeconds(nextRetry);
494  assertTransition("RUNNING", "FAILED");
495  assertTransition("FAILED", "PENDING");
496  await deps.prisma.automationRun.update({
497    where: { id: runId },
498    data: {
499      status: "PENDING",
500      error: asJson(errorObj),
501      retryCount: nextRetry
502    }
503  });
504  await runLog(deps.prisma, runId, "warn", `run retrying (attempt ${nextRetry})`, {
505    code: errorObj.code as string,
506    backoffSeconds: backoffSec
507  });
508 
509  // Re-enqueue with the backoff delay (no BullMQ attempts — the pipeline is
510  // the retry manager; this is a fresh job carrying the same runId).
511  const { Queue } = await import("bullmq");
512  const q = new Queue("automations", { connection: deps.redis });
513  try {
514    const row = await deps.prisma.automationRun.findUnique({
515      where: { id: runId },
516      select: { automationId: true, workspaceId: true, triggerEvent: true }
517    });
518    const te = (row?.triggerEvent as Record<string, unknown> | null) ?? {};
519    await q.add(
520      "run-automation",
521      {
522        runId,
523        automationId: row?.automationId ?? "",
524        workspaceId: row?.workspaceId ?? "",
525        triggerEvent: te,
526        eventId: (te.eventId as string) ?? runId
527      },
528      { delay: backoffSec * 1000, removeOnComplete: 1000, removeOnFail: 500 }
529    );
530  } finally {
531    await q.close();
532  }
533}
534 
535/**
536 * Mark a run FAILED (terminal; the API retry endpoint can revive it).
537 * Emits `automation.executed` at depth 0. Never throws.
538 */
539async function markFailed(
540  deps: RunPipelineDeps,
541  runId: string,
542  fromStatus: string,
543  errorObj: Record<string, unknown>,
544  opts: { durationMs: number; retries: number }
545): Promise<void> {
546  try {
547    if (fromStatus === "SUCCEEDED" || fromStatus === "DEAD_LETTER") return; // terminal
548    assertTransition(fromStatus as never, "FAILED");
549    const run = await deps.prisma.automationRun.findUnique({
550      where: { id: runId },
551      select: { triggerEvent: true, workspaceId: true, automationId: true }
552    });
553    await deps.prisma.automationRun.update({
554      where: { id: runId },
555      data: { status: "FAILED", finishedAt: new Date(), error: asJson(errorObj) }
556    });
557    const te = (run?.triggerEvent as Record<string, unknown> | null) ?? {};
558    const triggerType = typeof te.type === "string" ? te.type : "unknown";
559    await emitAutomationExecuted({
560      redis: deps.redis,
561      workspaceId: run?.workspaceId ?? "",
562      automationId: run?.automationId ?? "",
563      runId,
564      status: "FAILED",
565      triggerType,
566      durationMs: opts.durationMs,
567      error: { code: (errorObj.code as string) ?? "UNKNOWN" },
568      depth: depthOf(te)
569    });
570    await recordTerminal(deps, {
571      runId,
572      automationId: run?.automationId ?? "",
573      workspaceId: run?.workspaceId ?? "",
574      status: "FAILED",
575      triggerType,
576      durationMs: opts.durationMs,
577      retries: opts.retries,
578      depth: depthOf(te)
579    });
580  } catch (err) {
581    log.warn({ err: (err as Error).message, runId }, "markFailed bookkeeping failed");
582  }
583}
584