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