registry.ts · Source

registry.tsapps/worker/src/processors/automations/executors/registry.tsraw ← executors4,623 B← Back to the app
1/**
2 * apps/worker — automation executor registry (TASK-0112 §1/§3).
3 *
4 * `executorFor(type)` returns the executor for an action type. `validateAction`
5 * is the single runtime re-validation entry: it runs the SHARED
6 * `validateActionDto` (the same validator the API used at save) so a corrupt
7 * or drifted stored row fails FAST with a fatal INVALID_CONFIG error (no
8 * retry) instead of a cryptic executor crash. The processor (TASK-0113) calls
9 * `validateAction` then `executorFor(...).execute(...)`.
10 */
11import { validateActionDto, AUTOMATION_ACTION_TYPES, type AutomationActionType } from "@factory/shared";
12import { FatalError } from "./classify.js";
13import type { ActionExecutor, ActionResult, AutomationRunContext, ExecutorDeps } from "./types.js";
14 
15import updateRecord from "./update-record.js";
16import createPage from "./create-page.js";
17import sendEmail from "./send-email.js";
18import sendWebhook from "./send-webhook.js";
19import createNotification from "./create-notification.js";
20import assignUser from "./assign-user.js";
21import changeStatus from "./change-status.js";
22import callIntegration from "./call-integration.js";
23import runAiAction from "./run-ai-action.js";
24 
25const registry = new Map<string, ActionExecutor>(
26  [
27    updateRecord,
28    createPage,
29    sendEmail,
30    sendWebhook,
31    createNotification,
32    assignUser,
33    changeStatus,
34    callIntegration,
35    runAiAction
36  ].map((e) => [e.type, e])
37);
38 
39/**
40 * Test-injection seam (TASK-0113 integration): an override consulted BEFORE
41 * the registry by the pipeline's per-action entry point. Empty in production
42 * (behavior is byte-identical); the pipeline integration spec registers a
43 * failing executor here to drive the retry → dead-letter path hermetically
44 * (a real SEND_WEBHOOK failure requires DNS + network, which is not
45 * deterministic in a test container). Register for a real type — the action
46 * row's config still passes the shared schema validation first.
47 */
48const testOverrides = new Map<string, ActionExecutor>();
49 
50export function registerTestExecutor(type: string, executor: ActionExecutor): void {
51  testOverrides.set(type, executor);
52}
53 
54export function clearTestExecutors(): void {
55  testOverrides.clear();
56}
57 
58/** The executor for an action type (undefined for unknown types). */
59export function executorFor(type: string): ActionExecutor | undefined {
60  return testOverrides.get(type) ?? registry.get(type);
61}
62 
63export interface ValidatedAction {
64  type: string;
65  config: Record<string, unknown>;
66  retries: number;
67  timeoutMs: number;
68}
69 
70/**
71 * Runtime re-validation of a stored action. Throws FatalError INVALID_CONFIG
72 * when the type is unknown or the config fails the shared schema. Returns the
73 * normalized {type, config, retries, timeoutMs}.
74 */
75export function validateAction(action: { type: unknown; config: unknown; retries?: number; timeoutMs?: number }): ValidatedAction {
76  const type = action.type as string;
77  if (!type || !(AUTOMATION_ACTION_TYPES as readonly string[]).includes(type)) {
78    throw new FatalError("INVALID_CONFIG", `unknown action type: ${String(type)}`);
79  }
80  // exactOptionalPropertyTypes: only set retries/timeoutMs when present.
81  const dto: { type: AutomationActionType; config: Record<string, unknown>; retries?: number; timeoutMs?: number } = {
82    type: type as AutomationActionType,
83    config: (action.config ?? {}) as Record<string, unknown>
84  };
85  if (action.retries !== undefined) dto.retries = action.retries;
86  if (action.timeoutMs !== undefined) dto.timeoutMs = action.timeoutMs;
87  const res = validateActionDto(dto);
88  if (!res.ok) {
89    const first = res.errors[0];
90    throw new FatalError("INVALID_CONFIG", `invalid action config (${first?.path ?? "?"}): ${first?.message ?? "unknown"}`);
91  }
92  return {
93    type: res.normalized.type,
94    config: res.normalized.config,
95    retries: res.normalized.retries,
96    timeoutMs: res.normalized.timeoutMs
97  };
98}
99 
100/**
101 * Validate + execute one action (the processor's per-action entry point).
102 * Re-validates the stored row, dispatches to the executor. The processor
103 * wraps the returned result (or thrown error) per the retry/fatal contract.
104 */
105export async function executeAction(
106  action: { type: unknown; config: unknown; retries?: number; timeoutMs?: number },
107  ctx: AutomationRunContext,
108  deps: ExecutorDeps,
109  index: number
110): Promise<ActionResult> {
111  const validated = validateAction(action);
112  const executor = registry.get(validated.type);
113  if (!executor) throw new FatalError("INVALID_CONFIG", `no executor for ${validated.type}`);
114  return executor.execute(validated.config, ctx, deps, index);
115}
116