events.ts · Source

events.tsapps/worker/src/events.tsraw ← src6,430 B← Back to the app
1/**
2 * apps/worker — domain event catalog + envelope.
3 *
4 * The API publishes domain events onto the Redis stream `factory:events`
5 * (ARCHITECTURE.md §3/§9). Every entry is a JSON envelope:
6 *
7 *   { event: "page.created", payload: { ... }, ts: "2025-01-01T00:00:00.000Z" }
8 *
9 * This module is the single source of truth inside the worker for:
10 *  - the event names (EVENTS catalog),
11 *  - the zod envelope schema (`parseEvent`),
12 *  - per-consumer subscriptions (SEARCH_INDEX_EVENTS / WEBHOOK_EVENT_CATALOG /
13 *    AUTOMATION_EVENT_CATALOG) and the trigger mapping used by the
14 *    automation stream consumer.
15 *
16 * Note: the worker declares the catalog locally (rather than importing
17 * @factory/shared) so the app stays self-contained while shared keeps
18 * evolving; keep both in sync when adding event names.
19 */
20import { z } from "zod";
21import { getLogger } from "./logger.js";
22 
23const log = getLogger("events");
24 
25/** All domain events the platform can emit (ARCHITECTURE.md §9). */
26export const EVENTS = [
27  "page.created",
28  "page.updated",
29  "page.deleted",
30  "page.shared",
31  "database.created",
32  "database.updated",
33  "record.created",
34  "record.updated",
35  "comment.created",
36  "comment.deleted",
37  "notification.created",
38  "form.submitted",
39  "automation.executed",
40  "automation.manual",
41  "file.uploaded",
42  // TASK-0096: worker publishes this at S3 finalization (File row mime/size
43  // are set after `file.uploaded` fires at presign) → the search consumer
44  // re-upserts the file doc with the final metadata.
45  "file.finalized",
46  // TASK-0102: the files pipeline flagged the object infected (ClamAV).
47  // Stream-internal: the worker notifies workspace admins; NO search index
48  // update and NO `file.uploaded` is ever emitted for this file.
49  "file.virus_detected",
50  // TASK-0104 §3: a file was trashed (soft delete). The search index removes
51  // the doc; the object + version rows are retained until purge.
52  "file.deleted",
53  // TASK-0104 §8: a file's searchable fields changed (rename). The search
54  // index re-upserts the file doc so the title converges immediately.
55  "file.updated",
56  "api_key.created",
57  "user.login",
58  "user.logout",
59  "user.register",
60  "permission.changed",
61  "workspace.created"
62] as const;
63 
64export type DomainEventName = (typeof EVENTS)[number];
65 
66/**
67 * Stream envelope. `payload` is validated here only as a plain JSON object;
68 * each consumer zod-parses its own payload shape (defense in depth).
69 */
70export const eventEnvelopeSchema = z.object({
71  event: z.string().min(1),
72  payload: z.record(z.string(), z.unknown()).default({}),
73  ts: z.string().min(1)
74});
75 
76export type EventEnvelope = z.infer<typeof eventEnvelopeSchema>;
77 
78/**
79 * Parse one raw stream entry. Throws (zod) on a malformed envelope.
80 */
81export function parseEvent(raw: {
82  id: string;
83  fields: Record<string, string | undefined>;
84}): { id: string; envelope: EventEnvelope } {
85  let envelope: EventEnvelope;
86  try {
87    const event = raw.fields["event"];
88    const payload = raw.fields["payload"];
89    const ts = raw.fields["ts"];
90    envelope = eventEnvelopeSchema.parse({
91      event,
92      payload: payload ? JSON.parse(payload) : {},
93      ts: ts ?? new Date().toISOString()
94    });
95  } catch (err) {
96    throw new Error(`poison stream entry ${raw.id}: ${(err as Error).message}`);
97  }
98  return { id: raw.id, envelope };
99}
100 
101/**
102 * Parse-or-drop: returns the envelope, or null when the entry is malformed
103 * (logged). Consumers return early on null so the entry is XACKed — a poison
104 * entry must not be redelivered forever.
105 */
106export function tryParseEvent(raw: {
107  id: string;
108  fields: Record<string, string | undefined>;
109}): { id: string; envelope: EventEnvelope } | null {
110  try {
111    return parseEvent(raw);
112  } catch (err) {
113    log.error({ err: (err as Error).message }, "dropping poison stream entry");
114    return null;
115  }
116}
117 
118// ---------------------------------------------------------------------------
119// Per-consumer subscriptions (XREADGROUP consumer groups are per-queue, so a
120// consumer only needs the events its queue acts on — everything else is
121// XACKed as not-mine).
122// ---------------------------------------------------------------------------
123 
124/** search-index consumer: content + permission + user/workspace events.
125 * TASK-0096: user.register / user.login (with a workspaceId) upsert a
126 * per-workspace user doc; workspace.created triggers the initial index. */
127export const SEARCH_INDEX_EVENTS: readonly DomainEventName[] = [
128  "page.created",
129  "page.updated",
130  "page.deleted",
131  "page.shared",
132  "database.created",
133  "database.updated",
134  "record.created",
135  "record.updated",
136  "comment.created",
137  "comment.deleted",
138  "file.uploaded",
139  "file.finalized",
140  "file.deleted",
141  "file.updated",
142  "permission.changed",
143  "user.register",
144  "user.login",
145  "workspace.created"
146];
147 
148/**
149 * webhooks consumer: what owners can subscribe their endpoints to.
150 *
151 * TASK-0130: the SINGLE SOURCE is `@factory/shared` `WEBHOOK_EVENT_NAMES`
152 * (packages/shared/src/events/catalog.ts) — the 13-event public catalog. The
153 * old local list (17 names, incl. user.login/user.register/permission.changed/
154 * workspace.created) is gone: those platform-internal events were never
155 * offered to third-party subscribers (docs §3 lists exactly 13).
156 */
157import { WEBHOOK_EVENT_NAMES } from "@factory/shared";
158export const WEBHOOK_EVENT_CATALOG: readonly DomainEventName[] = WEBHOOK_EVENT_NAMES as readonly DomainEventName[];
159 
160/**
161 * automations consumer: events that can start an automation run.
162 *
163 * NOTE: `comment.created` is intentionally NOT triggerable — the Prisma
164 * AutomationTriggerType enum has no COMMENT_CREATED value, and the worker
165 * must not fabricate trigger types the schema doesn't support. The stream
166 * consumer XACKs comment events as "not automatable" until the schema adds
167 * the value (per AGENTS.md: schema wins over the brief; conflict recorded).
168 */
169export const AUTOMATION_EVENT_CATALOG: readonly DomainEventName[] = [
170  "record.created",
171  "record.updated",
172  "page.created",
173  "form.submitted",
174  "automation.manual"
175];
176 
177/** Domain event → AutomationTriggerType (for stream→trigger matching). */
178export const EVENT_TO_TRIGGER: Record<string, string> = {
179  "record.created": "RECORD_CREATED",
180  "record.updated": "RECORD_UPDATED",
181  "page.created": "PAGE_CREATED",
182  "form.submitted": "FORM_SUBMITTED",
183  "automation.manual": "MANUAL"
184};
185