automations.service.ts

automations.service.tsapps/api/src/modules/automations/automations.service.tsraw ← automations33,044 B← Back to the app
1/**
2 * TASK-0110 — automation service: CRUD, sub-resource upserts, status machine,
3 * manual runs, runs listing/detail/retry, and the incoming-webhook secret
4 * lifecycle (generate-once / KMS-wrapped / regenerate).
5 *
6 * Permission model (spec §1): automations are an EDIT capability —
7 * Owner/Admin always (FULL_ACCESS); Member with EDIT on the scoped database
8 * when `databaseId` is set. Enforcement via `assertCan` (server-side only).
9 *
10 * The BullMQ enqueue degrades best-effort (same pattern as the files
11 * pipeline): a queue outage logs a WARN and the run row stays PENDING —
12 * the TASK-0113 watchdog reclaims stale PENDING/RUNNING rows, so nothing is
13 * lost (documented in docs/architecture/automation.md).
14 */
15import { Prisma } from "@prisma/client";
16import { Inject, Injectable, Logger } from "@nestjs/common";
17import type { Request } from "express";
18import {
19  MANUAL_RUN_INPUT_MAX_BYTES,
20  RUN_INPUT_MAX_BYTES,
21  TERMINAL_RUN_STATUSES,
22  generateWebhookSecret,
23  isValidAutomationTransition,
24  truncateJson,
25  validateActionDto,
26  webhookSecretHash,
27  type AutomationStatus
28} from "@factory/shared";
29import { v4 as uuid } from "uuid";
30import { jobTraceMetadata } from "../../common/trace-metadata.js";
31import { EventBus } from "../../common/event-bus.js";
32import { ApiError } from "../../common/errors.js";
33import { RedisService } from "../../common/redis.service.js";
34import { AuditService } from "../../common/audit.service.js";
35import { assertCan } from "../../lib/authorize.js";
36import { PrismaService } from "../../common/prisma.service.js";
37import {
38  assertValidActions,
39  assertValidCondition,
40  assertValidTriggers,
41  type CreateAutomationInput,
42  type ListRunsQueryInput,
43  type PutActionsInput,
44  type PutConditionsInput,
45  type PutTriggersInput,
46  type RunInputInput,
47  type SetStatusInput,
48  type UpdateAutomationInput
49} from "./dto.js";
50import { aes256gcmDecrypt, aes256gcmEncrypt, keyFromSecret } from "../../lib/crypto.js";
51import { loadEnv } from "../../config/env.js";
52 
53type AutomationRow = {
54  id: string;
55  workspaceId: string;
56  name: string;
57  description: string | null;
58  databaseId: string | null;
59  status: string;
60  createdById: string;
61  createdAt: Date;
62  updatedAt: Date;
63  triggers: { id: string; type: string; config: unknown; position: number }[];
64  conditions: { id: string; expression: unknown; position: number }[];
65  actions: {
66    id: string;
67    type: string;
68    config: unknown;
69    position: number;
70    retries: number;
71    timeoutMs: number;
72  }[];
73};
74 
75/** Coerce an unknown value to a Prisma Json input (the API's DTOs hand us
76 * plain JS from Zod; Prisma's `InputJsonValue` type is narrower). */
77function toJson(v: unknown): Prisma.InputJsonValue {
78  return v as Prisma.InputJsonValue;
79}
80 
81@Injectable()
82export class AutomationsService {
83  private readonly log = new Logger(AutomationsService.name);
84 
85  constructor(
86    @Inject(PrismaService) private readonly prisma: PrismaService,
87    @Inject(RedisService) private readonly redis: RedisService,
88    @Inject(EventBus) private readonly events: EventBus,
89    @Inject(AuditService) private readonly audit: AuditService
90  ) {}
91 
92  // ---------------------------------------------------------------------
93  // Permission helpers
94  // ---------------------------------------------------------------------
95 
96  private async requireAutomation(id: string, userId: string, level: "VIEW" | "EDIT"): Promise<AutomationRow> {
97    const a = await this.prisma.automation.findUnique({
98      where: { id },
99      include: {
100        triggers: { orderBy: { position: "asc" } },
101        conditions: { orderBy: { position: "asc" } },
102        actions: { orderBy: { position: "asc" } }
103      }
104    });
105    if (!a) throw ApiError.notFound("Automation not found.");
106    const resource = a.databaseId ? { databaseId: a.databaseId } : undefined;
107    await assertCan(this.prisma, { workspaceId: a.workspaceId, userId, resource }, level);
108    return a;
109  }
110 
111  private async requireWorkspaceView(workspaceId: string, userId: string): Promise<void> {
112    await assertCan(this.prisma, { workspaceId, userId }, "VIEW");
113  }
114 
115  // ---------------------------------------------------------------------
116  // CRUD
117  // ---------------------------------------------------------------------
118 
119  async list(workspaceId: string, userId: string, query: { status?: string; cursor?: string; limit?: number }): Promise<{ data: unknown[]; nextCursor: string | null }> {
120    await this.requireWorkspaceView(workspaceId, userId);
121    const limit = Math.min(Math.max(query.limit ?? 50, 1), 200);
122    const cursor = query.cursor
123      ? (() => {
124          // Invalid cursor (bad base64url / non-JSON) → 400, not 500.
125          let c: { createdAt?: string; id?: string };
126          try {
127            c = JSON.parse(Buffer.from(query.cursor as string, "base64url").toString("utf8")) as { createdAt?: string; id?: string };
128          } catch {
129            throw ApiError.validation("invalid cursor");
130          }
131          if (!c || typeof c.createdAt !== "string" || typeof c.id !== "string") throw ApiError.validation("invalid cursor");
132          return { createdAt: c.createdAt, id: c.id };
133        })()
134      : undefined;
135 
136    const where: Record<string, unknown> = { workspaceId };
137    if (query.status) where.status = query.status;
138    if (cursor) {
139      where.OR = [
140        { createdAt: { lt: new Date(cursor.createdAt) } },
141        { createdAt: { equals: new Date(cursor.createdAt) }, id: { lt: cursor.id } }
142      ];
143    }
144 
145    const rows = await this.prisma.automation.findMany({
146      where: where as Prisma.AutomationWhereInput,
147      include: {
148        triggers: true,
149        conditions: true,
150        actions: true
151      },
152      orderBy: { createdAt: "desc" },
153      take: limit + 1
154    });
155    const more = rows.length > limit;
156    const page = rows.slice(0, limit);
157 
158    // Latest run per automation in this page (one query, grouped in JS).
159    const ids = page.map((p) => p.id);
160    const runs = ids.length
161      ? await this.prisma.automationRun.findMany({
162          where: { automationId: { in: ids } },
163          orderBy: { createdAt: "desc" },
164          select: { id: true, automationId: true, status: true, createdAt: true, finishedAt: true }
165        })
166      : [];
167    const latest = new Map<string, (typeof runs)[number]>();
168    for (const r of runs) if (!latest.has(r.automationId)) latest.set(r.automationId, r);
169 
170    return {
171      data: page.map((a) => ({
172        ...this.toSummary(a),
173        latestRun: latest.get(a.id) ?? null
174      })),
175      nextCursor: more && page.length
176        ? Buffer.from(JSON.stringify({ createdAt: page[page.length - 1]!.createdAt.toISOString(), id: page[page.length - 1]!.id })).toString("base64url")
177        : null
178    };
179  }
180 
181  async get(id: string, userId: string): Promise<unknown> {
182    const a = await this.requireAutomation(id, userId, "VIEW");
183    const [run] = await this.prisma.automationRun.findMany({
184      where: { automationId: id },
185      orderBy: { createdAt: "desc" },
186      take: 1,
187      select: { id: true, status: true, createdAt: true, finishedAt: true, retryCount: true, error: true }
188    });
189    return {
190      ...this.toDetail(a),
191      latestRun: run ?? null
192    };
193  }
194 
195  async create(workspaceId: string, userId: string, input: CreateAutomationInput, req: Request): Promise<unknown> {
196    // databaseId (optional) must exist in this workspace and not be archived.
197    if (input.databaseId) {
198      const db = await this.prisma.database.findUnique({ where: { id: input.databaseId } });
199      if (!db || db.workspaceId !== workspaceId) throw ApiError.unprocessable("databaseId must reference a database in this workspace", [{ field: "databaseId", issue: "unknown database" }]);
200      if (db.isArchived) throw ApiError.unprocessable("databaseId must reference an active database", [{ field: "databaseId", issue: "archived" }]);
201    }
202    await assertCan(this.prisma, { workspaceId, userId, resource: input.databaseId ? { databaseId: input.databaseId } : undefined }, "EDIT");
203 
204    // Validate optional inline sub-resources (same rules as the PUTs).
205    if (input.triggers) assertValidTriggers(input.triggers as PutTriggersInput["triggers"]);
206    if (input.condition !== undefined) assertValidCondition(input.condition);
207    if (input.actions) assertValidActions(input.actions as PutActionsInput["actions"]);
208 
209    const created = await this.prisma.$transaction(async (tx) => {
210      const a = await tx.automation.create({
211        data: { workspaceId, name: input.name, description: input.description ?? null, databaseId: input.databaseId ?? null, createdById: userId }
212      });
213      await this.writeSubResources(tx, a.id, input.triggers, input.condition, input.actions);
214      return a;
215    });
216    await this.audit.audit(req, req.auth, {
217      action: "automation.created",
218      resourceType: "automation",
219      resourceId: created.id,
220      workspaceId,
221      metadata: { name: created.name }
222    });
223    return this.get(created.id, userId);
224  }
225 
226  async update(id: string, userId: string, input: UpdateAutomationInput, req: Request): Promise<unknown> {
227    const a = await this.requireAutomation(id, userId, "EDIT");
228    if (input.databaseId !== undefined) {
229      if (input.databaseId === null) {
230        // un-scope — allowed.
231      } else {
232        const db = await this.prisma.database.findUnique({ where: { id: input.databaseId } });
233        if (!db || db.workspaceId !== a.workspaceId) throw ApiError.unprocessable("databaseId must reference a database in this workspace", [{ field: "databaseId", issue: "unknown database" }]);
234        if (db.isArchived) throw ApiError.unprocessable("databaseId must reference an active database", [{ field: "databaseId", issue: "archived" }]);
235        await assertCan(this.prisma, { workspaceId: a.workspaceId, userId, resource: { databaseId: input.databaseId } }, "EDIT");
236      }
237    }
238    await this.prisma.automation.update({
239      where: { id },
240      data: {
241        ...(input.name !== undefined ? { name: input.name } : {}),
242        ...(input.description !== undefined ? { description: input.description } : {}),
243        ...(input.databaseId !== undefined ? { databaseId: input.databaseId } : {})
244      }
245    });
246    await this.audit.audit(req, req.auth, {
247      action: "automation.updated",
248      resourceType: "automation",
249      resourceId: id,
250      workspaceId: a.workspaceId,
251      metadata: { changed: Object.keys(input) }
252    });
253    return this.get(id, userId);
254  }
255 
256  async setStatus(id: string, userId: string, to: SetStatusInput["status"], req: Request): Promise<unknown> {
257    const a = await this.requireAutomation(id, userId, "EDIT");
258    const from = a.status as AutomationStatus;
259    if (from === to) throw ApiError.conflict(`Automation is already ${from}.`, [{ reason: "no_op" }]);
260    if (!isValidAutomationTransition(from, to)) {
261      throw ApiError.conflict(`Invalid status transition ${from} → ${to}.`, [{ reason: "invalid_transition", from, to }]);
262    }
263    // Spec §2: the transition itself is the ONLY gate (6 valid pairs, 409
264    // otherwise). A DRAFT/ACTIVE automation without triggers simply matches
265    // nothing — the matcher skips it; no completeness gate here (recorded).
266    await this.prisma.automation.update({ where: { id }, data: { status: to } });
267    await this.audit.audit(req, req.auth, {
268      action: "automation.status_changed",
269      resourceType: "automation",
270      resourceId: id,
271      workspaceId: a.workspaceId,
272      metadata: { from, to }
273    });
274    this.log.log(`automation ${id} status ${from} → ${to}`);
275    return this.get(id, userId);
276  }
277 
278  /** DELETE → ARCHIVED (soft; runs retained for audit). */
279  async remove(id: string, userId: string, req: Request): Promise<void> {
280    const a = await this.requireAutomation(id, userId, "EDIT");
281    if (a.status === "ARCHIVED") return;
282    await this.prisma.automation.update({ where: { id }, data: { status: "ARCHIVED" } });
283    await this.audit.audit(req, req.auth, {
284      action: "automation.status_changed",
285      resourceType: "automation",
286      resourceId: id,
287      workspaceId: a.workspaceId,
288      metadata: { from: a.status, to: "ARCHIVED" }
289    });
290  }
291 
292  // ---------------------------------------------------------------------
293  // Sub-resource PUTs (editor contract — replace-set semantics)
294  // ---------------------------------------------------------------------
295 
296  async putTriggers(id: string, userId: string, body: PutTriggersInput, req: Request): Promise<unknown> {
297    const a = await this.requireAutomation(id, userId, "EDIT");
298    assertValidTriggers(body.triggers);
299    // WEBHOOK triggers: mint the one-time secret on FIRST create (the
300    // plaintext is returned exactly once, in this response). On RE-SAVE the
301    // client echoes back only the public `webhookId` (the KMS-wrapped secret
302    // is stripped on read and can never be sent back), so the stored
303    // `secretHash`/`secretEncrypted` must be preserved from the previous
304    // row — otherwise the editor's core operation (PUT triggers) would
305    // silently clobber the secret and break the live endpoint.
306    const storedByWebhookId = new Map<string, Record<string, unknown>>();
307    for (const t of a.triggers) if (t.type === "WEBHOOK") {
308      const wid = this.webhookIdOf(t.config);
309      if (wid) storedByWebhookId.set(String(wid), t.config as Record<string, unknown>);
310    }
311    const generated: { webhookId: string; secret: string }[] = [];
312    const triggers = body.triggers.map((t) => {
313      const config = { ...(t.config as Record<string, unknown>) } as Record<string, unknown>;
314      if (t.type === "WEBHOOK" && !config.webhookId) {
315        const secret = generateWebhookSecret();
316        config.webhookId = uuid().replace(/-/g, "").slice(0, 24);
317        config.secretHash = webhookSecretHash(secret);
318        config.secretEncrypted = this.wrapSecret(secret);
319        generated.push({ webhookId: String(config.webhookId), secret });
320      } else if (t.type === "WEBHOOK" && config.webhookId) {
321        const wid = String(config.webhookId);
322        const stored = storedByWebhookId.get(wid);
323        if (!stored) {
324          throw ApiError.validation("WEBHOOK trigger references an unknown webhookId — create a new webhook trigger instead.", [
325            { field: "triggers", issue: "unknown_webhook_id", received: wid }
326          ]);
327        }
328        // Preserve the server-managed secret material (never client-supplied).
329        config.secretHash = stored.secretHash;
330        config.secretEncrypted = stored.secretEncrypted;
331      }
332      return { type: t.type, config };
333    });
334 
335    await this.prisma.$transaction(async (tx) => {
336      await tx.automationTrigger.deleteMany({ where: { automationId: id } });
337      await tx.automationTrigger.createMany({
338        data: triggers.map((t, i) => ({ automationId: id, type: t.type, config: toJson(t.config), position: i }))
339      });
340    });
341    await this.audit.audit(req, req.auth, {
342      action: "automation.updated",
343      resourceType: "automation",
344      resourceId: id,
345      workspaceId: a.workspaceId,
346      metadata: { subresource: "triggers", count: triggers.length }
347    });
348    return {
349      automationId: id,
350      triggers: this.sanitizeTriggers(a.id, triggers),
351      webhooks: generated.map((g) => ({
352        webhookId: g.webhookId,
353        /** Plaintext — shown ONCE. It is KMS-wrapped at rest and never returned again. */
354        secret: g.secret,
355        url: `${loadEnv().WEBHOOK_BASE_URL}/webhooks/incoming/${g.webhookId}`
356      }))
357    };
358  }
359 
360  async putConditions(id: string, userId: string, body: PutConditionsInput, req: Request): Promise<unknown> {
361    const a = await this.requireAutomation(id, userId, "EDIT");
362    assertValidCondition(body.condition ?? null);
363    const tree = body.condition ?? null;
364    await this.prisma.$transaction(async (tx) => {
365      await tx.automationCondition.deleteMany({ where: { automationId: id } });
366      if (tree) await tx.automationCondition.create({ data: { automationId: id, expression: toJson(tree), position: 0 } });
367    });
368    await this.audit.audit(req, req.auth, {
369      action: "automation.updated",
370      resourceType: "automation",
371      resourceId: id,
372      workspaceId: a.workspaceId,
373      metadata: { subresource: "conditions", present: Boolean(tree) }
374    });
375    return { automationId: id, condition: tree };
376  }
377 
378  /**
379   * Shared `validateActionDto` + normalization: applies the schema defaults
380   * (e.g. UPDATE_RECORD.recordId → "from_trigger") so the stored row is
381   * self-contained for the worker (which re-validates but trusts the stored
382   * shape). The DTO layer has already 422'd invalid entries — this never
383   * throws for reachable input.
384   */
385  private normalizeActions(actions: PutActionsInput["actions"]) {
386    return actions.map((act) => {
387      const r = validateActionDto({ type: act.type as never, config: act.config, retries: act.retries, timeoutMs: act.timeoutMs });
388      if (!r.ok) throw ApiError.unprocessable(`Invalid action config.`, r.errors.map((e) => ({ field: e.path, issue: e.message })));
389      return r.normalized;
390    });
391  }
392 
393  async putActions(id: string, userId: string, body: PutActionsInput, req: Request): Promise<unknown> {
394    const a = await this.requireAutomation(id, userId, "EDIT");
395    assertValidActions(body.actions);
396    const normalized = this.normalizeActions(body.actions);
397    await this.prisma.$transaction(async (tx) => {
398      await tx.automationAction.deleteMany({ where: { automationId: id } });
399      await tx.automationAction.createMany({
400        data: normalized.map((act, i) => ({
401          automationId: id,
402          type: act.type,
403          config: toJson(act.config),
404          position: i,
405          retries: act.retries,
406          timeoutMs: act.timeoutMs
407        }))
408      });
409    });
410    await this.audit.audit(req, req.auth, {
411      action: "automation.updated",
412      resourceType: "automation",
413      resourceId: id,
414      workspaceId: a.workspaceId,
415      metadata: { subresource: "actions", count: body.actions.length }
416    });
417    return this.get(id, userId);
418  }
419 
420  // ---------------------------------------------------------------------
421  // Webhook secret lifecycle
422  // ---------------------------------------------------------------------
423 
424  /** Regenerate the webhook trigger's secret (invalidates the old one). */
425  async regenerateWebhookSecret(id: string, userId: string, req: Request): Promise<unknown> {
426    const a = await this.requireAutomation(id, userId, "EDIT");
427    const wt = a.triggers.find((t) => t.type === "WEBHOOK");
428    if (!wt) throw ApiError.notFound("Automation has no WEBHOOK trigger.");
429    const cfg = (wt.config ?? {}) as Record<string, unknown>;
430    const secret = generateWebhookSecret();
431    const next = {
432      ...cfg,
433      secretHash: webhookSecretHash(secret),
434      secretEncrypted: this.wrapSecret(secret)
435    };
436    await this.prisma.automationTrigger.update({ where: { id: wt.id }, data: { config: toJson(next) } });
437    await this.audit.audit(req, req.auth, {
438      action: "automation.webhook_secret_regenerated",
439      resourceType: "automation",
440      resourceId: id,
441      workspaceId: a.workspaceId,
442      metadata: { webhookId: cfg.webhookId }
443    });
444    return {
445      webhookId: String(cfg.webhookId),
446      secret,
447      url: `${loadEnv().WEBHOOK_BASE_URL}/webhooks/incoming/${cfg.webhookId}`
448    };
449  }
450 
451  /**
452   * KMS key for webhook secret wrapping (D5): `AUTOMATION_WEBHOOK_KMS_KEY`
453   * when set, else `MASTER_ENCRYPTION_KEY` (single-key deployments).
454   */
455  private kmsKey(): Buffer {
456    const env = loadEnv();
457    const key = env.AUTOMATION_WEBHOOK_KMS_KEY || env.MASTER_ENCRYPTION_KEY;
458    if (!key) throw new Error("AUTOMATION_WEBHOOK_KMS_KEY (or MASTER_ENCRYPTION_KEY) is required for WEBHOOK triggers");
459    return keyFromSecret(key);
460  }
461 
462  private wrapSecret(secret: string): string {
463    return aes256gcmEncrypt(secret, this.kmsKey());
464  }
465 
466  /** KMS-unwrap the stored secret (used by the incoming-webhook endpoint
467   * for signature verification). */
468  unwrapSecret(encrypted: string): string {
469    return aes256gcmDecrypt(encrypted, this.kmsKey());
470  }
471 
472  private webhookIdOf(config: unknown): string | undefined {
473    return config && typeof config === "object" ? (config as Record<string, unknown>).webhookId as string | undefined : undefined;
474  }
475 
476  /** Strip server-managed fields from trigger configs before returning them. */
477  private sanitizeTriggers(_automationId: string, triggers: { type: string; config: Record<string, unknown> }[]) {
478    return triggers.map((t) => {
479      const config = { ...t.config };
480      if (t.type === "WEBHOOK") {
481        delete config.secretEncrypted;
482        delete config.secretHash;
483        const wh = config.webhookId as string | undefined;
484        config.url = wh ? `${loadEnv().WEBHOOK_BASE_URL}/webhooks/incoming/${wh}` : undefined;
485      }
486      return { type: t.type, config };
487    });
488  }
489 
490  // ---------------------------------------------------------------------
491  // Manual run (202 { runId }) — DRAFT dry-run path included
492  // ---------------------------------------------------------------------
493 
494  async manualRun(id: string, userId: string, body: RunInputInput, req: Request): Promise<{ runId: string }> {
495    const a = await this.requireAutomation(id, userId, "EDIT");
496    if (a.actions.length === 0) throw ApiError.unprocessable("Cannot run: automation has no actions.", [{ reason: "no_actions" }]);
497    const triggerEvent: Record<string, unknown> = { type: "manual", actorId: userId };
498    if (body.input !== undefined) {
499      const { value, truncated } = truncateJson(body.input, MANUAL_RUN_INPUT_MAX_BYTES);
500      triggerEvent.input = value;
501      if (truncated) triggerEvent.inputTruncated = true;
502    }
503    const run = await this.prisma.automationRun.create({
504      data: {
505        workspaceId: a.workspaceId,
506        automationId: id,
507        status: "PENDING",
508        triggerEvent: toJson(triggerEvent),
509        input: toJson(truncateJson(triggerEvent, RUN_INPUT_MAX_BYTES).value),
510        retryCount: 0
511      }
512    });
513    await this.enqueueRun(run.id, id, a.workspaceId, triggerEvent);
514    await this.audit.audit(req, req.auth, {
515      action: "automation.manual_triggered",
516      resourceType: "automation",
517      resourceId: id,
518      workspaceId: a.workspaceId,
519      metadata: { runId: run.id }
520    });
521    return { runId: run.id };
522  }
523 
524  /**
525   * Enqueue `run-automation` on the `automations` BullMQ queue (TASK-0113
526   * job shape: `{ runId, automationId, workspaceId, triggerEvent, eventId }`).
527   * Best-effort: on queue failure the run stays PENDING and the watchdog
528   * reclaims it (documented degrade, same as the files pipeline).
529   */
530  private async enqueueRun(runId: string, automationId: string, workspaceId: string, triggerEvent: Record<string, unknown>): Promise<void> {
531    await this.enqueueRunPublic(runId, automationId, workspaceId, triggerEvent, uuid());
532  }
533 
534  /** Public variant (the incoming-webhook controller supplies its own eventId). */
535  async enqueueRunPublic(runId: string, automationId: string, workspaceId: string, triggerEvent: Record<string, unknown>, eventId: string): Promise<void> {
536    try {
537      const { Queue } = await import("bullmq");
538      const q = new Queue("automations", { connection: this.redis.getRedis().options as never });
539      await q.add("run-automation", { runId, automationId, workspaceId, triggerEvent, eventId }, {
540        removeOnComplete: 1000,
541        removeOnFail: 500,
542        ...jobTraceMetadata()
543      });
544      await q.close();
545    } catch (err) {
546      this.log.warn(`automation run ${runId} enqueue failed (run stays PENDING; watchdog reclaims): ${(err as Error).message}`);
547    }
548  }
549 
550  /**
551   * Create a PENDING run for a webhook trigger (the incoming controller
552   * already verified signature + rate limit). `input` is the truncated
553   * triggerEvent for run storage.
554   */
555  async createIncomingRun(opts: {
556    workspaceId: string;
557    automationId: string;
558    triggerEvent: Record<string, unknown>;
559    input: unknown;
560  }): Promise<{ id: string }> {
561    const run = await this.prisma.automationRun.create({
562      data: {
563        workspaceId: opts.workspaceId,
564        automationId: opts.automationId,
565        status: "PENDING",
566        triggerEvent: toJson(opts.triggerEvent),
567        input: toJson(opts.input),
568        retryCount: 0
569      }
570    });
571    return { id: run.id };
572  }
573 
574  /** Minimal webhook hit log (no cookie/auth/signature values ever). */
575  logWebhookHit(webhookId: string, automationId: string, runId: string, safeHeaders: Record<string, string>): void {
576    this.log.log(`webhook ${webhookId} → automation ${automationId} → run ${runId} headers=${JSON.stringify(safeHeaders)}`);
577  }
578 
579  // ---------------------------------------------------------------------
580  // Runs (list / detail / retry) — the TASK-0114 history UI contract
581  // ---------------------------------------------------------------------
582 
583  async listRuns(
584    id: string,
585    userId: string,
586    query: ListRunsQueryInput
587  ): Promise<{ data: unknown[]; nextCursor: string | null }> {
588    const a = await this.requireAutomation(id, userId, "EDIT");
589    const limit = query.limit ?? 50;
590    const where: Record<string, unknown> = { automationId: id, ...(query.status ? { status: query.status } : {}) };
591    if (query.cursor) {
592      // Invalid cursor (bad base64url / non-JSON) → 400, not 500.
593      let c: { createdAt?: string; id?: string };
594      try {
595        c = JSON.parse(Buffer.from(query.cursor, "base64url").toString("utf8")) as { createdAt?: string; id?: string };
596      } catch {
597        throw ApiError.validation("invalid cursor");
598      }
599      if (!c || typeof c.createdAt !== "string" || typeof c.id !== "string") throw ApiError.validation("invalid cursor");
600      where.OR = [
601        { createdAt: { lt: new Date(c.createdAt) } },
602        { createdAt: { equals: new Date(c.createdAt) }, id: { lt: c.id } }
603      ];
604    }
605    const rows = await this.prisma.automationRun.findMany({
606      where: where as Prisma.AutomationRunWhereInput,
607      orderBy: { createdAt: "desc" },
608      take: limit + 1,
609      select: { id: true, status: true, triggerEvent: true, startedAt: true, finishedAt: true, retryCount: true, error: true, createdAt: true }
610    });
611    const more = rows.length > limit;
612    const page = rows.slice(0, limit);
613    return {
614      data: page.map((r) => this.toRunSummary(r, a.workspaceId)),
615      nextCursor:
616        more && page.length
617          ? Buffer.from(JSON.stringify({ createdAt: page[page.length - 1]!.createdAt.toISOString(), id: page[page.length - 1]!.id })).toString("base64url")
618          : null
619    };
620  }
621 
622  async getRun(id: string, runId: string, userId: string): Promise<unknown> {
623    const a = await this.requireAutomation(id, userId, "EDIT");
624    const run = await this.prisma.automationRun.findFirst({
625      where: { id: runId, automationId: id },
626      select: {
627        id: true,
628        status: true,
629        triggerEvent: true,
630        input: true,
631        output: true,
632        error: true,
633        startedAt: true,
634        finishedAt: true,
635        retryCount: true,
636        createdAt: true,
637        logs: { orderBy: { createdAt: "asc" }, select: { level: true, message: true, data: true, createdAt: true } }
638      }
639    });
640    if (!run) throw ApiError.notFound("Run not found.");
641    return {
642      ...this.toRunSummary(run as never, a.workspaceId),
643      input: run.input,
644      output: run.output,
645      logs: run.logs
646    };
647  }
648 
649  /** Retry a FAILED/DEAD_LETTER run: reset retryCount, re-enqueue (spec §4e). */
650  async retryRun(id: string, runId: string, userId: string, req: Request): Promise<{ runId: string; status: string }> {
651    const a = await this.requireAutomation(id, userId, "EDIT");
652    const run = await this.prisma.automationRun.findFirst({ where: { id: runId, automationId: id } });
653    if (!run) throw ApiError.notFound("Run not found.");
654    if (!(TERMINAL_RUN_STATUSES as readonly string[]).includes(run.status) || run.status === "SUCCEEDED") {
655      throw ApiError.conflict(`Only FAILED/DEAD_LETTER runs can be retried (run is ${run.status}).`, [{ reason: "not_retryable", status: run.status }]);
656    }
657    await this.prisma.automationRun.update({
658      where: { id: run.id },
659      data: { status: "PENDING", retryCount: 0, startedAt: null, finishedAt: null, error: Prisma.JsonNull }
660    });
661    const te = run.triggerEvent as Record<string, unknown> | null;
662    await this.enqueueRun(run.id, id, a.workspaceId, te ?? { type: "manual" });
663    await this.audit.audit(req, req.auth, {
664      action: "automation.run_retried",
665      resourceType: "automation",
666      resourceId: id,
667      workspaceId: a.workspaceId,
668      metadata: { runId: run.id }
669    });
670    return { runId: run.id, status: "PENDING" };
671  }
672 
673  // ---------------------------------------------------------------------
674  // DTO shaping
675  // ---------------------------------------------------------------------
676 
677  private toSummary(a: {
678    id: string;
679    name: string;
680    description: string | null;
681    databaseId: string | null;
682    status: string;
683    createdAt: Date;
684    updatedAt: Date;
685    createdById: string;
686    triggers: { type: string }[];
687    conditions: unknown[];
688    actions: { type: string }[];
689  }) {
690    return {
691      id: a.id,
692      name: a.name,
693      description: a.description,
694      databaseId: a.databaseId,
695      status: a.status,
696      triggers: a.triggers.map((t) => t.type),
697      hasCondition: a.conditions.length > 0,
698      actions: a.actions.map((t) => t.type),
699      createdAt: a.createdAt,
700      updatedAt: a.updatedAt,
701      createdById: a.createdById
702    };
703  }
704 
705  private toDetail(a: AutomationRow) {
706    return {
707      ...this.toSummary(a),
708      triggers: this.sanitizeTriggers(a.id, a.triggers.map((t) => ({ type: t.type, config: (t.config ?? {}) as Record<string, unknown> }))),
709      condition: a.conditions[0]?.expression ?? null,
710      actions: a.actions.map((act) => ({
711        type: act.type,
712        config: act.config,
713        retries: act.retries,
714        timeoutMs: act.timeoutMs
715      }))
716    };
717  }
718 
719  private toRunSummary(
720    r: {
721      id: string;
722      status: string;
723      triggerEvent: unknown;
724      startedAt: Date | null;
725      finishedAt: Date | null;
726      retryCount: number;
727      error: unknown;
728      createdAt: Date;
729    },
730    _workspaceId: string
731  ) {
732    const te = (r.triggerEvent ?? {}) as Record<string, unknown>;
733    return {
734      id: r.id,
735      status: r.status,
736      triggerType: typeof te.type === "string" ? te.type : null,
737      startedAt: r.startedAt,
738      finishedAt: r.finishedAt,
739      retryCount: r.retryCount,
740      error: r.error ? truncateJson(r.error, 8 * 1024).value : null,
741      createdAt: r.createdAt
742    };
743  }
744 
745  /** Write optional inline sub-resources (create flow). */
746  private async writeSubResources(
747    tx: Prisma.TransactionClient,
748    automationId: string,
749    triggers?: PutTriggersInput["triggers"],
750    condition?: unknown,
751    actions?: PutActionsInput["actions"]
752  ): Promise<void> {
753    if (triggers && triggers.length) {
754      const normalized = triggers.map((t) => {
755        const config = { ...(t.config ?? {}) } as Record<string, unknown>;
756        if (t.type === "WEBHOOK" && !config.webhookId) {
757          const secret = generateWebhookSecret();
758          config.webhookId = uuid().replace(/-/g, "").slice(0, 24);
759          config.secretHash = webhookSecretHash(secret);
760          config.secretEncrypted = this.wrapSecret(secret);
761        }
762        return { type: t.type, config };
763      });
764      await tx.automationTrigger.createMany({
765        data: normalized.map((t, i) => ({ automationId, type: t.type, config: toJson(t.config), position: i }))
766      });
767    }
768    if (condition !== undefined && condition !== null) {
769      await tx.automationCondition.create({ data: { automationId, expression: toJson(condition), position: 0 } });
770    }
771    if (actions && actions.length) {
772      // Same normalization as the PUT (schema defaults persisted, not
773      // re-derived at read time).
774      const normalized = this.normalizeActions(actions);
775      await tx.automationAction.createMany({
776        data: normalized.map((act, i) => ({
777          automationId,
778          type: act.type,
779          config: toJson(act.config),
780          position: i,
781          retries: act.retries,
782          timeoutMs: act.timeoutMs
783        }))
784      });
785    }
786  }
787 
788  /** Exposed for the incoming-webhook controller (validation only — the
789   * controller owns the HTTP concerns). */
790  findActiveWebhook(webhookId: string) {
791    return this.prisma.automationTrigger.findFirst({
792      where: {
793        type: "WEBHOOK",
794        config: { path: ["webhookId"], equals: webhookId },
795        automation: { status: "ACTIVE", workspace: { deletedAt: null } }
796      },
797      include: {
798        automation: { select: { id: true, name: true, workspaceId: true, status: true, createdById: true } }
799      }
800    });
801  }
802}
803