Pre-launch — Gluecron is in final validation. Public signups and git hosting for non-owner users open after launch review.
ccantynz/Gluecron.com
CodeIssuesPull RequestsActionsSecurityInsightsSettings
✨ AI
ExplainAsk AIWorkspaceSpec to PRTestsNL SearchDebt MapArchaeology
More
CommitsDiscussionsWikiProjectsReleasesContributorsPulseGatesDeploymentsPipelineAgents
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.tsBlame983 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) {
a28cedeClaude418 await rm(dir, { recursive: true, force: true }).catch((err) => {
419 console.warn(
420 `[workflow-runner] cleanup rm failed for ${dir}:`,
421 err instanceof Error ? err.message : err
422 );
423 });
eafe8c6Claude424 return { error: `git clone failed: ${truncate(cloneErr, 2048)}` };
425 }
426 } catch (err) {
a28cedeClaude427 await rm(dir, { recursive: true, force: true }).catch((err) => {
428 console.warn(
429 `[workflow-runner] cleanup rm failed for ${dir}:`,
430 err instanceof Error ? err.message : err
431 );
432 });
eafe8c6Claude433 return { error: `git clone spawn failed: ${(err as Error).message}` };
434 }
435
436 // Check out the exact sha if given; otherwise if a ref is given, try it;
437 // otherwise leave the default branch checked out.
438 const target = commitSha || ref;
439 if (target) {
440 try {
441 const coProc = Bun.spawn(
442 ["git", "checkout", "--quiet", "--detach", target],
443 { cwd: checkoutDir, stdout: "pipe", stderr: "pipe" }
444 );
445 const coErr = await new Response(coProc.stderr as ReadableStream)
446 .text()
447 .catch(() => "");
448 const coExit = await coProc.exited;
449 if (coExit !== 0) {
a28cedeClaude450 await rm(dir, { recursive: true, force: true }).catch((err) => {
451 console.warn(
452 `[workflow-runner] cleanup rm failed for ${dir}:`,
453 err instanceof Error ? err.message : err
454 );
455 });
eafe8c6Claude456 return { error: `git checkout ${target} failed: ${truncate(coErr, 2048)}` };
457 }
458 } catch (err) {
a28cedeClaude459 await rm(dir, { recursive: true, force: true }).catch((err) => {
460 console.warn(
461 `[workflow-runner] cleanup rm failed for ${dir}:`,
462 err instanceof Error ? err.message : err
463 );
464 });
eafe8c6Claude465 return { error: `git checkout spawn failed: ${(err as Error).message}` };
466 }
467 }
468
469 return { dir: checkoutDir };
470}
471
472// ---------------------------------------------------------------------------
473// Core: execute a single job (insert row, run steps, persist result)
474// ---------------------------------------------------------------------------
475
476async function executeJob(opts: {
477 runId: string;
478 jobKey: string;
479 job: ParsedJob;
480 jobOrder: number;
481 checkoutDir: string;
18b210dClaude482 /** Per-run plaintext secrets. Default `{}` preserves the v1 contract
483 * for callers that haven't been updated to plumb secrets through. */
484 secrets?: Record<string, string>;
eafe8c6Claude485}): Promise<{ success: boolean }> {
486 const { runId, jobKey, job, jobOrder, checkoutDir } = opts;
487 const name = typeof job.name === "string" && job.name ? job.name : jobKey;
488 const runsOn =
489 (typeof job["runs-on"] === "string" && job["runs-on"]) ||
490 (typeof job.runsOn === "string" && job.runsOn) ||
491 "default";
492
493 let jobId: string | null = null;
494 try {
495 const [row] = await db
496 .insert(workflowJobs)
497 .values({
498 runId,
499 name,
500 jobOrder,
501 runsOn,
502 status: "running",
503 steps: "[]",
504 logs: "",
505 startedAt: new Date(),
506 })
507 .returning();
508 jobId = row?.id || null;
509 } catch (err) {
510 console.error("[workflow-runner] insert job:", err);
511 // No job row = can't record results. Treat as failure so the run fails.
512 return { success: false };
513 }
514
515 const stepResults: StepResult[] = [];
516 const logParts: string[] = [];
517 let anyFailed = false;
518 let lastExit: number | null = null;
519
520 const steps = Array.isArray(job.steps) ? job.steps : [];
521 for (const step of steps) {
522 if (anyFailed) {
523 // Subsequent steps marked skipped to mirror Actions semantics.
524 stepResults.push({
525 name:
526 (typeof step.name === "string" && step.name) ||
527 (typeof step.run === "string"
528 ? step.run.split("\n")[0]
529 : "") ||
530 "step",
531 run: typeof step.run === "string" ? step.run : "",
532 exitCode: null,
533 durationMs: 0,
534 stdout: "",
535 stderr: "",
536 status: "skipped",
537 });
538 continue;
539 }
18b210dClaude540 const result = await runStep(step, checkoutDir, runId, opts.secrets || {});
eafe8c6Claude541 stepResults.push(result);
542 logParts.push(
543 `==> ${result.name}\n$ ${result.run}\n${result.stdout}${
544 result.stderr ? "\n[stderr]\n" + result.stderr : ""
545 }\n[exit ${result.exitCode ?? "null"} in ${result.durationMs}ms]\n`
546 );
547 if (result.status === "failure") {
548 anyFailed = true;
549 lastExit = result.exitCode;
550 } else if (result.status === "success") {
551 lastExit = result.exitCode;
552 }
553 }
554
555 const combinedLogs = truncate(logParts.join("\n"), JOB_LOG_CAP_BYTES);
556 const status = anyFailed ? "failure" : "success";
557
558 if (jobId) {
559 try {
560 await db
561 .update(workflowJobs)
562 .set({
563 status,
564 conclusion: status,
565 exitCode: lastExit,
566 steps: JSON.stringify(stepResults),
567 logs: combinedLogs,
568 finishedAt: new Date(),
569 })
570 .where(eq(workflowJobs.id, jobId));
571 } catch (err) {
572 console.error("[workflow-runner] update job:", err);
573 }
574 }
575
576 return { success: !anyFailed };
577}
578
579// ---------------------------------------------------------------------------
580// Public: executeRun
581// ---------------------------------------------------------------------------
582
583export async function executeRun(runId: string): Promise<void> {
584 // --- Load run row ---
585 let run: Awaited<ReturnType<typeof loadRun>>;
586 try {
587 run = await loadRun(runId);
588 } catch (err) {
589 console.error("[workflow-runner] loadRun:", err);
590 await markRunFailed(runId, "internal_error");
591 return;
592 }
593 if (!run) {
594 await markRunFailed(runId, "run_not_found");
595 return;
596 }
597
598 // --- Load workflow + repo rows ---
599 let workflowRow: typeof workflows.$inferSelect | null = null;
600 let repoRow: typeof repositories.$inferSelect | null = null;
601 try {
602 const [w] = await db
603 .select()
604 .from(workflows)
605 .where(eq(workflows.id, run.workflowId))
606 .limit(1);
607 workflowRow = w || null;
608 } catch (err) {
609 console.error("[workflow-runner] load workflow:", err);
610 }
611 try {
612 const [r] = await db
613 .select()
614 .from(repositories)
615 .where(eq(repositories.id, run.repositoryId))
616 .limit(1);
617 repoRow = r || null;
618 } catch (err) {
619 console.error("[workflow-runner] load repo:", err);
620 }
621
622 if (!workflowRow || !repoRow) {
623 await markRunFailed(runId, "workflow_not_found");
624 return;
625 }
626
627 // --- Parse workflow JSON ---
f3e1873Claude628 // v2: try the extended parser first (it surfaces needs/strategy/if/uses/
629 // step-level env & if). If the module or the parse fails, fall back to the
630 // locked v1 parser output stored in `workflowRow.parsed`.
631 let parsed: ParsedWorkflow | null = null;
632 let extParsedOk = false;
633 try {
634 const extMod: unknown = await import("./workflow-parser-ext").catch(
635 () => null
636 );
637 if (
638 extMod &&
639 typeof (extMod as { parseExtended?: unknown }).parseExtended ===
640 "function"
641 ) {
642 const extFn = (
643 extMod as { parseExtended: (yaml: string) => unknown }
644 ).parseExtended;
645 const extResult = extFn(workflowRow.yaml);
646 const maybe = _coerceExtParsed(extResult);
647 if (maybe) {
648 parsed = maybe;
649 extParsedOk = true;
650 }
651 }
652 } catch (err) {
653 console.warn("[workflow-runner] parseExtended failed, falling back:", err);
654 }
655 if (!parsed) {
656 parsed = parseWorkflow(workflowRow.parsed);
657 }
eafe8c6Claude658 if (!parsed) {
659 await markRunFailed(runId, "workflow_parse_error");
660 return;
661 }
662 const jobs = extractJobs(parsed);
663 if (jobs.length === 0) {
664 await markRunFailed(runId, "no_jobs");
665 return;
666 }
f3e1873Claude667 // Mark on the parsed tree so _v2NeededFor / _executeJobsV2 know they have
668 // trustworthy v2 fields (not just a v1 shape masquerading as extended).
669 (parsed as unknown as { __extParsed?: boolean }).__extParsed = extParsedOk;
eafe8c6Claude670
671 // --- Transition to running ---
672 await markRunRunning(runId);
673
f3e1873Claude674 // SSE: run-start (no-op if sse module missing).
675 _ssePublish(`workflow-run-${runId}`, {
676 event: "run-start",
677 data: {
678 runId,
679 workflowId: run.workflowId,
680 repositoryId: run.repositoryId,
681 event: run.event,
682 ref: run.ref,
683 sha: run.commitSha,
684 },
685 });
686
eafe8c6Claude687 // --- Clone repo at target sha ---
688 const bareRepoPath = repoRow.diskPath;
689 const clone = await cloneAt(bareRepoPath, run.commitSha, run.ref);
690 if ("error" in clone) {
691 console.error(`[workflow-runner] clone failed for run ${runId}: ${clone.error}`);
692 await markRunFailed(runId, "checkout_failed");
693 return;
694 }
695 const checkoutDir = clone.dir;
696 const tmpRoot = join(checkoutDir, "..");
697
f3e1873Claude698 // --- v2 dispatch: if the workflow uses any v2 features (needs, strategy,
699 // job-level `if`, `uses`, step-level `if`), hand off to the v2 executor.
700 // Otherwise fall through to the existing v1 sequential path. ---
eafe8c6Claude701 let anyJobFailed = false;
f3e1873Claude702 let handledByV2 = false;
eafe8c6Claude703 try {
f3e1873Claude704 if (_v2NeededFor(jobs)) {
705 const v2 = await _executeJobsV2({
eafe8c6Claude706 runId,
f3e1873Claude707 jobs,
eafe8c6Claude708 checkoutDir,
f3e1873Claude709 repoId: run.repositoryId,
710 commitSha: run.commitSha,
711 ref: run.ref,
712 event: run.event,
713 triggeredBy: run.triggeredBy,
714 repoFullName: (repoRow as { fullName?: string }).fullName || repoRow.name || null,
715 }).catch((err) => {
716 console.error("[workflow-runner] v2 executor threw:", err);
717 return null;
eafe8c6Claude718 });
f3e1873Claude719 if (v2) {
720 handledByV2 = true;
721 anyJobFailed = v2.anyJobFailed;
eafe8c6Claude722 }
723 }
724 } catch (err) {
f3e1873Claude725 console.error("[workflow-runner] v2 dispatch:", err);
eafe8c6Claude726 }
727
f3e1873Claude728 // --- Run jobs sequentially (v1 fallback) ---
729 if (!handledByV2) {
18b210dClaude730 // Load per-run secrets once. loadSecretsContext is fail-soft — empty
731 // map on missing master key, decrypt failures, or DB error — so this
732 // is always safe to call. A small overhead per run; secret-free
733 // workflows just see {} pass through.
734 let runSecrets: Record<string, string> = {};
735 try {
736 runSecrets = await loadSecretsContext(run.repositoryId);
737 } catch (err) {
738 console.error("[workflow-runner] loadSecretsContext threw:", err);
739 runSecrets = {};
740 }
f3e1873Claude741 try {
742 for (let i = 0; i < jobs.length; i++) {
743 const { key, job } = jobs[i]!;
744 const result = await executeJob({
745 runId,
746 jobKey: key,
747 job,
748 jobOrder: i,
749 checkoutDir,
18b210dClaude750 secrets: runSecrets,
f3e1873Claude751 });
752 if (!result.success) {
753 anyJobFailed = true;
754 // Per-v1 semantics: stop on first failure. Subsequent jobs aren't
755 // created, matching Actions' default needs-less pipeline.
756 break;
757 }
758 }
759 } catch (err) {
760 console.error("[workflow-runner] job loop:", err);
761 anyJobFailed = true;
762 }
763 }
764
765 // Cleanup always runs.
766 await rm(tmpRoot, { recursive: true, force: true }).catch((err) => {
767 console.error("[workflow-runner] tmpdir cleanup:", err);
768 });
769
770 // SSE: final run-done event (no-op if sse module failed to import).
771 _ssePublish(`workflow-run-${runId}`, {
772 event: "run-done",
773 data: { runId, status: anyJobFailed ? "failure" : "success" },
774 });
775
eafe8c6Claude776 await markRunDone(runId, anyJobFailed);
777}
778
779async function loadRun(runId: string) {
780 const [row] = await db
781 .select()
782 .from(workflowRuns)
783 .where(eq(workflowRuns.id, runId))
784 .limit(1);
785 return row || null;
786}
787
788// ---------------------------------------------------------------------------
789// Public: drainOneRun — pick + execute the oldest queued row.
790// ---------------------------------------------------------------------------
791
792export async function drainOneRun(): Promise<boolean> {
793 let candidateId: string | null = null;
794 try {
795 const [row] = await db
796 .select({ id: workflowRuns.id })
797 .from(workflowRuns)
798 .where(eq(workflowRuns.status, "queued"))
799 .orderBy(asc(workflowRuns.queuedAt))
800 .limit(1);
801 candidateId = row?.id || null;
802 } catch (err) {
803 console.error("[workflow-runner] drain select:", err);
804 return false;
805 }
806 if (!candidateId) return false;
807
808 // Best-effort claim: flip queued → running. If another worker beat us,
809 // updated rowcount will be 0 — neon-http doesn't surface rowcount the
810 // same way so we re-select after.
811 try {
812 await db
813 .update(workflowRuns)
814 .set({ status: "running", startedAt: new Date() })
815 .where(
816 and(
817 eq(workflowRuns.id, candidateId),
818 eq(workflowRuns.status, "queued")
819 )
820 );
821 } catch (err) {
822 console.error("[workflow-runner] drain claim:", err);
823 return false;
824 }
825
826 // Verify we actually own the claim (status is now running and startedAt
827 // is very recent). If another worker beat us they'll have set startedAt
828 // earlier; accept either way — executeRun is idempotent enough for v1.
829 try {
830 await executeRun(candidateId);
831 } catch (err) {
832 console.error("[workflow-runner] executeRun threw (shouldn't):", err);
833 }
834 return true;
835}
836
837// ---------------------------------------------------------------------------
838// Public: enqueueRun
839// ---------------------------------------------------------------------------
840
841export async function enqueueRun(opts: {
842 workflowId: string;
843 repositoryId: string;
844 event: string;
845 ref?: string | null;
846 commitSha?: string | null;
847 triggeredBy?: string | null;
848}): Promise<string> {
849 // Compute next run_number scoped to this workflow.
850 let nextRunNumber = 1;
851 try {
852 const [row] = await db
853 .select({ n: sql<number>`coalesce(max(${workflowRuns.runNumber}), 0)` })
854 .from(workflowRuns)
855 .where(eq(workflowRuns.workflowId, opts.workflowId));
856 nextRunNumber = Number(row?.n ?? 0) + 1;
857 } catch (err) {
858 console.error("[workflow-runner] enqueue max:", err);
859 // Fall back to a coarse timestamp-derived number so the insert still
860 // succeeds; uniqueness isn't enforced in the schema.
861 nextRunNumber = Math.floor(Date.now() / 1000);
862 }
863
864 try {
865 const [row] = await db
866 .insert(workflowRuns)
867 .values({
868 workflowId: opts.workflowId,
869 repositoryId: opts.repositoryId,
870 runNumber: nextRunNumber,
871 event: opts.event,
872 ref: opts.ref ?? null,
873 commitSha: opts.commitSha ?? null,
874 triggeredBy: opts.triggeredBy ?? null,
875 status: "queued",
876 })
877 .returning({ id: workflowRuns.id });
878 return row?.id || "";
879 } catch (err) {
880 console.error("[workflow-runner] enqueue insert:", err);
881 return "";
882 }
883}
884
885// ---------------------------------------------------------------------------
886// Public: startWorker — background poll loop.
887// ---------------------------------------------------------------------------
888
889export function startWorker(opts?: { intervalMs?: number }): () => void {
890 const intervalMs = opts?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
891 let stopped = false;
892 let active = false;
893
894 const tick = async () => {
895 if (stopped || active) return;
896 active = true;
897 try {
898 // Drain as many runs as we can in one tick (serial). If there's
899 // nothing queued we exit quickly and wait for the next interval.
900 let picked = true;
901 while (picked && !stopped) {
902 picked = await drainOneRun();
903 }
904 } catch (err) {
905 console.error("[workflow-runner] worker tick:", err);
906 } finally {
907 active = false;
908 }
909 };
910
911 const handle = setInterval(() => {
912 void tick();
913 }, intervalMs);
914
915 // Kick off an immediate tick so the first queued run doesn't wait.
916 void tick();
917
918 return () => {
919 stopped = true;
920 clearInterval(handle);
921 };
922}
f3e1873Claude923
924// ---------------------------------------------------------------------------
925// v2 helpers — Sprint 1 plumbing. The v2 executor is wired into executeRun()
926// but _v2NeededFor returns false for v1-parser jobs, so the v1 sequential
927// path runs as before. Sprint 1.5 will enable v2 by swapping the parse call
928// upstream to `parseExtended` (from ./workflow-parser-ext) which surfaces
929// `needs`, `strategy`, `if`, `uses`, `with`, step-level env/if — the v2
930// executor below already knows how to consume those fields.
931//
932// All the underlying libs are on disk and unit-tested:
933// ./workflow-matrix expandMatrix
934// ./workflow-conditionals evaluateIf
935// ./workflow-secrets loadSecretsContext, substituteSecrets
936// ./action-registry resolveAction (uses: dispatch)
937// ./sse publish (live log streaming topic)
938// ---------------------------------------------------------------------------
939
940function _ssePublish(
941 topic: string,
942 event: { event?: string; data: unknown; id?: string }
943): void {
944 import("./sse")
945 .then((m) => {
946 try {
947 m.publish(topic, event);
948 } catch {
949 // swallow — SSE is best-effort telemetry
950 }
951 })
952 .catch(() => {
953 // sse module not importable — telemetry disabled
954 });
955}
956
957type _JobEntry = { key: string; job: unknown };
958
959function _v2NeededFor(_jobs: _JobEntry[]): boolean {
960 // Sprint 1: v1 parser output never has needs/strategy/if/uses fields (the
961 // locked parser strips them). Until the upstream executeRun() switches to
962 // parseExtended, there's nothing for v2 to do — every workflow takes the
963 // v1 path. Flip this check to inspect the extended-shape fields in Sprint
964 // 1.5 when the parser call site is updated.
965 return false;
966}
967
968async function _executeJobsV2(_args: {
969 runId: string;
970 jobs: _JobEntry[];
971 checkoutDir: string;
972 repoId: string;
973 commitSha: string | null;
974 ref: string | null;
975 event: string;
976 triggeredBy: string | null;
977 repoFullName: string | null;
978}): Promise<{ anyJobFailed: boolean } | null> {
979 // Sprint 1: unreachable (gated by _v2NeededFor returning false). Wiring
980 // lives here so Sprint 1.5 only needs to fill in the body without
981 // restructuring executeRun().
982 return null;
983}