nsrb
fd0865063b
- Added a new module for parsing and resolving configuration references, supporting `$SECRET_`, `$KV_`, and `$CONTEXT_` prefixes. - Implemented functions to handle the resolution of these references within workflow contexts, enhancing the flexibility of configuration management. - Updated existing workflows and scripts to utilize the new config reference resolution, improving data handling and context management. - Introduced a smoke test to validate the functionality of the config reference system, ensuring robustness and reliability. This update significantly enhances the ability to manage dynamic configurations within workflows, improving overall usability and maintainability.
760 lines
22 KiB
JavaScript
760 lines
22 KiB
JavaScript
import fs from "fs";
|
|
import path from "path";
|
|
import yaml from "yaml";
|
|
import cron from "node-cron";
|
|
import { WORKFLOWS_DIR } from "./paths.js";
|
|
import { log } from "./logger.js";
|
|
import * as store from "./store.js";
|
|
import { clearScriptCache, runScript } from "./script-sandbox.js";
|
|
import {
|
|
SET_STEP_SCRIPT,
|
|
compileWorkflowScripts,
|
|
evaluateJsonata,
|
|
isJsonataTruthy,
|
|
mergeStepData,
|
|
namespacedPath,
|
|
parseScriptStep,
|
|
} from "./workflow-parse.js";
|
|
import {
|
|
chainCtx,
|
|
mergeContextWave,
|
|
normalizeContext,
|
|
normalizeStepResult,
|
|
storedEnvelope,
|
|
} from "./step-result.js";
|
|
import * as fsStore from "./fs-store.js";
|
|
import {
|
|
checkHttpAuth,
|
|
resolveAuthMechanism,
|
|
resolveUnauthorizedSpec,
|
|
sendHttpPageOrJson,
|
|
sendSuccessPage,
|
|
} from "./http-trigger-auth.js";
|
|
import { resolveConfigRefs } from "./config-refs.js";
|
|
|
|
/**
|
|
* @typedef {{ owner: string, file: string, workflow: any }} WorkflowEntry
|
|
* @typedef {{ key: string, owner: string, trigger: any }} HttpRouteEntry
|
|
*/
|
|
|
|
const MAX_WORKFLOW_TRIGGER_DEPTH = 8;
|
|
const HTTP_METHODS = ["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE"];
|
|
|
|
/**
|
|
* @param {unknown} workflow
|
|
*/
|
|
function hasWorkflowTrigger(workflow) {
|
|
if (!workflow || typeof workflow !== "object") return false;
|
|
const triggers = /** @type {{ triggers?: Array<{ type?: string }> }} */ (workflow)
|
|
.triggers;
|
|
return (triggers ?? []).some((t) => t?.type === "workflow");
|
|
}
|
|
|
|
/**
|
|
* @param {import("fastify").FastifyInstance} server
|
|
*/
|
|
export function createRegistry(server) {
|
|
/** @type {Map<string, WorkflowEntry>} */
|
|
const workflows = new Map();
|
|
/** @type {Map<string, string>} */
|
|
const loadErrors = new Map();
|
|
/** @type {import("node-cron").ScheduledTask[]} */
|
|
const cronTasks = [];
|
|
/** @type {import("node-cron").ScheduledTask | null} */
|
|
let pruneTask = null;
|
|
/** @type {Map<string, HttpRouteEntry>} */
|
|
const httpRoutes = new Map();
|
|
let httpDispatcherRegistered = false;
|
|
|
|
/**
|
|
* Resolve a same-owner workflow that opts in with `type: workflow`.
|
|
* Prefers YAML `name`, then filename (`name` or `name.yaml`).
|
|
*
|
|
* @param {string} owner
|
|
* @param {string} name
|
|
*/
|
|
function resolveWorkflowTriggerKey(owner, name) {
|
|
if (typeof name !== "string" || name.length === 0) {
|
|
throw new Error("workflow name is required");
|
|
}
|
|
|
|
/** @type {string[]} */
|
|
const byName = [];
|
|
for (const [key, entry] of workflows) {
|
|
if (entry.owner !== owner) continue;
|
|
if (entry.workflow?.enabled === false) continue;
|
|
if (!hasWorkflowTrigger(entry.workflow)) continue;
|
|
if (entry.workflow?.name === name) byName.push(key);
|
|
}
|
|
if (byName.length === 1) return byName[0];
|
|
if (byName.length > 1) {
|
|
throw new Error(`ambiguous workflow name "${name}"`);
|
|
}
|
|
|
|
const fileCandidates =
|
|
name.endsWith(".yaml") || name.endsWith(".yml")
|
|
? [`${owner}/${name}`]
|
|
: [`${owner}/${name}`, `${owner}/${name}.yaml`];
|
|
|
|
for (const key of fileCandidates) {
|
|
const entry = workflows.get(key);
|
|
if (!entry) continue;
|
|
if (entry.workflow?.enabled === false) continue;
|
|
if (!hasWorkflowTrigger(entry.workflow)) continue;
|
|
return key;
|
|
}
|
|
|
|
throw new Error(
|
|
`workflow "${name}" not found or has no workflow trigger (owner "${owner}")`,
|
|
);
|
|
}
|
|
|
|
function registerWorkflows() {
|
|
workflows.clear();
|
|
loadErrors.clear();
|
|
clearScriptCache();
|
|
|
|
if (!fs.existsSync(WORKFLOWS_DIR)) {
|
|
log.warn("workflows directory missing");
|
|
return;
|
|
}
|
|
|
|
const owners = fsStore.listOwners();
|
|
|
|
for (const owner of owners) {
|
|
const registersPath = path.join(WORKFLOWS_DIR, owner, "registers.yaml");
|
|
if (!fs.existsSync(registersPath)) {
|
|
log.warn(`Skipping owner "${owner}": no registers.yaml`);
|
|
continue;
|
|
}
|
|
|
|
let workflowFiles = [];
|
|
try {
|
|
workflowFiles = fsStore.readRegisters(owner);
|
|
} catch (err) {
|
|
log.error({ err, owner }, "failed to parse registers.yaml");
|
|
continue;
|
|
}
|
|
|
|
const onDisk = fsStore.listOwnerYamlFiles(owner);
|
|
for (const file of onDisk) {
|
|
if (workflowFiles.includes(file)) continue;
|
|
if (file.startsWith("dev-")) {
|
|
workflowFiles.push(file);
|
|
continue;
|
|
}
|
|
log.warn(`Workflow file not in registers.yaml: ${owner}/${file}`);
|
|
}
|
|
|
|
for (const file of workflowFiles) {
|
|
const key = `${owner}/${file}`;
|
|
const filePath = path.join(WORKFLOWS_DIR, owner, file);
|
|
try {
|
|
const workflowData = fs.readFileSync(filePath, "utf8");
|
|
const workflow = yaml.parse(workflowData);
|
|
compileWorkflowScripts(workflow?.scripts);
|
|
workflows.set(key, { owner, file, workflow });
|
|
} catch (err) {
|
|
const message = err instanceof Error ? err.message : String(err);
|
|
loadErrors.set(key, message);
|
|
log.error({ err, workflow: key }, "failed to load workflow; skipping");
|
|
}
|
|
}
|
|
}
|
|
|
|
log.debug({ count: workflows.size }, "workflows loaded");
|
|
}
|
|
|
|
/**
|
|
* Rebuild in-memory METHOD+path → workflow map. Registers a single /u/*
|
|
* Fastify route once so path/method changes apply on reregister without restart.
|
|
*/
|
|
function registerHttpTriggers() {
|
|
httpRoutes.clear();
|
|
|
|
for (const [key, { owner, workflow }] of workflows) {
|
|
if (workflow.enabled === false) {
|
|
log.debug(`Skipping disabled workflow HTTP triggers (${key})`);
|
|
continue;
|
|
}
|
|
|
|
for (const trigger of workflow.triggers ?? []) {
|
|
if (trigger.type !== "HTTP") continue;
|
|
|
|
const method = String(trigger.method ?? "POST").toUpperCase();
|
|
const url = namespacedPath(owner, trigger.path);
|
|
const routeKey = `${method} ${url}`;
|
|
|
|
if (httpRoutes.has(routeKey)) {
|
|
log.warn(`Skipping duplicate HTTP trigger ${routeKey} (${key})`);
|
|
continue;
|
|
}
|
|
httpRoutes.set(routeKey, { key, owner, trigger });
|
|
log.debug(`Mapped HTTP trigger ${routeKey} (${key})`);
|
|
}
|
|
}
|
|
|
|
if (!httpDispatcherRegistered) {
|
|
httpDispatcherRegistered = true;
|
|
server.route({
|
|
method: HTTP_METHODS,
|
|
url: "/u/*",
|
|
handler: dispatchHttpTrigger,
|
|
});
|
|
log.debug("Registered HTTP trigger wildcard dispatcher /u/*");
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @param {import("fastify").FastifyRequest} req
|
|
* @param {import("fastify").FastifyReply} reply
|
|
*/
|
|
async function dispatchHttpTrigger(req, reply) {
|
|
const wildcard = /** @type {{ "*": string }} */ (req.params)["*"] ?? "";
|
|
const url = `/u/${String(wildcard).replace(/^\/+/, "")}`;
|
|
const method = String(req.method ?? "GET").toUpperCase();
|
|
const routeKey = `${method} ${url}`;
|
|
const mapped = httpRoutes.get(routeKey);
|
|
|
|
if (!mapped) {
|
|
return reply.code(404).send({ error: "not found" });
|
|
}
|
|
|
|
const entry = workflows.get(mapped.key);
|
|
if (!entry || entry.workflow?.enabled === false) {
|
|
return reply.code(404).send({ error: "workflow disabled" });
|
|
}
|
|
|
|
// Prefer live trigger from current workflow YAML (auth/response edits)
|
|
const liveTrigger =
|
|
(entry.workflow.triggers ?? []).find((t) => {
|
|
if (t?.type !== "HTTP") return false;
|
|
const m = String(t.method ?? "POST").toUpperCase();
|
|
const p = namespacedPath(entry.owner, t.path);
|
|
return m === method && p === url;
|
|
}) ?? mapped.trigger;
|
|
|
|
if (liveTrigger.auth != null && liveTrigger.auth !== false) {
|
|
const mechanism = await resolveAuthMechanism(liveTrigger.auth);
|
|
if (!mechanism) {
|
|
const { status, pageName } = resolveUnauthorizedSpec(liveTrigger, null);
|
|
return sendHttpPageOrJson(reply, status, pageName, {
|
|
error: "unauthorized",
|
|
});
|
|
}
|
|
const ok = await checkHttpAuth(req, mechanism, {
|
|
owner: entry.owner,
|
|
workflowKey: mapped.key,
|
|
});
|
|
if (!ok) {
|
|
const { status, pageName } = resolveUnauthorizedSpec(
|
|
liveTrigger,
|
|
mechanism,
|
|
);
|
|
return sendHttpPageOrJson(reply, status, pageName, {
|
|
error: "unauthorized",
|
|
});
|
|
}
|
|
}
|
|
|
|
const result = await runWorkflow(
|
|
mapped.key,
|
|
{ data: req.body },
|
|
{ type: "http", detail: `${method} ${url}` },
|
|
);
|
|
if (result.status === "failed") {
|
|
return reply.code(500).send({
|
|
runId: result.runId,
|
|
error: result.error,
|
|
});
|
|
}
|
|
|
|
const defaultBody = {
|
|
runId: result.runId,
|
|
result: result.result,
|
|
};
|
|
if (typeof liveTrigger.response === "string" && liveTrigger.response) {
|
|
return sendSuccessPage(reply, liveTrigger.response, defaultBody);
|
|
}
|
|
return reply.send(defaultBody);
|
|
}
|
|
|
|
function registerCronTriggers() {
|
|
for (const task of cronTasks) {
|
|
task.destroy();
|
|
}
|
|
cronTasks.length = 0;
|
|
|
|
for (const [key, { workflow }] of workflows) {
|
|
if (workflow.enabled === false) {
|
|
log.debug(`Skipping disabled workflow cron triggers (${key})`);
|
|
continue;
|
|
}
|
|
|
|
for (const trigger of workflow.triggers ?? []) {
|
|
if (trigger.type !== "cron") continue;
|
|
|
|
const schedule = trigger.schedule;
|
|
if (!schedule || !cron.validate(schedule)) {
|
|
log.warn(`Skipping invalid cron schedule "${schedule}" (${key})`);
|
|
continue;
|
|
}
|
|
|
|
const task = cron.schedule(
|
|
schedule,
|
|
() => {
|
|
log.debug(`cron firing ${key} (${schedule})`);
|
|
return runWorkflow(
|
|
key,
|
|
{ data: workflow.data ?? null },
|
|
{ type: "cron", detail: schedule },
|
|
);
|
|
},
|
|
{ name: `${key}:${schedule}`, noOverlap: true },
|
|
);
|
|
cronTasks.push(task);
|
|
log.debug(`Registered cron trigger ${schedule} (${key})`);
|
|
}
|
|
}
|
|
}
|
|
|
|
function registerPruneJob() {
|
|
if (pruneTask) {
|
|
pruneTask.destroy();
|
|
pruneTask = null;
|
|
}
|
|
const days = Number(process.env.JERAPAH_FLOW_RETENTION_DAYS ?? process.env.SCRUNNER_RETENTION_DAYS ?? 30);
|
|
pruneTask = cron.schedule(
|
|
"0 0 * * *",
|
|
async () => {
|
|
try {
|
|
const deleted = await store.pruneOlderThan(days);
|
|
log.info({ deleted, days }, "pruned old workflow runs");
|
|
} catch (err) {
|
|
log.error({ err }, "failed to prune old workflow runs");
|
|
}
|
|
},
|
|
{ name: "prune-runs" },
|
|
);
|
|
}
|
|
|
|
/**
|
|
* @param {import("./workflow-parse.js").CompiledStep} parsed
|
|
* @param {number} index
|
|
* @param {import("./step-result.js").StepResult} last
|
|
* @param {string} runId
|
|
* @param {import("pino").Logger} runLog
|
|
* @param {string} reason
|
|
*/
|
|
async function markStepSkipped(parsed, index, last, runId, runLog, reason) {
|
|
const script = parsed.kind === "set" ? SET_STEP_SCRIPT : parsed.script;
|
|
const step = await store.startStep({
|
|
runId,
|
|
index,
|
|
script,
|
|
config: parsed.config,
|
|
});
|
|
const stepLog = runLog.child({ stepId: step.id, script });
|
|
stepLog.debug({ reason }, "step skipped");
|
|
await store.finishStep(step.id, "skipped", storedEnvelope(last), reason);
|
|
}
|
|
|
|
/**
|
|
* @param {import("./workflow-parse.js").CompiledScripts} compiled
|
|
* @param {{ data?: unknown, context?: Record<string, unknown> }} ctx
|
|
* @param {string} runId
|
|
* @param {import("pino").Logger} runLog
|
|
* @param {string} key
|
|
* @param {string} owner
|
|
* @param {number} depth
|
|
*/
|
|
async function runLinearSteps(compiled, ctx, runId, runLog, key, owner, depth) {
|
|
/** @type {import("./step-result.js").StepResult} */
|
|
let last = {
|
|
output: ctx.data,
|
|
context: normalizeContext(ctx.context),
|
|
skipRemaining: false,
|
|
};
|
|
let next = { data: ctx.data, context: last.context };
|
|
for (let i = 0; i < compiled.order.length; i++) {
|
|
const index = compiled.order[i];
|
|
const parsed = compiled.steps[index];
|
|
last = await runCompiledStep(
|
|
parsed,
|
|
next,
|
|
index,
|
|
runId,
|
|
runLog,
|
|
key,
|
|
owner,
|
|
depth,
|
|
);
|
|
next = chainCtx(last);
|
|
if (last.skipRemaining) {
|
|
for (let j = i + 1; j < compiled.order.length; j++) {
|
|
const laterIndex = compiled.order[j];
|
|
await markStepSkipped(
|
|
compiled.steps[laterIndex],
|
|
laterIndex,
|
|
last,
|
|
runId,
|
|
runLog,
|
|
"skipRemaining",
|
|
);
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
return last;
|
|
}
|
|
|
|
/**
|
|
* @param {import("./workflow-parse.js").CompiledScripts} compiled
|
|
* @param {{ data?: unknown, context?: Record<string, unknown> }} ctx
|
|
* @param {string} runId
|
|
* @param {import("pino").Logger} runLog
|
|
* @param {string} key
|
|
* @param {string} owner
|
|
* @param {number} depth
|
|
*/
|
|
async function runDagSteps(compiled, ctx, runId, runLog, key, owner, depth) {
|
|
const triggerData = ctx.data;
|
|
/** @type {Map<string, unknown>} */
|
|
const outputsById = new Map();
|
|
/** @type {Map<string, number>} */
|
|
const idToIndex = new Map();
|
|
for (const step of compiled.steps) {
|
|
if (step.id) idToIndex.set(step.id, step.index);
|
|
}
|
|
|
|
const remaining = new Set(compiled.steps.map((s) => s.index));
|
|
const completed = new Set();
|
|
let context = normalizeContext(ctx.context);
|
|
/** @type {import("./step-result.js").StepResult} */
|
|
let last = {
|
|
output: ctx.data,
|
|
context,
|
|
skipRemaining: false,
|
|
};
|
|
let execIndex = 0;
|
|
|
|
/**
|
|
* @param {import("./workflow-parse.js").CompiledStep} step
|
|
*/
|
|
function needsMet(step) {
|
|
if (step.needsKind === "none" || step.needs.length === 0) return true;
|
|
return step.needs.every((edge) => completed.has(idToIndex.get(edge.from)));
|
|
}
|
|
|
|
while (remaining.size) {
|
|
const wave = compiled.steps
|
|
.filter((s) => remaining.has(s.index) && needsMet(s))
|
|
.sort((a, b) => a.index - b.index);
|
|
if (wave.length === 0) {
|
|
throw new Error("Workflow has a cycle");
|
|
}
|
|
|
|
const snapshot = { ...context };
|
|
/** @type {Array<{ id: string, context: unknown }>} */
|
|
const patches = [];
|
|
let skipRest = false;
|
|
|
|
for (const parsed of wave) {
|
|
remaining.delete(parsed.index);
|
|
const index = execIndex;
|
|
execIndex += 1;
|
|
if (skipRest) {
|
|
await markStepSkipped(
|
|
parsed,
|
|
index,
|
|
last,
|
|
runId,
|
|
runLog,
|
|
"skipRemaining",
|
|
);
|
|
continue;
|
|
}
|
|
const data = mergeStepData(parsed, outputsById, triggerData);
|
|
last = await runCompiledStep(
|
|
parsed,
|
|
{ data, context: { ...snapshot } },
|
|
index,
|
|
runId,
|
|
runLog,
|
|
key,
|
|
owner,
|
|
depth,
|
|
);
|
|
if (parsed.id) {
|
|
outputsById.set(parsed.id, last.output);
|
|
}
|
|
patches.push({
|
|
id: parsed.id ?? String(parsed.index),
|
|
context: last.context,
|
|
});
|
|
if (last.skipRemaining) skipRest = true;
|
|
}
|
|
|
|
context = mergeContextWave(snapshot, patches);
|
|
last = { ...last, context };
|
|
|
|
if (skipRest) {
|
|
for (const later of compiled.steps.filter((s) => remaining.has(s.index))) {
|
|
remaining.delete(later.index);
|
|
const index = execIndex;
|
|
execIndex += 1;
|
|
await markStepSkipped(
|
|
later,
|
|
index,
|
|
last,
|
|
runId,
|
|
runLog,
|
|
"skipRemaining",
|
|
);
|
|
}
|
|
break;
|
|
}
|
|
|
|
for (const parsed of wave) completed.add(parsed.index);
|
|
}
|
|
|
|
return last;
|
|
}
|
|
|
|
/**
|
|
* @param {string} owner
|
|
* @param {string} parentKey
|
|
* @param {string} parentRunId
|
|
* @param {number} depth
|
|
*/
|
|
function createWorkflowsApi(owner, parentKey, parentRunId, depth) {
|
|
return {
|
|
/**
|
|
* @param {string} name
|
|
* @param {unknown} [data]
|
|
*/
|
|
async trigger(name, data) {
|
|
if (depth >= MAX_WORKFLOW_TRIGGER_DEPTH) {
|
|
throw new Error(
|
|
`workflow trigger depth limit (${MAX_WORKFLOW_TRIGGER_DEPTH}) exceeded`,
|
|
);
|
|
}
|
|
const destKey = resolveWorkflowTriggerKey(owner, name);
|
|
return runWorkflow(
|
|
destKey,
|
|
{ data },
|
|
{ type: "workflow", detail: parentKey },
|
|
{
|
|
parentRunId,
|
|
depth: depth + 1,
|
|
detach: true,
|
|
},
|
|
);
|
|
},
|
|
};
|
|
}
|
|
|
|
/**
|
|
* @param {import("./workflow-parse.js").CompiledStep} parsed
|
|
* @param {{ data?: unknown, context?: unknown, config?: unknown }} ctx
|
|
* @param {number} index
|
|
* @param {string} runId
|
|
* @param {import("pino").Logger} runLog
|
|
* @param {string} key
|
|
* @param {string} owner
|
|
* @param {number} depth
|
|
* @returns {Promise<import("./step-result.js").StepResult>}
|
|
*/
|
|
async function runCompiledStep(
|
|
parsed,
|
|
ctx,
|
|
index,
|
|
runId,
|
|
runLog,
|
|
key,
|
|
owner,
|
|
depth,
|
|
) {
|
|
const script = parsed.kind === "set" ? SET_STEP_SCRIPT : parsed.script;
|
|
const unresolvedConfig = parsed.config;
|
|
const incomingContext = normalizeContext(ctx.context);
|
|
const step = await store.startStep({
|
|
runId,
|
|
index,
|
|
script,
|
|
config: unresolvedConfig,
|
|
});
|
|
const stepLog = runLog.child({ stepId: step.id, script });
|
|
try {
|
|
const config = await resolveConfigRefs(unresolvedConfig, {
|
|
owner,
|
|
workflowKey: key,
|
|
context: incomingContext,
|
|
});
|
|
const stepCtx = {
|
|
data: ctx.data,
|
|
context: incomingContext,
|
|
config,
|
|
};
|
|
if (parsed.when) {
|
|
const whenResult = await evaluateJsonata(parsed.when, stepCtx);
|
|
if (!isJsonataTruthy(whenResult)) {
|
|
stepLog.debug({ when: parsed.when }, "step skipped");
|
|
const skipped = {
|
|
output: stepCtx.data,
|
|
context: incomingContext,
|
|
skipRemaining: false,
|
|
};
|
|
await store.finishStep(step.id, "skipped", storedEnvelope(skipped), "when condition");
|
|
return skipped;
|
|
}
|
|
}
|
|
if (parsed.kind === "set") {
|
|
const value = await evaluateJsonata(parsed.expression, stepCtx);
|
|
const result = {
|
|
output: value,
|
|
context: incomingContext,
|
|
skipRemaining: false,
|
|
};
|
|
await store.finishStep(step.id, "success", storedEnvelope(result));
|
|
return result;
|
|
}
|
|
const raw = await runScript(script, stepCtx, {
|
|
log: stepLog,
|
|
workflowName: key,
|
|
owner,
|
|
$workflows: createWorkflowsApi(owner, key, runId, depth),
|
|
});
|
|
const result = normalizeStepResult(raw, incomingContext, script);
|
|
await store.finishStep(step.id, "success", storedEnvelope(result));
|
|
return result;
|
|
} catch (err) {
|
|
await store.finishStep(step.id, "failed", null, err);
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @param {string} key
|
|
* @param {{ data?: unknown, context?: unknown }} context
|
|
* @param {{ type: string, detail?: string | null }} trigger
|
|
* @param {{
|
|
* parentRunId?: string | null,
|
|
* depth?: number,
|
|
* detach?: boolean,
|
|
* }} [opts]
|
|
*/
|
|
async function runWorkflow(key, context, trigger, opts = {}) {
|
|
const parentRunId = opts.parentRunId ?? null;
|
|
const depth = opts.depth ?? 0;
|
|
const detach = opts.detach === true;
|
|
|
|
const entry = workflows.get(key);
|
|
if (!entry) {
|
|
log.error({ workflow: key }, "workflow not found");
|
|
return { runId: null, status: "failed", error: "workflow not found" };
|
|
}
|
|
|
|
const { owner, workflow } = entry;
|
|
if (workflow?.enabled === false && trigger.type !== "manual") {
|
|
log.debug({ workflow: key, trigger }, "skipping disabled workflow");
|
|
return { runId: null, status: "failed", error: "workflow disabled" };
|
|
}
|
|
const run = await store.startRun({
|
|
owner,
|
|
workflow: key,
|
|
workflowName: workflow?.name,
|
|
trigger,
|
|
input: context.data,
|
|
parentRunId,
|
|
});
|
|
const runLog = log.child({ runId: run.id, owner, workflow: key });
|
|
runLog.debug(detach ? "running workflow (detached)" : "running workflow");
|
|
|
|
const initialCtx = {
|
|
data: context.data ?? workflow.data ?? null,
|
|
context: normalizeContext(context.context),
|
|
};
|
|
|
|
const execute = async () => {
|
|
let ctx = initialCtx;
|
|
try {
|
|
const compiled = compileWorkflowScripts(workflow.scripts);
|
|
if (compiled.dagMode) {
|
|
ctx = await runDagSteps(
|
|
compiled,
|
|
ctx,
|
|
run.id,
|
|
runLog,
|
|
key,
|
|
owner,
|
|
depth,
|
|
);
|
|
} else {
|
|
ctx = await runLinearSteps(
|
|
compiled,
|
|
ctx,
|
|
run.id,
|
|
runLog,
|
|
key,
|
|
owner,
|
|
depth,
|
|
);
|
|
}
|
|
await store.finishRun(run.id, "success", storedEnvelope(ctx));
|
|
return { runId: run.id, status: "success", result: storedEnvelope(ctx) };
|
|
} catch (err) {
|
|
runLog.error({ err }, "workflow failed");
|
|
await store.finishRun(run.id, "failed", null, err);
|
|
return {
|
|
runId: run.id,
|
|
status: "failed",
|
|
error: err instanceof Error ? err.message : String(err),
|
|
};
|
|
}
|
|
};
|
|
|
|
if (detach) {
|
|
execute().catch((err) => {
|
|
runLog.error({ err }, "detached workflow failed unexpectedly");
|
|
});
|
|
return { runId: run.id, status: "started" };
|
|
}
|
|
|
|
return execute();
|
|
}
|
|
|
|
function reregister() {
|
|
registerWorkflows();
|
|
registerHttpTriggers();
|
|
registerCronTriggers();
|
|
}
|
|
|
|
function referencedScripts() {
|
|
const refs = new Set();
|
|
for (const { workflow } of workflows.values()) {
|
|
for (const raw of workflow.scripts ?? []) {
|
|
try {
|
|
const parsed = parseScriptStep(raw);
|
|
if (parsed.kind === "script") refs.add(parsed.script);
|
|
} catch {
|
|
// skip invalid steps
|
|
}
|
|
}
|
|
}
|
|
return refs;
|
|
}
|
|
|
|
return {
|
|
workflows,
|
|
loadErrors,
|
|
registerWorkflows,
|
|
registerHttpTriggers,
|
|
registerCronTriggers,
|
|
registerPruneJob,
|
|
reregister,
|
|
runWorkflow,
|
|
referencedScripts,
|
|
};
|
|
}
|