matcher.ts · Source

matcher.tsapps/worker/src/processors/automations/matcher.tsraw ← automations7,932 B← Back to the app
1/**
2 * apps/worker — automation trigger matcher (TASK-0113 / TASK-0110 matcher).
3 *
4 * For each `factory:events` stream entry, find the ACTIVE automations whose
5 * trigger(s) of the matching type scope-match the event, dedupe on the
6 * event id, create a PENDING `automation_runs` row, and enqueue a
7 * `run-automation` job on the `automations` queue.
8 *
9 * Stream→trigger map (events.ts `EVENT_TO_TRIGGER`):
10 *   record.created  → RECORD_CREATED
11 *   record.updated  → RECORD_UPDATED | PROPERTY_CHANGED
12 *   page.created    → PAGE_CREATED
13 *   form.submitted  → FORM_SUBMITTED
14 * MANUAL / SCHEDULE / WEBHOOK are enqueued by other paths (API manual run,
15 * schedule reconciler, incoming webhook) — the matcher ignores them.
16 *
17 * Scope rules:
18 *  - the event's workspaceId must equal the automation's (when present),
19 *  - record/form triggers: the automation's databaseId (or the trigger's)
20 *    must equal the event's databaseId (when a database is set),
21 *  - record.updated: an OPTIONAL `config.properties` list narrows the match
22 *    to events whose `changedProperties` intersect it (a create event has
23 *    `changedProperties = ["*"]`, which matches any list);
24 *  - property.changed: `config.properties` is REQUIRED (the intersection
25 *    must be non-empty).
26 *
27 * Dedupe: `auto:dedupe:{automationId}:{eventId}` (shared `dedupeKey`),
28 * SET NX EX 600 — a redelivered stream entry (consumer-group retry) does not
29 * re-fire the same automation for the same event.
30 */
31import type { PrismaClient } from "@prisma/client";
32import type { Redis } from "ioredis";
33import { Queue } from "bullmq";
34import { dedupeKey, truncateJson } from "@factory/shared";
35import { asJson } from "../../lib/prisma.js";
36import { getLogger } from "../../logger.js";
37import { QUEUES } from "../../queues.js";
38 
39const log = getLogger("automation-matcher");
40 
41/** The run row's stored `input` cap (spec: 64 KB, truncated). */
42export const RUN_INPUT_MAX_BYTES = 64 * 1024;
43const DEDUPE_TTL_S = 600;
44 
45const DB_TRIGGER_TYPES = new Set(["RECORD_CREATED", "RECORD_UPDATED", "PROPERTY_CHANGED", "FORM_SUBMITTED"]);
46 
47/** Build the run's stored triggerEvent (the job carries the same object). */
48export function buildTriggerEvent(eventName: string, payload: Record<string, unknown>): Record<string, unknown> {
49  // Flatten the event payload to the top level (recordId / databaseId /
50  // changedProperties / workspaceId become first-class) AND keep `payload`
51  // intact (so `event.payload.<x>` interpolation + condition refs work).
52  return {
53    type: eventName,
54    payload,
55    ...payload
56  };
57}
58 
59/** Does this trigger scope-match the event? */
60export function triggerScopeMatches(
61  trigger: { type: string; config: unknown },
62  automation: { workspaceId: string; databaseId: string | null },
63  payload: Record<string, unknown>
64): boolean {
65  const p = payload as Record<string, unknown>;
66  if (typeof p.workspaceId === "string" && p.workspaceId !== automation.workspaceId) return false;
67 
68  if (DB_TRIGGER_TYPES.has(trigger.type)) {
69    const dbId = automation.databaseId ?? ((trigger.config as Record<string, unknown> | null)?.databaseId as string | undefined);
70    if (dbId && p.databaseId !== dbId) return false;
71  }
72 
73  if (trigger.type === "RECORD_UPDATED" || trigger.type === "PROPERTY_CHANGED") {
74    const want: string[] = Array.isArray((trigger.config as Record<string, unknown> | null)?.properties)
75      ? ((trigger.config as Record<string, unknown>).properties as string[])
76      : [];
77    if (want.length > 0) {
78      // Fail closed: an event WITHOUT changedProperties never fires a
79      // property-scoped trigger (spec: "an event lacking changedProperties
80      // never fires PROPERTY_CHANGED"). A create event carries
81      // changedProperties = ["*"], which matches any list.
82      const changedRaw = p.changedProperties;
83      if (!Array.isArray(changedRaw)) return false;
84      const changed = changedRaw as string[];
85      const hit = changed.includes("*") || want.some((w) => changed.includes(w));
86      if (!hit) return false;
87    }
88  }
89  return true;
90}
91 
92export interface MatcherEntry {
93  /** The stream entry id (the dedupe eventId). */
94  id: string;
95  event: string;
96  payload: Record<string, unknown>;
97}
98 
99export interface MatcherDeps {
100  prisma: PrismaClient;
101  redis: Redis;
102}
103 
104/**
105 * Match one stream entry against all ACTIVE automations and enqueue the
106 * resulting runs. Returns the number of runs enqueued. Never throws for a
107 * single-automation failure (logged + the entry is still ACKed by the
108 * consumer so one bad row cannot wedge the group).
109 *
110 * §7 observability: every decision bumps a Redis counter —
111 * `automations:matched:{type}` and `automations:skipped:{reason}`
112 * (filter_mismatch / deduped).
113 */
114export async function matchAndEnqueue(
115  deps: MatcherDeps,
116  triggerType: string,
117  entry: MatcherEntry
118): Promise<number> {
119  const automations = await deps.prisma.automation.findMany({
120    where: { status: "ACTIVE" },
121    include: { triggers: { where: { type: triggerType as never } } }
122  });
123 
124  const bump = (key: string) => {
125    void deps.redis.incr(key).catch(() => undefined);
126  };
127 
128  let enqueued = 0;
129  for (const a of automations) {
130    const matching = a.triggers.filter((t) => triggerScopeMatches(t, a, entry.payload));
131    if (matching.length === 0) {
132      bump("automations:skipped:filter_mismatch");
133      continue;
134    }
135 
136    // Dedupe on (automationId, eventId): a redelivery must not re-fire.
137    try {
138      const got = await deps.redis.set(dedupeKey(a.id, entry.id), "1", "EX", DEDUPE_TTL_S, "NX");
139      if (got === null) {
140        bump("automations:skipped:deduped");
141        continue;
142      }
143    } catch (err) {
144      log.warn({ err: (err as Error).message, automationId: a.id }, "dedupe check failed (proceeding)");
145    }
146 
147    try {
148      // Load the automation's condition tree + action count for the run's
149      // stored `input` (spec: { triggerEvent (truncated), conditions,
150      // actionCount }).
151      const full = await deps.prisma.automation.findUnique({
152        where: { id: a.id },
153        select: {
154          conditions: { select: { expression: true } },
155          _count: { select: { actions: true } }
156        }
157      });
158      const triggerEvent = buildTriggerEvent(entry.event, entry.payload);
159      const { value: truncatedTrigger, truncated } = truncateJson(triggerEvent, RUN_INPUT_MAX_BYTES);
160      const inputObj: Record<string, unknown> = {
161        triggerEvent: truncated ? { ...(truncatedTrigger as Record<string, unknown>), truncated: true } : truncatedTrigger,
162        conditions: full?.conditions.length ? full.conditions[0]?.expression ?? null : null,
163        actionCount: full?._count.actions ?? 0
164      };
165      const run = await deps.prisma.automationRun.create({
166        data: {
167          workspaceId: a.workspaceId,
168          automationId: a.id,
169          status: "PENDING",
170          triggerEvent: asJson(triggerEvent),
171          input: asJson(inputObj),
172          retryCount: 0
173        }
174      });
175      await enqueueRunJob(deps.redis, {
176        runId: run.id,
177        automationId: a.id,
178        workspaceId: a.workspaceId,
179        triggerEvent,
180        eventId: entry.id
181      });
182      enqueued++;
183      bump(`automations:matched:${triggerType.toLowerCase()}`);
184    } catch (err) {
185      log.error({ err: (err as Error).message, automationId: a.id, entryId: entry.id }, "matcher enqueue failed for automation");
186    }
187  }
188  return enqueued;
189}
190 
191/** Enqueue a `run-automation` job (best-effort; the watchdog reclaims orphans). */
192export async function enqueueRunJob(
193  redis: Redis,
194  job: { runId: string; automationId: string; workspaceId: string; triggerEvent: Record<string, unknown>; eventId: string }
195): Promise<void> {
196  const q = new Queue(QUEUES.automations, { connection: redis });
197  try {
198    await q.add("run-automation", job, { removeOnComplete: 1000, removeOnFail: 500 });
199  } finally {
200    await q.close();
201  }
202}
203