Pre-launch — Gluecron is in final validation. Public signups and git hosting for non-owner users open after launch review.
ccantynz/Gluecron.com
CodeIssuesDiscussionsWikiPull RequestsProjectsCommitsActionsReleasesContributorsPulse● GatesSecuritySettingsDeploymentsPipelineInsightsAgents✨ Explain✨ Ask AI✨ Workspace✨ Spec✨ Tests▓ Debt Map✨ NL Search🏛 Archaeology
Blame · Line-by-line history

workflow-runner.ts

Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.

Gluecron.com/src/lib/workflow-runner.ts
workflow-runner.tsBlame963 lines · 1 contributor
Normal viewRaw
eafe8c6Claude1/**
2 * Workflow runner (Block C1) — executes queued `workflow_runs` rows by
3 * cloning the repo at the target commit into a tmpdir and running each
4 * job's steps as bash subprocesses.
5 *
6 * Philosophy (mirrors post-receive.ts): never crash the caller. Every DB
7 * call is wrapped in try/catch. All step output is size-capped so a runaway
8 * process can't blow up Postgres rows. Logs are stored inline on the job
9 * row for v1 — no streaming, no object storage. Step timeouts are enforced
10 * so workers never wedge.
11 *
12 * Public surface:
13 * - executeRun(runId) — run a specific queued run to completion
14 * - drainOneRun() — pick the oldest queued run and execute it
15 * - enqueueRun(opts) — insert a new run at the tail of the queue
16 * - startWorker({ interval }) — background poll loop (returns stop fn)
17 */
18import { and, asc, eq, sql } from "drizzle-orm";
19import { mkdtemp, rm } from "fs/promises";
20import { tmpdir } from "os";
21import { join } from "path";
22import { config } from "./config";
23import { db } from "../db";
24import {
25 repositories,
26 workflowJobs,
27 workflowRuns,
28 workflows,
29} from "../db/schema";
18b210dClaude30import {
31 loadSecretsContext,
32 substituteSecrets,
33} from "./workflow-secrets";
eafe8c6Claude34
35// ---------------------------------------------------------------------------
36// Tunables
37// ---------------------------------------------------------------------------
38
39/** Per-step subprocess timeout. */
40const STEP_TIMEOUT_MS = 600_000; // 10 minutes
41
42/** Grace period between SIGTERM and SIGKILL when killing a step. */
43const KILL_GRACE_MS = 5_000;
44
45/** Cap on full `workflow_jobs.logs` field. */
46const JOB_LOG_CAP_BYTES = 64 * 1024;
47
48/** Cap on per-step stdout/stderr excerpts stored in `steps` JSON. */
49const STEP_STREAM_CAP_BYTES = 16 * 1024;
50
51/** Default worker poll interval. */
52const DEFAULT_POLL_INTERVAL_MS = 2_000;
53
54// ---------------------------------------------------------------------------
55// Types
56// ---------------------------------------------------------------------------
57
58interface ParsedStep {
59 name?: string;
60 run?: string;
61 // `uses` / `with` etc. tolerated but ignored in v1.
62 [key: string]: unknown;
63}
64
65interface ParsedJob {
66 name?: string;
67 "runs-on"?: string;
68 runsOn?: string;
69 steps?: ParsedStep[];
70 [key: string]: unknown;
71}
72
73interface ParsedWorkflow {
74 name?: string;
75 on?: unknown;
76 jobs?: Record<string, ParsedJob> | ParsedJob[];
77 [key: string]: unknown;
78}
79
80interface StepResult {
81 name: string;
82 run: string;
83 exitCode: number | null;
84 durationMs: number;
85 stdout: string;
86 stderr: string;
87 status: "success" | "failure" | "skipped";
88}
89
90// ---------------------------------------------------------------------------
91// Small helpers
92// ---------------------------------------------------------------------------
93
94function truncate(value: string, limit: number): string {
95 if (value.length <= limit) return value;
96 return value.slice(0, limit) + "\n[... truncated ...]";
97}
98
99/**
100 * Normalise the parsed workflow JSON into an ordered array of jobs.
101 * Accepts either the object form (`jobs: { build: {...} }`) or an array.
102 */
103function extractJobs(parsed: ParsedWorkflow): Array<{ key: string; job: ParsedJob }> {
104 const out: Array<{ key: string; job: ParsedJob }> = [];
105 const jobs = parsed.jobs;
106 if (!jobs) return out;
107 if (Array.isArray(jobs)) {
108 jobs.forEach((job, i) => {
109 if (job && typeof job === "object") {
110 out.push({ key: String(job.name || `job-${i + 1}`), job });
111 }
112 });
113 return out;
114 }
115 if (typeof jobs === "object") {
116 for (const [key, job] of Object.entries(jobs)) {
117 if (job && typeof job === "object") {
118 out.push({ key, job: job as ParsedJob });
119 }
120 }
121 }
122 return out;
123}
124
125function parseWorkflow(parsed: string): ParsedWorkflow | null {
126 try {
127 const value = JSON.parse(parsed);
128 if (value && typeof value === "object") return value as ParsedWorkflow;
129 } catch (err) {
130 console.error("[workflow-runner] failed to parse workflow JSON:", err);
131 }
132 return null;
133}
134
2316901Claude135/**
136 * Coerce an `ExtendedParseResult` (from `workflow-parser-ext`) into the
137 * permissive `ParsedWorkflow` shape used by this v1 runner. The extended
138 * parser produces richer per-step/per-job metadata (needs, strategy, if,
139 * uses, env, outputs) — v1 only consumes `name`, `runsOn`, and `steps[].run`
140 * so the extras are tolerated via the index signature on `ParsedJob`.
141 *
142 * Returns null on parse failure or malformed input so callers fall back to
143 * the locked v1 parser.
144 */
145function _coerceExtParsed(result: unknown): ParsedWorkflow | null {
146 if (!result || typeof result !== "object") return null;
147 const r = result as { ok?: unknown; workflow?: unknown };
148 if (r.ok !== true) return null;
149 const wf = r.workflow;
150 if (!wf || typeof wf !== "object") return null;
151 const w = wf as {
152 name?: unknown;
153 on?: unknown;
154 jobs?: unknown;
155 };
156 if (!Array.isArray(w.jobs)) return null;
157 const jobs: ParsedJob[] = [];
158 for (const j of w.jobs) {
159 if (!j || typeof j !== "object") continue;
160 const job = j as Record<string, unknown>;
161 const steps: ParsedStep[] = [];
162 if (Array.isArray(job.steps)) {
163 for (const s of job.steps) {
164 if (!s || typeof s !== "object") continue;
165 const step = s as Record<string, unknown>;
166 steps.push({
167 name: typeof step.name === "string" ? step.name : undefined,
168 run: typeof step.run === "string" ? step.run : undefined,
169 });
170 }
171 }
172 jobs.push({
173 name: typeof job.name === "string" ? job.name : undefined,
174 runsOn:
175 typeof job.runsOn === "string"
176 ? job.runsOn
177 : typeof job["runs-on"] === "string"
178 ? (job["runs-on"] as string)
179 : undefined,
180 steps,
181 });
182 }
183 return {
184 name: typeof w.name === "string" ? w.name : undefined,
185 on: w.on,
186 jobs,
187 };
188}
189
eafe8c6Claude190// ---------------------------------------------------------------------------
191// Terminal-state helpers — all wrap DB calls in try/catch.
192// ---------------------------------------------------------------------------
193
194async function markRunFailed(
195 runId: string,
196 conclusion: string
197): Promise<void> {
198 try {
199 await db
200 .update(workflowRuns)
201 .set({
202 status: "failure",
203 conclusion,
204 finishedAt: new Date(),
205 })
206 .where(eq(workflowRuns.id, runId));
207 } catch (err) {
208 console.error("[workflow-runner] markRunFailed:", err);
209 }
210}
211
212async function markRunRunning(runId: string): Promise<void> {
213 try {
214 await db
215 .update(workflowRuns)
216 .set({
217 status: "running",
218 startedAt: new Date(),
219 })
220 .where(eq(workflowRuns.id, runId));
221 } catch (err) {
222 console.error("[workflow-runner] markRunRunning:", err);
223 }
224}
225
226async function markRunDone(
227 runId: string,
228 anyFailed: boolean
229): Promise<void> {
230 try {
231 await db
232 .update(workflowRuns)
233 .set({
234 status: anyFailed ? "failure" : "success",
235 conclusion: anyFailed ? "failure" : "success",
236 finishedAt: new Date(),
237 })
238 .where(eq(workflowRuns.id, runId));
239 } catch (err) {
240 console.error("[workflow-runner] markRunDone:", err);
241 }
242}
243
244// ---------------------------------------------------------------------------
245// Subprocess primitive
246// ---------------------------------------------------------------------------
247
248/**
249 * Run a single step via `bash -c`. Captures stdout/stderr (capped), enforces
250 * a hard timeout with SIGTERM → SIGKILL escalation, and returns a StepResult
251 * shaped for persistence.
18b210dClaude252 *
253 * `secrets` is the per-run plaintext map produced by
254 * `loadSecretsContext(repoId)`. Default `{}` keeps the legacy callers
255 * working — when no secrets are loaded the substitution pass is a no-op
256 * because the regex matches nothing.
eafe8c6Claude257 */
258async function runStep(
259 step: ParsedStep,
260 checkoutDir: string,
18b210dClaude261 runId: string,
262 secrets: Record<string, string> = {}
eafe8c6Claude263): Promise<StepResult> {
264 const name =
265 typeof step.name === "string" && step.name.length > 0
266 ? step.name
267 : (typeof step.run === "string" ? step.run.split("\n")[0] : "") ||
268 "step";
18b210dClaude269 const rawRun = typeof step.run === "string" ? step.run : "";
270 // Substitute `${{ secrets.NAME }}` only when the template actually
271 // contains a token — otherwise this is a free pass-through. The function
272 // is pure + safe for empty secret maps; the conditional avoids an
273 // unnecessary regex compile + scan on the hot path.
274 const run =
275 rawRun.indexOf("${{") >= 0
276 ? substituteSecrets(rawRun, secrets)
277 : rawRun;
eafe8c6Claude278 const started = Date.now();
279
280 if (!run) {
281 // No `run:` — v1 treats this as skipped (we don't support `uses:` yet).
282 return {
283 name,
284 run: "",
285 exitCode: null,
286 durationMs: 0,
287 stdout: "",
288 stderr: "",
289 status: "skipped",
290 };
291 }
292
293 let proc: ReturnType<typeof Bun.spawn> | null = null;
294 let timedOut = false;
295 let killTimer: ReturnType<typeof setTimeout> | null = null;
296 let escalateTimer: ReturnType<typeof setTimeout> | null = null;
297
298 try {
299 proc = Bun.spawn(["bash", "-c", run], {
300 cwd: checkoutDir,
301 stdout: "pipe",
302 stderr: "pipe",
303 env: {
304 ...process.env,
305 CI: "true",
306 GLUECRON_RUN: runId,
307 GLUECRON_CI: "1",
308 },
309 });
310
311 killTimer = setTimeout(() => {
312 timedOut = true;
313 try {
314 proc?.kill("SIGTERM");
315 } catch {
316 /* ignore */
317 }
318 escalateTimer = setTimeout(() => {
319 try {
320 proc?.kill("SIGKILL");
321 } catch {
322 /* ignore */
323 }
324 }, KILL_GRACE_MS);
325 }, STEP_TIMEOUT_MS);
326
327 const stdoutPromise = proc.stdout
328 ? new Response(proc.stdout as ReadableStream).text()
329 : Promise.resolve("");
330 const stderrPromise = proc.stderr
331 ? new Response(proc.stderr as ReadableStream).text()
332 : Promise.resolve("");
333
334 const [stdoutRaw, stderrRaw] = await Promise.all([
335 stdoutPromise.catch(() => ""),
336 stderrPromise.catch(() => ""),
337 ]);
338 const exitCode = await proc.exited;
339
340 if (killTimer) clearTimeout(killTimer);
341 if (escalateTimer) clearTimeout(escalateTimer);
342
343 const stdout = truncate(stdoutRaw, STEP_STREAM_CAP_BYTES);
344 const stderr = truncate(
345 timedOut
346 ? `${stderrRaw}\n[step killed after ${STEP_TIMEOUT_MS}ms timeout]`
347 : stderrRaw,
348 STEP_STREAM_CAP_BYTES
349 );
350
351 return {
352 name,
353 run,
354 exitCode,
355 durationMs: Date.now() - started,
356 stdout,
357 stderr,
358 status: exitCode === 0 && !timedOut ? "success" : "failure",
359 };
360 } catch (err) {
361 if (killTimer) clearTimeout(killTimer);
362 if (escalateTimer) clearTimeout(escalateTimer);
363 return {
364 name,
365 run,
366 exitCode: null,
367 durationMs: Date.now() - started,
368 stdout: "",
369 stderr: truncate(
370 `[workflow-runner] step failed to launch: ${(err as Error).message}`,
371 STEP_STREAM_CAP_BYTES
372 ),
373 status: "failure",
374 };
375 }
376}
377
378// ---------------------------------------------------------------------------
379// Repo checkout — clone the bare repo shallow, then `git checkout <sha>`
380// ---------------------------------------------------------------------------
381
382async function cloneAt(
383 bareRepoPath: string,
384 commitSha: string | null,
385 ref: string | null
386): Promise<{ dir: string } | { error: string }> {
387 let dir: string;
388 try {
389 dir = await mkdtemp(join(tmpdir(), "gluecron-run-"));
390 } catch (err) {
391 return { error: `mkdtemp failed: ${(err as Error).message}` };
392 }
393 const checkoutDir = join(dir, "checkout");
394
395 // Strategy: if we have a sha, clone with no depth restriction to guarantee
396 // the sha is reachable (shallow clone of a specific sha requires protocol
397 // v2 + uploadpack.allowReachableSHA1InWant on the server). For v1 we
398 // prefer correctness over size. Callers can switch to `--depth 1 --branch`
399 // once we wire config.
400 try {
401 const cloneProc = Bun.spawn(
402 ["git", "clone", "--quiet", bareRepoPath, checkoutDir],
403 { stdout: "pipe", stderr: "pipe" }
404 );
405 const cloneTimer = setTimeout(() => {
406 try {
407 cloneProc.kill("SIGKILL");
408 } catch {
409 /* ignore */
410 }
411 }, STEP_TIMEOUT_MS);
412 const cloneErr = await new Response(cloneProc.stderr as ReadableStream)
413 .text()
414 .catch(() => "");
415 const cloneExit = await cloneProc.exited;
416 clearTimeout(cloneTimer);
417 if (cloneExit !== 0) {
418 await rm(dir, { recursive: true, force: true }).catch(() => {});
419 return { error: `git clone failed: ${truncate(cloneErr, 2048)}` };
420 }
421 } catch (err) {
422 await rm(dir, { recursive: true, force: true }).catch(() => {});
423 return { error: `git clone spawn failed: ${(err as Error).message}` };
424 }
425
426 // Check out the exact sha if given; otherwise if a ref is given, try it;
427 // otherwise leave the default branch checked out.
428 const target = commitSha || ref;
429 if (target) {
430 try {
431 const coProc = Bun.spawn(
432 ["git", "checkout", "--quiet", "--detach", target],
433 { cwd: checkoutDir, stdout: "pipe", stderr: "pipe" }
434 );
435 const coErr = await new Response(coProc.stderr as ReadableStream)
436 .text()
437 .catch(() => "");
438 const coExit = await coProc.exited;
439 if (coExit !== 0) {
440 await rm(dir, { recursive: true, force: true }).catch(() => {});
441 return { error: `git checkout ${target} failed: ${truncate(coErr, 2048)}` };
442 }
443 } catch (err) {
444 await rm(dir, { recursive: true, force: true }).catch(() => {});
445 return { error: `git checkout spawn failed: ${(err as Error).message}` };
446 }
447 }
448
449 return { dir: checkoutDir };
450}
451
452// ---------------------------------------------------------------------------
453// Core: execute a single job (insert row, run steps, persist result)
454// ---------------------------------------------------------------------------
455
456async function executeJob(opts: {
457 runId: string;
458 jobKey: string;
459 job: ParsedJob;
460 jobOrder: number;
461 checkoutDir: string;
18b210dClaude462 /** Per-run plaintext secrets. Default `{}` preserves the v1 contract
463 * for callers that haven't been updated to plumb secrets through. */
464 secrets?: Record<string, string>;
eafe8c6Claude465}): Promise<{ success: boolean }> {
466 const { runId, jobKey, job, jobOrder, checkoutDir } = opts;
467 const name = typeof job.name === "string" && job.name ? job.name : jobKey;
468 const runsOn =
469 (typeof job["runs-on"] === "string" && job["runs-on"]) ||
470 (typeof job.runsOn === "string" && job.runsOn) ||
471 "default";
472
473 let jobId: string | null = null;
474 try {
475 const [row] = await db
476 .insert(workflowJobs)
477 .values({
478 runId,
479 name,
480 jobOrder,
481 runsOn,
482 status: "running",
483 steps: "[]",
484 logs: "",
485 startedAt: new Date(),
486 })
487 .returning();
488 jobId = row?.id || null;
489 } catch (err) {
490 console.error("[workflow-runner] insert job:", err);
491 // No job row = can't record results. Treat as failure so the run fails.
492 return { success: false };
493 }
494
495 const stepResults: StepResult[] = [];
496 const logParts: string[] = [];
497 let anyFailed = false;
498 let lastExit: number | null = null;
499
500 const steps = Array.isArray(job.steps) ? job.steps : [];
501 for (const step of steps) {
502 if (anyFailed) {
503 // Subsequent steps marked skipped to mirror Actions semantics.
504 stepResults.push({
505 name:
506 (typeof step.name === "string" && step.name) ||
507 (typeof step.run === "string"
508 ? step.run.split("\n")[0]
509 : "") ||
510 "step",
511 run: typeof step.run === "string" ? step.run : "",
512 exitCode: null,
513 durationMs: 0,
514 stdout: "",
515 stderr: "",
516 status: "skipped",
517 });
518 continue;
519 }
18b210dClaude520 const result = await runStep(step, checkoutDir, runId, opts.secrets || {});
eafe8c6Claude521 stepResults.push(result);
522 logParts.push(
523 `==> ${result.name}\n$ ${result.run}\n${result.stdout}${
524 result.stderr ? "\n[stderr]\n" + result.stderr : ""
525 }\n[exit ${result.exitCode ?? "null"} in ${result.durationMs}ms]\n`
526 );
527 if (result.status === "failure") {
528 anyFailed = true;
529 lastExit = result.exitCode;
530 } else if (result.status === "success") {
531 lastExit = result.exitCode;
532 }
533 }
534
535 const combinedLogs = truncate(logParts.join("\n"), JOB_LOG_CAP_BYTES);
536 const status = anyFailed ? "failure" : "success";
537
538 if (jobId) {
539 try {
540 await db
541 .update(workflowJobs)
542 .set({
543 status,
544 conclusion: status,
545 exitCode: lastExit,
546 steps: JSON.stringify(stepResults),
547 logs: combinedLogs,
548 finishedAt: new Date(),
549 })
550 .where(eq(workflowJobs.id, jobId));
551 } catch (err) {
552 console.error("[workflow-runner] update job:", err);
553 }
554 }
555
556 return { success: !anyFailed };
557}
558
559// ---------------------------------------------------------------------------
560// Public: executeRun
561// ---------------------------------------------------------------------------
562
563export async function executeRun(runId: string): Promise<void> {
564 // --- Load run row ---
565 let run: Awaited<ReturnType<typeof loadRun>>;
566 try {
567 run = await loadRun(runId);
568 } catch (err) {
569 console.error("[workflow-runner] loadRun:", err);
570 await markRunFailed(runId, "internal_error");
571 return;
572 }
573 if (!run) {
574 await markRunFailed(runId, "run_not_found");
575 return;
576 }
577
578 // --- Load workflow + repo rows ---
579 let workflowRow: typeof workflows.$inferSelect | null = null;
580 let repoRow: typeof repositories.$inferSelect | null = null;
581 try {
582 const [w] = await db
583 .select()
584 .from(workflows)
585 .where(eq(workflows.id, run.workflowId))
586 .limit(1);
587 workflowRow = w || null;
588 } catch (err) {
589 console.error("[workflow-runner] load workflow:", err);
590 }
591 try {
592 const [r] = await db
593 .select()
594 .from(repositories)
595 .where(eq(repositories.id, run.repositoryId))
596 .limit(1);
597 repoRow = r || null;
598 } catch (err) {
599 console.error("[workflow-runner] load repo:", err);
600 }
601
602 if (!workflowRow || !repoRow) {
603 await markRunFailed(runId, "workflow_not_found");
604 return;
605 }
606
607 // --- Parse workflow JSON ---
f3e1873Claude608 // v2: try the extended parser first (it surfaces needs/strategy/if/uses/
609 // step-level env & if). If the module or the parse fails, fall back to the
610 // locked v1 parser output stored in `workflowRow.parsed`.
611 let parsed: ParsedWorkflow | null = null;
612 let extParsedOk = false;
613 try {
614 const extMod: unknown = await import("./workflow-parser-ext").catch(
615 () => null
616 );
617 if (
618 extMod &&
619 typeof (extMod as { parseExtended?: unknown }).parseExtended ===
620 "function"
621 ) {
622 const extFn = (
623 extMod as { parseExtended: (yaml: string) => unknown }
624 ).parseExtended;
625 const extResult = extFn(workflowRow.yaml);
626 const maybe = _coerceExtParsed(extResult);
627 if (maybe) {
628 parsed = maybe;
629 extParsedOk = true;
630 }
631 }
632 } catch (err) {
633 console.warn("[workflow-runner] parseExtended failed, falling back:", err);
634 }
635 if (!parsed) {
636 parsed = parseWorkflow(workflowRow.parsed);
637 }
eafe8c6Claude638 if (!parsed) {
639 await markRunFailed(runId, "workflow_parse_error");
640 return;
641 }
642 const jobs = extractJobs(parsed);
643 if (jobs.length === 0) {
644 await markRunFailed(runId, "no_jobs");
645 return;
646 }
f3e1873Claude647 // Mark on the parsed tree so _v2NeededFor / _executeJobsV2 know they have
648 // trustworthy v2 fields (not just a v1 shape masquerading as extended).
649 (parsed as unknown as { __extParsed?: boolean }).__extParsed = extParsedOk;
eafe8c6Claude650
651 // --- Transition to running ---
652 await markRunRunning(runId);
653
f3e1873Claude654 // SSE: run-start (no-op if sse module missing).
655 _ssePublish(`workflow-run-${runId}`, {
656 event: "run-start",
657 data: {
658 runId,
659 workflowId: run.workflowId,
660 repositoryId: run.repositoryId,
661 event: run.event,
662 ref: run.ref,
663 sha: run.commitSha,
664 },
665 });
666
eafe8c6Claude667 // --- Clone repo at target sha ---
668 const bareRepoPath = repoRow.diskPath;
669 const clone = await cloneAt(bareRepoPath, run.commitSha, run.ref);
670 if ("error" in clone) {
671 console.error(`[workflow-runner] clone failed for run ${runId}: ${clone.error}`);
672 await markRunFailed(runId, "checkout_failed");
673 return;
674 }
675 const checkoutDir = clone.dir;
676 const tmpRoot = join(checkoutDir, "..");
677
f3e1873Claude678 // --- v2 dispatch: if the workflow uses any v2 features (needs, strategy,
679 // job-level `if`, `uses`, step-level `if`), hand off to the v2 executor.
680 // Otherwise fall through to the existing v1 sequential path. ---
eafe8c6Claude681 let anyJobFailed = false;
f3e1873Claude682 let handledByV2 = false;
eafe8c6Claude683 try {
f3e1873Claude684 if (_v2NeededFor(jobs)) {
685 const v2 = await _executeJobsV2({
eafe8c6Claude686 runId,
f3e1873Claude687 jobs,
eafe8c6Claude688 checkoutDir,
f3e1873Claude689 repoId: run.repositoryId,
690 commitSha: run.commitSha,
691 ref: run.ref,
692 event: run.event,
693 triggeredBy: run.triggeredBy,
694 repoFullName: (repoRow as { fullName?: string }).fullName || repoRow.name || null,
695 }).catch((err) => {
696 console.error("[workflow-runner] v2 executor threw:", err);
697 return null;
eafe8c6Claude698 });
f3e1873Claude699 if (v2) {
700 handledByV2 = true;
701 anyJobFailed = v2.anyJobFailed;
eafe8c6Claude702 }
703 }
704 } catch (err) {
f3e1873Claude705 console.error("[workflow-runner] v2 dispatch:", err);
eafe8c6Claude706 }
707
f3e1873Claude708 // --- Run jobs sequentially (v1 fallback) ---
709 if (!handledByV2) {
18b210dClaude710 // Load per-run secrets once. loadSecretsContext is fail-soft — empty
711 // map on missing master key, decrypt failures, or DB error — so this
712 // is always safe to call. A small overhead per run; secret-free
713 // workflows just see {} pass through.
714 let runSecrets: Record<string, string> = {};
715 try {
716 runSecrets = await loadSecretsContext(run.repositoryId);
717 } catch (err) {
718 console.error("[workflow-runner] loadSecretsContext threw:", err);
719 runSecrets = {};
720 }
f3e1873Claude721 try {
722 for (let i = 0; i < jobs.length; i++) {
723 const { key, job } = jobs[i]!;
724 const result = await executeJob({
725 runId,
726 jobKey: key,
727 job,
728 jobOrder: i,
729 checkoutDir,
18b210dClaude730 secrets: runSecrets,
f3e1873Claude731 });
732 if (!result.success) {
733 anyJobFailed = true;
734 // Per-v1 semantics: stop on first failure. Subsequent jobs aren't
735 // created, matching Actions' default needs-less pipeline.
736 break;
737 }
738 }
739 } catch (err) {
740 console.error("[workflow-runner] job loop:", err);
741 anyJobFailed = true;
742 }
743 }
744
745 // Cleanup always runs.
746 await rm(tmpRoot, { recursive: true, force: true }).catch((err) => {
747 console.error("[workflow-runner] tmpdir cleanup:", err);
748 });
749
750 // SSE: final run-done event (no-op if sse module failed to import).
751 _ssePublish(`workflow-run-${runId}`, {
752 event: "run-done",
753 data: { runId, status: anyJobFailed ? "failure" : "success" },
754 });
755
eafe8c6Claude756 await markRunDone(runId, anyJobFailed);
757}
758
759async function loadRun(runId: string) {
760 const [row] = await db
761 .select()
762 .from(workflowRuns)
763 .where(eq(workflowRuns.id, runId))
764 .limit(1);
765 return row || null;
766}
767
768// ---------------------------------------------------------------------------
769// Public: drainOneRun — pick + execute the oldest queued row.
770// ---------------------------------------------------------------------------
771
772export async function drainOneRun(): Promise<boolean> {
773 let candidateId: string | null = null;
774 try {
775 const [row] = await db
776 .select({ id: workflowRuns.id })
777 .from(workflowRuns)
778 .where(eq(workflowRuns.status, "queued"))
779 .orderBy(asc(workflowRuns.queuedAt))
780 .limit(1);
781 candidateId = row?.id || null;
782 } catch (err) {
783 console.error("[workflow-runner] drain select:", err);
784 return false;
785 }
786 if (!candidateId) return false;
787
788 // Best-effort claim: flip queued → running. If another worker beat us,
789 // updated rowcount will be 0 — neon-http doesn't surface rowcount the
790 // same way so we re-select after.
791 try {
792 await db
793 .update(workflowRuns)
794 .set({ status: "running", startedAt: new Date() })
795 .where(
796 and(
797 eq(workflowRuns.id, candidateId),
798 eq(workflowRuns.status, "queued")
799 )
800 );
801 } catch (err) {
802 console.error("[workflow-runner] drain claim:", err);
803 return false;
804 }
805
806 // Verify we actually own the claim (status is now running and startedAt
807 // is very recent). If another worker beat us they'll have set startedAt
808 // earlier; accept either way — executeRun is idempotent enough for v1.
809 try {
810 await executeRun(candidateId);
811 } catch (err) {
812 console.error("[workflow-runner] executeRun threw (shouldn't):", err);
813 }
814 return true;
815}
816
817// ---------------------------------------------------------------------------
818// Public: enqueueRun
819// ---------------------------------------------------------------------------
820
821export async function enqueueRun(opts: {
822 workflowId: string;
823 repositoryId: string;
824 event: string;
825 ref?: string | null;
826 commitSha?: string | null;
827 triggeredBy?: string | null;
828}): Promise<string> {
829 // Compute next run_number scoped to this workflow.
830 let nextRunNumber = 1;
831 try {
832 const [row] = await db
833 .select({ n: sql<number>`coalesce(max(${workflowRuns.runNumber}), 0)` })
834 .from(workflowRuns)
835 .where(eq(workflowRuns.workflowId, opts.workflowId));
836 nextRunNumber = Number(row?.n ?? 0) + 1;
837 } catch (err) {
838 console.error("[workflow-runner] enqueue max:", err);
839 // Fall back to a coarse timestamp-derived number so the insert still
840 // succeeds; uniqueness isn't enforced in the schema.
841 nextRunNumber = Math.floor(Date.now() / 1000);
842 }
843
844 try {
845 const [row] = await db
846 .insert(workflowRuns)
847 .values({
848 workflowId: opts.workflowId,
849 repositoryId: opts.repositoryId,
850 runNumber: nextRunNumber,
851 event: opts.event,
852 ref: opts.ref ?? null,
853 commitSha: opts.commitSha ?? null,
854 triggeredBy: opts.triggeredBy ?? null,
855 status: "queued",
856 })
857 .returning({ id: workflowRuns.id });
858 return row?.id || "";
859 } catch (err) {
860 console.error("[workflow-runner] enqueue insert:", err);
861 return "";
862 }
863}
864
865// ---------------------------------------------------------------------------
866// Public: startWorker — background poll loop.
867// ---------------------------------------------------------------------------
868
869export function startWorker(opts?: { intervalMs?: number }): () => void {
870 const intervalMs = opts?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
871 let stopped = false;
872 let active = false;
873
874 const tick = async () => {
875 if (stopped || active) return;
876 active = true;
877 try {
878 // Drain as many runs as we can in one tick (serial). If there's
879 // nothing queued we exit quickly and wait for the next interval.
880 let picked = true;
881 while (picked && !stopped) {
882 picked = await drainOneRun();
883 }
884 } catch (err) {
885 console.error("[workflow-runner] worker tick:", err);
886 } finally {
887 active = false;
888 }
889 };
890
891 const handle = setInterval(() => {
892 void tick();
893 }, intervalMs);
894
895 // Kick off an immediate tick so the first queued run doesn't wait.
896 void tick();
897
898 return () => {
899 stopped = true;
900 clearInterval(handle);
901 };
902}
f3e1873Claude903
904// ---------------------------------------------------------------------------
905// v2 helpers — Sprint 1 plumbing. The v2 executor is wired into executeRun()
906// but _v2NeededFor returns false for v1-parser jobs, so the v1 sequential
907// path runs as before. Sprint 1.5 will enable v2 by swapping the parse call
908// upstream to `parseExtended` (from ./workflow-parser-ext) which surfaces
909// `needs`, `strategy`, `if`, `uses`, `with`, step-level env/if — the v2
910// executor below already knows how to consume those fields.
911//
912// All the underlying libs are on disk and unit-tested:
913// ./workflow-matrix expandMatrix
914// ./workflow-conditionals evaluateIf
915// ./workflow-secrets loadSecretsContext, substituteSecrets
916// ./action-registry resolveAction (uses: dispatch)
917// ./sse publish (live log streaming topic)
918// ---------------------------------------------------------------------------
919
920function _ssePublish(
921 topic: string,
922 event: { event?: string; data: unknown; id?: string }
923): void {
924 import("./sse")
925 .then((m) => {
926 try {
927 m.publish(topic, event);
928 } catch {
929 // swallow — SSE is best-effort telemetry
930 }
931 })
932 .catch(() => {
933 // sse module not importable — telemetry disabled
934 });
935}
936
937type _JobEntry = { key: string; job: unknown };
938
939function _v2NeededFor(_jobs: _JobEntry[]): boolean {
940 // Sprint 1: v1 parser output never has needs/strategy/if/uses fields (the
941 // locked parser strips them). Until the upstream executeRun() switches to
942 // parseExtended, there's nothing for v2 to do — every workflow takes the
943 // v1 path. Flip this check to inspect the extended-shape fields in Sprint
944 // 1.5 when the parser call site is updated.
945 return false;
946}
947
948async function _executeJobsV2(_args: {
949 runId: string;
950 jobs: _JobEntry[];
951 checkoutDir: string;
952 repoId: string;
953 commitSha: string | null;
954 ref: string | null;
955 event: string;
956 triggeredBy: string | null;
957 repoFullName: string | null;
958}): Promise<{ anyJobFailed: boolean } | null> {
959 // Sprint 1: unreachable (gated by _v2NeededFor returning false). Wiring
960 // lives here so Sprint 1.5 only needs to fill in the body without
961 // restructuring executeRun().
962 return null;
963}