dead-letter.ts · Source

dead-letter.tsapps/worker/src/processors/automations/dead-letter.tsraw ← automations5,621 B← Back to the app
1/**
2 * apps/worker — automation dead-letter + terminal bookkeeping (TASK-0113 §4/§8).
3 *
4 * `deadLetterRun` is the single place a run becomes DEAD_LETTER (retryCount
5 * hit the cap): terminal run row, the frozen `automation.executed` domain
6 * event (`status: "dead_letter"`), an in-app notification to every
7 * Admin+Owner of the workspace, an `automation.dead_lettered` audit row
8 * (system actor), and the `automations:dead_letter{workspace}` counter.
9 *
10 * `emitAutomationExecuted` is the §8 emitter: the frozen payload
11 * `{ workspaceId, automationId, runId, status, triggerType, durationMs,
12 * error? }` is emitted on EVERY terminal status, but ONLY at depth 0 — a
13 * chained run (depth ≥ 1) does NOT re-emit, so an A→B→… chain produces
14 * exactly one `automation.executed` (the root's), not a storm per hop.
15 */
16import type { PrismaClient } from "@prisma/client";
17import type { Redis } from "ioredis";
18import { asJson } from "../../lib/prisma.js";
19import { getLogger } from "../../logger.js";
20import { assertTransition } from "./run-state.js";
21import { runLog } from "./log.js";
22import type { WorkerEnv } from "../../config.js";
23import type { AutomationRunStatus } from "@prisma/client";
24 
25const log = getLogger("automation-dlq");
26 
27export interface ExecutedEventInput {
28  redis: Redis;
29  workspaceId: string;
30  automationId: string;
31  runId: string;
32  status: AutomationRunStatus;
33  triggerType: string;
34  durationMs: number;
35  error?: { code: string };
36  /** Incoming loop depth (0 for human/API-originated). Only depth 0 emits. */
37  depth: number;
38}
39 
40/**
41 * Emit the frozen `automation.executed` event. No-op (returns false) when
42 * depth ≥ 1 — chained runs don't re-emit (event-storm guard, §8).
43 */
44export async function emitAutomationExecuted(input: ExecutedEventInput): Promise<boolean> {
45  if (input.depth >= 1) return false;
46  const payload: Record<string, unknown> = {
47    workspaceId: input.workspaceId,
48    automationId: input.automationId,
49    runId: input.runId,
50    status: input.status,
51    triggerType: input.triggerType,
52    durationMs: input.durationMs
53  };
54  if (input.error) payload.error = { code: input.error.code };
55  await input.redis.xadd(
56    "factory:events",
57    "*",
58    "event",
59    "automation.executed",
60    "payload",
61    JSON.stringify(payload),
62    "ts",
63    new Date().toISOString()
64  );
65  return true;
66}
67 
68/**
69 * Mark a run DEAD_LETTER + all the §4 side effects. Idempotent-ish: only
70 * valid from FAILED (the transition is asserted). Never throws — a DLQ
71 * bookkeeping failure is a WARN (the run row is already terminal-safe).
72 */
73export async function deadLetterRun(
74  prisma: PrismaClient,
75  redis: Redis,
76  env: WorkerEnv,
77  args: {
78    runId: string;
79    workspaceId: string;
80    automationId: string;
81    automationName: string;
82    fromStatus: AutomationRunStatus;
83    code: string;
84    message: string;
85    actionIndex?: number;
86    actionType?: string;
87    attempts: number;
88    triggerType: string;
89    durationMs: number;
90    depth: number;
91  }
92): Promise<void> {
93  try {
94    assertTransition(args.fromStatus, "DEAD_LETTER");
95    const now = new Date();
96    await prisma.automationRun.update({
97      where: { id: args.runId },
98      data: {
99        status: "DEAD_LETTER",
100        finishedAt: now,
101        error: asJson({
102          code: args.code,
103          message: args.message,
104          actionIndex: args.actionIndex ?? null,
105          actionType: args.actionType ?? null,
106          attempts: args.attempts
107        })
108      }
109    });
110    await runLog(prisma, args.runId, "error", "run dead-lettered", {
111      code: args.code,
112      message: args.message,
113      attempts: args.attempts,
114      actionIndex: args.actionIndex ?? null,
115      actionType: args.actionType ?? null
116    });
117 
118    await emitAutomationExecuted({
119      redis,
120      workspaceId: args.workspaceId,
121      automationId: args.automationId,
122      runId: args.runId,
123      status: "DEAD_LETTER",
124      triggerType: args.triggerType,
125      durationMs: args.durationMs,
126      error: { code: args.code },
127      depth: args.depth
128    });
129 
130    // Admin notification (in-app + email when SMTP is configured).
131    const { notifyWorkspaceAdmins } = await import("../../lib/notify.js");
132    await notifyWorkspaceAdmins(
133      prisma,
134      redis,
135      env.SMTP_HOST,
136      args.workspaceId,
137      `Automation "${args.automationName}" dead-lettered`,
138      `Failed after ${args.attempts} attempts: ${args.code} — ${args.message}`,
139      { type: "automation", id: args.automationId },
140      "AUTOMATION"
141    );
142 
143    // Audit (system actor).
144    try {
145      await prisma.auditLog.create({
146        data: {
147          actorType: "SYSTEM",
148          actorId: `automation:${args.automationId}`,
149          workspaceId: args.workspaceId,
150          resourceType: "automation_run",
151          resourceId: args.runId,
152          action: "automation.dead_lettered",
153          metadata: asJson({ automationId: args.automationId, runId: args.runId, code: args.code, attempts: args.attempts })
154        }
155      });
156    } catch (err) {
157      log.warn({ err: (err as Error).message, runId: args.runId }, "dead-letter audit write failed");
158    }
159 
160    // Metric counter (lightweight — no prometheus infra in the worker).
161    await redis.incr(`automations:dead_letter:${args.workspaceId}`).catch(() => undefined);
162 
163    log.error(
164      { runId: args.runId, automationId: args.automationId, code: args.code, attempts: args.attempts },
165      "automation run → DEAD_LETTER"
166    );
167  } catch (err) {
168    log.error({ err: (err as Error).message, runId: args.runId }, "dead-letter bookkeeping failed");
169  }
170}
171