1
15import type { Redis } from "ioredis";
16import {
17 AUTOMATION_MAX_DEPTH,
18 interpolate,
19 nextAutomationContext,
20 normalizeConditionValue,
21 type InterpolationVars
22} from "@factory/shared";
23import { unwrapEnvelope } from "../condition-evaluator.js";
24import { getLogger } from "../../../logger.js";
25import { FatalError } from "./classify.js";
26import type { AutomationRunContext } from "./types.js";
27
28const log = getLogger("automation-exec");
29
30export interface PropertyMetaLite {
31 id: string;
32 name: string;
33 type: string;
34 isTitle: boolean;
35 config: Record<string, unknown> | null;
36}
37
38
39export async function loadProperties(
40 prisma: { databaseProperty: any },
41 databaseId: string
42): Promise<PropertyMetaLite[]> {
43 return prisma.databaseProperty.findMany({
44 where: { databaseId, deletedAt: null },
45 select: { id: true, name: true, type: true, isTitle: true, config: true }
46 });
47}
48
49
53export function resolveRecordId(cfg: Record<string, unknown>, ctx: AutomationRunContext): string {
54 const v = cfg.recordId;
55 if (typeof v === "string" && v !== "from_trigger") return v;
56 if (ctx.recordId) return ctx.recordId;
57 throw new FatalError("NO_TRIGGER_RECORD", "action targets the trigger record but the event has no record");
58}
59
60
66export function resolveUserId(
67 target: unknown,
68 ctx: AutomationRunContext,
69 payloadUserIds: string[] = []
70): string {
71 if (typeof target === "string" && target !== "from_trigger" && target !== "trigger_user" && target !== "automation_creator") {
72 return target;
73 }
74 if (target === "automation_creator") return ctx.actorUserId;
75
76 const p = ctx.triggerEvent as Record<string, unknown>;
77 for (const k of ["createdBy", "actorId", "submittedBy", "userId"]) {
78 if (typeof p[k] === "string" && p[k]) return p[k] as string;
79 }
80 if (payloadUserIds.length > 0) return payloadUserIds[0]!;
81 throw new FatalError("NO_TARGET_USER", `cannot resolve user target "${String(target)}" from the event`);
82}
83
84
91export async function emitDomainEvent(
92 redis: Redis,
93 event: string,
94 payload: Record<string, unknown>,
95 ctx: AutomationRunContext,
96 extra?: Record<string, unknown>
97): Promise<boolean> {
98 if (ctx.depth >= AUTOMATION_MAX_DEPTH) {
99 log.warn({ runId: ctx.runId, event, depth: ctx.depth }, "depth guard: event not emitted");
100 return false;
101 }
102 const ac = nextAutomationContext(ctx.runId, ctx.automationId, ctx.depth);
103 await redis.xadd(
104 "factory:events",
105 "*",
106 "event",
107 event,
108 "payload",
109 JSON.stringify({ ...payload, ...extra, automationContext: ac }),
110 "ts",
111 new Date().toISOString()
112 );
113 return true;
114}
115
116
117export async function auditSystem(
118 prisma: { auditLog: any },
119 ctx: AutomationRunContext,
120 entry: { action: string; resourceType: string; resourceId: string; metadata?: Record<string, unknown> }
121): Promise<void> {
122 try {
123 await prisma.auditLog.create({
124 data: {
125 actorType: "SYSTEM",
126 actorId: `automation:${ctx.automationId}`,
127 workspaceId: ctx.workspaceId,
128 resourceType: entry.resourceType,
129 resourceId: entry.resourceId,
130 action: entry.action,
131 metadata: { ...(entry.metadata ?? {}), onBehalfOf: "automation", automationId: ctx.automationId, runId: ctx.runId }
132 }
133 });
134 } catch (err) {
135 log.warn({ err: (err as Error).message, runId: ctx.runId }, "automation audit write failed");
136 }
137}
138
139
143export async function runWithTimeout<T>(ms: number, task: () => Promise<T>): Promise<T> {
144 let timer: ReturnType<typeof setTimeout> | undefined;
145 try {
146 return await Promise.race([
147 task(),
148 new Promise<never>((_, reject) => {
149 timer = setTimeout(() => reject(new FatalError("ACTION_TIMEOUT", `action timed out after ${ms} ms`)), ms);
150 timer.unref?.();
151 })
152 ]);
153 } finally {
154 if (timer) clearTimeout(timer);
155 }
156}
157
158
163export function interpolateConfig(
164 raw: string,
165 vars: InterpolationVars,
166 max: number,
167 field: string,
168 warnings: string[]
169): string {
170 const { value, warnings: w } = interpolate(raw, vars);
171 warnings.push(...w);
172 if (Buffer.byteLength(value, "utf8") > max) {
173 throw new FatalError("INTERPOLATION_OVERFLOW", `${field} exceeds ${max} bytes after interpolation`);
174 }
175 return value;
176}
177
178
179export function interpolateDeep(
180 value: unknown,
181 vars: InterpolationVars,
182 warnings: string[],
183 path: string = ""
184): unknown {
185 if (typeof value === "string") return interpolateConfig(value, vars, 64 * 1024, path || "(root)", warnings);
186 if (Array.isArray(value)) return value.map((v, i) => interpolateDeep(v, vars, warnings, `${path}[${i}]`));
187 if (value && typeof value === "object") {
188 const out: Record<string, unknown> = {};
189 for (const [k, v] of Object.entries(value as Record<string, unknown>)) {
190 out[k] = interpolateDeep(v, vars, warnings, path ? `${path}.${k}` : k);
191 }
192 return out;
193 }
194 return value;
195}
196
197
203export async function buildVars(
204 prisma: { databaseRecord: any; databaseProperty: any },
205 ctx: AutomationRunContext
206): Promise<InterpolationVars> {
207 const vars: InterpolationVars = {
208 record: null,
209 event: ctx.triggerEvent,
210 input: ctx.input,
211 automation: { id: ctx.automationId, name: ctx.automationName },
212 run: { id: ctx.runId }
213 };
214 if (ctx.recordId) {
215 const row = await prisma.databaseRecord.findUnique({
216 where: { id: ctx.recordId },
217 select: { id: true, properties: { select: { value: true, property: { select: { name: true, type: true, config: true } } } } }
218 });
219 if (row) {
220 const flat: Record<string, unknown> = { id: row.id };
221 for (const rp of row.properties) {
222 flat[rp.property.name] = normalizeConditionValue(
223 rp.property.type as string,
224 rp.property.config as Record<string, unknown> | null,
225 unwrapEnvelope(rp.property.type as string, rp.value)
226 );
227 }
228 vars.record = flat;
229 }
230 }
231 return vars;
232}
233
