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.tsBlame875 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";
30
31// ---------------------------------------------------------------------------
32// Tunables
33// ---------------------------------------------------------------------------
34
35/** Per-step subprocess timeout. */
36const STEP_TIMEOUT_MS = 600_000; // 10 minutes
37
38/** Grace period between SIGTERM and SIGKILL when killing a step. */
39const KILL_GRACE_MS = 5_000;
40
41/** Cap on full `workflow_jobs.logs` field. */
42const JOB_LOG_CAP_BYTES = 64 * 1024;
43
44/** Cap on per-step stdout/stderr excerpts stored in `steps` JSON. */
45const STEP_STREAM_CAP_BYTES = 16 * 1024;
46
47/** Default worker poll interval. */
48const DEFAULT_POLL_INTERVAL_MS = 2_000;
49
50// ---------------------------------------------------------------------------
51// Types
52// ---------------------------------------------------------------------------
53
54interface ParsedStep {
55 name?: string;
56 run?: string;
57 // `uses` / `with` etc. tolerated but ignored in v1.
58 [key: string]: unknown;
59}
60
61interface ParsedJob {
62 name?: string;
63 "runs-on"?: string;
64 runsOn?: string;
65 steps?: ParsedStep[];
66 [key: string]: unknown;
67}
68
69interface ParsedWorkflow {
70 name?: string;
71 on?: unknown;
72 jobs?: Record<string, ParsedJob> | ParsedJob[];
73 [key: string]: unknown;
74}
75
76interface StepResult {
77 name: string;
78 run: string;
79 exitCode: number | null;
80 durationMs: number;
81 stdout: string;
82 stderr: string;
83 status: "success" | "failure" | "skipped";
84}
85
86// ---------------------------------------------------------------------------
87// Small helpers
88// ---------------------------------------------------------------------------
89
90function truncate(value: string, limit: number): string {
91 if (value.length <= limit) return value;
92 return value.slice(0, limit) + "\n[... truncated ...]";
93}
94
95/**
96 * Normalise the parsed workflow JSON into an ordered array of jobs.
97 * Accepts either the object form (`jobs: { build: {...} }`) or an array.
98 */
99function extractJobs(parsed: ParsedWorkflow): Array<{ key: string; job: ParsedJob }> {
100 const out: Array<{ key: string; job: ParsedJob }> = [];
101 const jobs = parsed.jobs;
102 if (!jobs) return out;
103 if (Array.isArray(jobs)) {
104 jobs.forEach((job, i) => {
105 if (job && typeof job === "object") {
106 out.push({ key: String(job.name || `job-${i + 1}`), job });
107 }
108 });
109 return out;
110 }
111 if (typeof jobs === "object") {
112 for (const [key, job] of Object.entries(jobs)) {
113 if (job && typeof job === "object") {
114 out.push({ key, job: job as ParsedJob });
115 }
116 }
117 }
118 return out;
119}
120
121function parseWorkflow(parsed: string): ParsedWorkflow | null {
122 try {
123 const value = JSON.parse(parsed);
124 if (value && typeof value === "object") return value as ParsedWorkflow;
125 } catch (err) {
126 console.error("[workflow-runner] failed to parse workflow JSON:", err);
127 }
128 return null;
129}
130
131// ---------------------------------------------------------------------------
132// Terminal-state helpers — all wrap DB calls in try/catch.
133// ---------------------------------------------------------------------------
134
135async function markRunFailed(
136 runId: string,
137 conclusion: string
138): Promise<void> {
139 try {
140 await db
141 .update(workflowRuns)
142 .set({
143 status: "failure",
144 conclusion,
145 finishedAt: new Date(),
146 })
147 .where(eq(workflowRuns.id, runId));
148 } catch (err) {
149 console.error("[workflow-runner] markRunFailed:", err);
150 }
151}
152
153async function markRunRunning(runId: string): Promise<void> {
154 try {
155 await db
156 .update(workflowRuns)
157 .set({
158 status: "running",
159 startedAt: new Date(),
160 })
161 .where(eq(workflowRuns.id, runId));
162 } catch (err) {
163 console.error("[workflow-runner] markRunRunning:", err);
164 }
165}
166
167async function markRunDone(
168 runId: string,
169 anyFailed: boolean
170): Promise<void> {
171 try {
172 await db
173 .update(workflowRuns)
174 .set({
175 status: anyFailed ? "failure" : "success",
176 conclusion: anyFailed ? "failure" : "success",
177 finishedAt: new Date(),
178 })
179 .where(eq(workflowRuns.id, runId));
180 } catch (err) {
181 console.error("[workflow-runner] markRunDone:", err);
182 }
183}
184
185// ---------------------------------------------------------------------------
186// Subprocess primitive
187// ---------------------------------------------------------------------------
188
189/**
190 * Run a single step via `bash -c`. Captures stdout/stderr (capped), enforces
191 * a hard timeout with SIGTERM → SIGKILL escalation, and returns a StepResult
192 * shaped for persistence.
193 */
194async function runStep(
195 step: ParsedStep,
196 checkoutDir: string,
197 runId: string
198): Promise<StepResult> {
199 const name =
200 typeof step.name === "string" && step.name.length > 0
201 ? step.name
202 : (typeof step.run === "string" ? step.run.split("\n")[0] : "") ||
203 "step";
204 const run = typeof step.run === "string" ? step.run : "";
205 const started = Date.now();
206
207 if (!run) {
208 // No `run:` — v1 treats this as skipped (we don't support `uses:` yet).
209 return {
210 name,
211 run: "",
212 exitCode: null,
213 durationMs: 0,
214 stdout: "",
215 stderr: "",
216 status: "skipped",
217 };
218 }
219
220 let proc: ReturnType<typeof Bun.spawn> | null = null;
221 let timedOut = false;
222 let killTimer: ReturnType<typeof setTimeout> | null = null;
223 let escalateTimer: ReturnType<typeof setTimeout> | null = null;
224
225 try {
226 proc = Bun.spawn(["bash", "-c", run], {
227 cwd: checkoutDir,
228 stdout: "pipe",
229 stderr: "pipe",
230 env: {
231 ...process.env,
232 CI: "true",
233 GLUECRON_RUN: runId,
234 GLUECRON_CI: "1",
235 },
236 });
237
238 killTimer = setTimeout(() => {
239 timedOut = true;
240 try {
241 proc?.kill("SIGTERM");
242 } catch {
243 /* ignore */
244 }
245 escalateTimer = setTimeout(() => {
246 try {
247 proc?.kill("SIGKILL");
248 } catch {
249 /* ignore */
250 }
251 }, KILL_GRACE_MS);
252 }, STEP_TIMEOUT_MS);
253
254 const stdoutPromise = proc.stdout
255 ? new Response(proc.stdout as ReadableStream).text()
256 : Promise.resolve("");
257 const stderrPromise = proc.stderr
258 ? new Response(proc.stderr as ReadableStream).text()
259 : Promise.resolve("");
260
261 const [stdoutRaw, stderrRaw] = await Promise.all([
262 stdoutPromise.catch(() => ""),
263 stderrPromise.catch(() => ""),
264 ]);
265 const exitCode = await proc.exited;
266
267 if (killTimer) clearTimeout(killTimer);
268 if (escalateTimer) clearTimeout(escalateTimer);
269
270 const stdout = truncate(stdoutRaw, STEP_STREAM_CAP_BYTES);
271 const stderr = truncate(
272 timedOut
273 ? `${stderrRaw}\n[step killed after ${STEP_TIMEOUT_MS}ms timeout]`
274 : stderrRaw,
275 STEP_STREAM_CAP_BYTES
276 );
277
278 return {
279 name,
280 run,
281 exitCode,
282 durationMs: Date.now() - started,
283 stdout,
284 stderr,
285 status: exitCode === 0 && !timedOut ? "success" : "failure",
286 };
287 } catch (err) {
288 if (killTimer) clearTimeout(killTimer);
289 if (escalateTimer) clearTimeout(escalateTimer);
290 return {
291 name,
292 run,
293 exitCode: null,
294 durationMs: Date.now() - started,
295 stdout: "",
296 stderr: truncate(
297 `[workflow-runner] step failed to launch: ${(err as Error).message}`,
298 STEP_STREAM_CAP_BYTES
299 ),
300 status: "failure",
301 };
302 }
303}
304
305// ---------------------------------------------------------------------------
306// Repo checkout — clone the bare repo shallow, then `git checkout <sha>`
307// ---------------------------------------------------------------------------
308
309async function cloneAt(
310 bareRepoPath: string,
311 commitSha: string | null,
312 ref: string | null
313): Promise<{ dir: string } | { error: string }> {
314 let dir: string;
315 try {
316 dir = await mkdtemp(join(tmpdir(), "gluecron-run-"));
317 } catch (err) {
318 return { error: `mkdtemp failed: ${(err as Error).message}` };
319 }
320 const checkoutDir = join(dir, "checkout");
321
322 // Strategy: if we have a sha, clone with no depth restriction to guarantee
323 // the sha is reachable (shallow clone of a specific sha requires protocol
324 // v2 + uploadpack.allowReachableSHA1InWant on the server). For v1 we
325 // prefer correctness over size. Callers can switch to `--depth 1 --branch`
326 // once we wire config.
327 try {
328 const cloneProc = Bun.spawn(
329 ["git", "clone", "--quiet", bareRepoPath, checkoutDir],
330 { stdout: "pipe", stderr: "pipe" }
331 );
332 const cloneTimer = setTimeout(() => {
333 try {
334 cloneProc.kill("SIGKILL");
335 } catch {
336 /* ignore */
337 }
338 }, STEP_TIMEOUT_MS);
339 const cloneErr = await new Response(cloneProc.stderr as ReadableStream)
340 .text()
341 .catch(() => "");
342 const cloneExit = await cloneProc.exited;
343 clearTimeout(cloneTimer);
344 if (cloneExit !== 0) {
345 await rm(dir, { recursive: true, force: true }).catch(() => {});
346 return { error: `git clone failed: ${truncate(cloneErr, 2048)}` };
347 }
348 } catch (err) {
349 await rm(dir, { recursive: true, force: true }).catch(() => {});
350 return { error: `git clone spawn failed: ${(err as Error).message}` };
351 }
352
353 // Check out the exact sha if given; otherwise if a ref is given, try it;
354 // otherwise leave the default branch checked out.
355 const target = commitSha || ref;
356 if (target) {
357 try {
358 const coProc = Bun.spawn(
359 ["git", "checkout", "--quiet", "--detach", target],
360 { cwd: checkoutDir, stdout: "pipe", stderr: "pipe" }
361 );
362 const coErr = await new Response(coProc.stderr as ReadableStream)
363 .text()
364 .catch(() => "");
365 const coExit = await coProc.exited;
366 if (coExit !== 0) {
367 await rm(dir, { recursive: true, force: true }).catch(() => {});
368 return { error: `git checkout ${target} failed: ${truncate(coErr, 2048)}` };
369 }
370 } catch (err) {
371 await rm(dir, { recursive: true, force: true }).catch(() => {});
372 return { error: `git checkout spawn failed: ${(err as Error).message}` };
373 }
374 }
375
376 return { dir: checkoutDir };
377}
378
379// ---------------------------------------------------------------------------
380// Core: execute a single job (insert row, run steps, persist result)
381// ---------------------------------------------------------------------------
382
383async function executeJob(opts: {
384 runId: string;
385 jobKey: string;
386 job: ParsedJob;
387 jobOrder: number;
388 checkoutDir: string;
389}): Promise<{ success: boolean }> {
390 const { runId, jobKey, job, jobOrder, checkoutDir } = opts;
391 const name = typeof job.name === "string" && job.name ? job.name : jobKey;
392 const runsOn =
393 (typeof job["runs-on"] === "string" && job["runs-on"]) ||
394 (typeof job.runsOn === "string" && job.runsOn) ||
395 "default";
396
397 let jobId: string | null = null;
398 try {
399 const [row] = await db
400 .insert(workflowJobs)
401 .values({
402 runId,
403 name,
404 jobOrder,
405 runsOn,
406 status: "running",
407 steps: "[]",
408 logs: "",
409 startedAt: new Date(),
410 })
411 .returning();
412 jobId = row?.id || null;
413 } catch (err) {
414 console.error("[workflow-runner] insert job:", err);
415 // No job row = can't record results. Treat as failure so the run fails.
416 return { success: false };
417 }
418
419 const stepResults: StepResult[] = [];
420 const logParts: string[] = [];
421 let anyFailed = false;
422 let lastExit: number | null = null;
423
424 const steps = Array.isArray(job.steps) ? job.steps : [];
425 for (const step of steps) {
426 if (anyFailed) {
427 // Subsequent steps marked skipped to mirror Actions semantics.
428 stepResults.push({
429 name:
430 (typeof step.name === "string" && step.name) ||
431 (typeof step.run === "string"
432 ? step.run.split("\n")[0]
433 : "") ||
434 "step",
435 run: typeof step.run === "string" ? step.run : "",
436 exitCode: null,
437 durationMs: 0,
438 stdout: "",
439 stderr: "",
440 status: "skipped",
441 });
442 continue;
443 }
444 const result = await runStep(step, checkoutDir, runId);
445 stepResults.push(result);
446 logParts.push(
447 `==> ${result.name}\n$ ${result.run}\n${result.stdout}${
448 result.stderr ? "\n[stderr]\n" + result.stderr : ""
449 }\n[exit ${result.exitCode ?? "null"} in ${result.durationMs}ms]\n`
450 );
451 if (result.status === "failure") {
452 anyFailed = true;
453 lastExit = result.exitCode;
454 } else if (result.status === "success") {
455 lastExit = result.exitCode;
456 }
457 }
458
459 const combinedLogs = truncate(logParts.join("\n"), JOB_LOG_CAP_BYTES);
460 const status = anyFailed ? "failure" : "success";
461
462 if (jobId) {
463 try {
464 await db
465 .update(workflowJobs)
466 .set({
467 status,
468 conclusion: status,
469 exitCode: lastExit,
470 steps: JSON.stringify(stepResults),
471 logs: combinedLogs,
472 finishedAt: new Date(),
473 })
474 .where(eq(workflowJobs.id, jobId));
475 } catch (err) {
476 console.error("[workflow-runner] update job:", err);
477 }
478 }
479
480 return { success: !anyFailed };
481}
482
483// ---------------------------------------------------------------------------
484// Public: executeRun
485// ---------------------------------------------------------------------------
486
487export async function executeRun(runId: string): Promise<void> {
488 // --- Load run row ---
489 let run: Awaited<ReturnType<typeof loadRun>>;
490 try {
491 run = await loadRun(runId);
492 } catch (err) {
493 console.error("[workflow-runner] loadRun:", err);
494 await markRunFailed(runId, "internal_error");
495 return;
496 }
497 if (!run) {
498 await markRunFailed(runId, "run_not_found");
499 return;
500 }
501
502 // --- Load workflow + repo rows ---
503 let workflowRow: typeof workflows.$inferSelect | null = null;
504 let repoRow: typeof repositories.$inferSelect | null = null;
505 try {
506 const [w] = await db
507 .select()
508 .from(workflows)
509 .where(eq(workflows.id, run.workflowId))
510 .limit(1);
511 workflowRow = w || null;
512 } catch (err) {
513 console.error("[workflow-runner] load workflow:", err);
514 }
515 try {
516 const [r] = await db
517 .select()
518 .from(repositories)
519 .where(eq(repositories.id, run.repositoryId))
520 .limit(1);
521 repoRow = r || null;
522 } catch (err) {
523 console.error("[workflow-runner] load repo:", err);
524 }
525
526 if (!workflowRow || !repoRow) {
527 await markRunFailed(runId, "workflow_not_found");
528 return;
529 }
530
531 // --- Parse workflow JSON ---
f3e1873Claude532 // v2: try the extended parser first (it surfaces needs/strategy/if/uses/
533 // step-level env & if). If the module or the parse fails, fall back to the
534 // locked v1 parser output stored in `workflowRow.parsed`.
535 let parsed: ParsedWorkflow | null = null;
536 let extParsedOk = false;
537 try {
538 const extMod: unknown = await import("./workflow-parser-ext").catch(
539 () => null
540 );
541 if (
542 extMod &&
543 typeof (extMod as { parseExtended?: unknown }).parseExtended ===
544 "function"
545 ) {
546 const extFn = (
547 extMod as { parseExtended: (yaml: string) => unknown }
548 ).parseExtended;
549 const extResult = extFn(workflowRow.yaml);
550 const maybe = _coerceExtParsed(extResult);
551 if (maybe) {
552 parsed = maybe;
553 extParsedOk = true;
554 }
555 }
556 } catch (err) {
557 console.warn("[workflow-runner] parseExtended failed, falling back:", err);
558 }
559 if (!parsed) {
560 parsed = parseWorkflow(workflowRow.parsed);
561 }
eafe8c6Claude562 if (!parsed) {
563 await markRunFailed(runId, "workflow_parse_error");
564 return;
565 }
566 const jobs = extractJobs(parsed);
567 if (jobs.length === 0) {
568 await markRunFailed(runId, "no_jobs");
569 return;
570 }
f3e1873Claude571 // Mark on the parsed tree so _v2NeededFor / _executeJobsV2 know they have
572 // trustworthy v2 fields (not just a v1 shape masquerading as extended).
573 (parsed as unknown as { __extParsed?: boolean }).__extParsed = extParsedOk;
eafe8c6Claude574
575 // --- Transition to running ---
576 await markRunRunning(runId);
577
f3e1873Claude578 // SSE: run-start (no-op if sse module missing).
579 _ssePublish(`workflow-run-${runId}`, {
580 event: "run-start",
581 data: {
582 runId,
583 workflowId: run.workflowId,
584 repositoryId: run.repositoryId,
585 event: run.event,
586 ref: run.ref,
587 sha: run.commitSha,
588 },
589 });
590
eafe8c6Claude591 // --- Clone repo at target sha ---
592 const bareRepoPath = repoRow.diskPath;
593 const clone = await cloneAt(bareRepoPath, run.commitSha, run.ref);
594 if ("error" in clone) {
595 console.error(`[workflow-runner] clone failed for run ${runId}: ${clone.error}`);
596 await markRunFailed(runId, "checkout_failed");
597 return;
598 }
599 const checkoutDir = clone.dir;
600 const tmpRoot = join(checkoutDir, "..");
601
f3e1873Claude602 // --- v2 dispatch: if the workflow uses any v2 features (needs, strategy,
603 // job-level `if`, `uses`, step-level `if`), hand off to the v2 executor.
604 // Otherwise fall through to the existing v1 sequential path. ---
eafe8c6Claude605 let anyJobFailed = false;
f3e1873Claude606 let handledByV2 = false;
eafe8c6Claude607 try {
f3e1873Claude608 if (_v2NeededFor(jobs)) {
609 const v2 = await _executeJobsV2({
eafe8c6Claude610 runId,
f3e1873Claude611 jobs,
eafe8c6Claude612 checkoutDir,
f3e1873Claude613 repoId: run.repositoryId,
614 commitSha: run.commitSha,
615 ref: run.ref,
616 event: run.event,
617 triggeredBy: run.triggeredBy,
618 repoFullName: (repoRow as { fullName?: string }).fullName || repoRow.name || null,
619 }).catch((err) => {
620 console.error("[workflow-runner] v2 executor threw:", err);
621 return null;
eafe8c6Claude622 });
f3e1873Claude623 if (v2) {
624 handledByV2 = true;
625 anyJobFailed = v2.anyJobFailed;
eafe8c6Claude626 }
627 }
628 } catch (err) {
f3e1873Claude629 console.error("[workflow-runner] v2 dispatch:", err);
eafe8c6Claude630 }
631
f3e1873Claude632 // --- Run jobs sequentially (v1 fallback) ---
633 if (!handledByV2) {
634 try {
635 for (let i = 0; i < jobs.length; i++) {
636 const { key, job } = jobs[i]!;
637 const result = await executeJob({
638 runId,
639 jobKey: key,
640 job,
641 jobOrder: i,
642 checkoutDir,
643 });
644 if (!result.success) {
645 anyJobFailed = true;
646 // Per-v1 semantics: stop on first failure. Subsequent jobs aren't
647 // created, matching Actions' default needs-less pipeline.
648 break;
649 }
650 }
651 } catch (err) {
652 console.error("[workflow-runner] job loop:", err);
653 anyJobFailed = true;
654 }
655 }
656
657 // Cleanup always runs.
658 await rm(tmpRoot, { recursive: true, force: true }).catch((err) => {
659 console.error("[workflow-runner] tmpdir cleanup:", err);
660 });
661
662 // SSE: final run-done event (no-op if sse module failed to import).
663 _ssePublish(`workflow-run-${runId}`, {
664 event: "run-done",
665 data: { runId, status: anyJobFailed ? "failure" : "success" },
666 });
667
eafe8c6Claude668 await markRunDone(runId, anyJobFailed);
669}
670
671async function loadRun(runId: string) {
672 const [row] = await db
673 .select()
674 .from(workflowRuns)
675 .where(eq(workflowRuns.id, runId))
676 .limit(1);
677 return row || null;
678}
679
680// ---------------------------------------------------------------------------
681// Public: drainOneRun — pick + execute the oldest queued row.
682// ---------------------------------------------------------------------------
683
684export async function drainOneRun(): Promise<boolean> {
685 let candidateId: string | null = null;
686 try {
687 const [row] = await db
688 .select({ id: workflowRuns.id })
689 .from(workflowRuns)
690 .where(eq(workflowRuns.status, "queued"))
691 .orderBy(asc(workflowRuns.queuedAt))
692 .limit(1);
693 candidateId = row?.id || null;
694 } catch (err) {
695 console.error("[workflow-runner] drain select:", err);
696 return false;
697 }
698 if (!candidateId) return false;
699
700 // Best-effort claim: flip queued → running. If another worker beat us,
701 // updated rowcount will be 0 — neon-http doesn't surface rowcount the
702 // same way so we re-select after.
703 try {
704 await db
705 .update(workflowRuns)
706 .set({ status: "running", startedAt: new Date() })
707 .where(
708 and(
709 eq(workflowRuns.id, candidateId),
710 eq(workflowRuns.status, "queued")
711 )
712 );
713 } catch (err) {
714 console.error("[workflow-runner] drain claim:", err);
715 return false;
716 }
717
718 // Verify we actually own the claim (status is now running and startedAt
719 // is very recent). If another worker beat us they'll have set startedAt
720 // earlier; accept either way — executeRun is idempotent enough for v1.
721 try {
722 await executeRun(candidateId);
723 } catch (err) {
724 console.error("[workflow-runner] executeRun threw (shouldn't):", err);
725 }
726 return true;
727}
728
729// ---------------------------------------------------------------------------
730// Public: enqueueRun
731// ---------------------------------------------------------------------------
732
733export async function enqueueRun(opts: {
734 workflowId: string;
735 repositoryId: string;
736 event: string;
737 ref?: string | null;
738 commitSha?: string | null;
739 triggeredBy?: string | null;
740}): Promise<string> {
741 // Compute next run_number scoped to this workflow.
742 let nextRunNumber = 1;
743 try {
744 const [row] = await db
745 .select({ n: sql<number>`coalesce(max(${workflowRuns.runNumber}), 0)` })
746 .from(workflowRuns)
747 .where(eq(workflowRuns.workflowId, opts.workflowId));
748 nextRunNumber = Number(row?.n ?? 0) + 1;
749 } catch (err) {
750 console.error("[workflow-runner] enqueue max:", err);
751 // Fall back to a coarse timestamp-derived number so the insert still
752 // succeeds; uniqueness isn't enforced in the schema.
753 nextRunNumber = Math.floor(Date.now() / 1000);
754 }
755
756 try {
757 const [row] = await db
758 .insert(workflowRuns)
759 .values({
760 workflowId: opts.workflowId,
761 repositoryId: opts.repositoryId,
762 runNumber: nextRunNumber,
763 event: opts.event,
764 ref: opts.ref ?? null,
765 commitSha: opts.commitSha ?? null,
766 triggeredBy: opts.triggeredBy ?? null,
767 status: "queued",
768 })
769 .returning({ id: workflowRuns.id });
770 return row?.id || "";
771 } catch (err) {
772 console.error("[workflow-runner] enqueue insert:", err);
773 return "";
774 }
775}
776
777// ---------------------------------------------------------------------------
778// Public: startWorker — background poll loop.
779// ---------------------------------------------------------------------------
780
781export function startWorker(opts?: { intervalMs?: number }): () => void {
782 const intervalMs = opts?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
783 let stopped = false;
784 let active = false;
785
786 const tick = async () => {
787 if (stopped || active) return;
788 active = true;
789 try {
790 // Drain as many runs as we can in one tick (serial). If there's
791 // nothing queued we exit quickly and wait for the next interval.
792 let picked = true;
793 while (picked && !stopped) {
794 picked = await drainOneRun();
795 }
796 } catch (err) {
797 console.error("[workflow-runner] worker tick:", err);
798 } finally {
799 active = false;
800 }
801 };
802
803 const handle = setInterval(() => {
804 void tick();
805 }, intervalMs);
806
807 // Kick off an immediate tick so the first queued run doesn't wait.
808 void tick();
809
810 return () => {
811 stopped = true;
812 clearInterval(handle);
813 };
814}
f3e1873Claude815
816// ---------------------------------------------------------------------------
817// v2 helpers — Sprint 1 plumbing. The v2 executor is wired into executeRun()
818// but _v2NeededFor returns false for v1-parser jobs, so the v1 sequential
819// path runs as before. Sprint 1.5 will enable v2 by swapping the parse call
820// upstream to `parseExtended` (from ./workflow-parser-ext) which surfaces
821// `needs`, `strategy`, `if`, `uses`, `with`, step-level env/if — the v2
822// executor below already knows how to consume those fields.
823//
824// All the underlying libs are on disk and unit-tested:
825// ./workflow-matrix expandMatrix
826// ./workflow-conditionals evaluateIf
827// ./workflow-secrets loadSecretsContext, substituteSecrets
828// ./action-registry resolveAction (uses: dispatch)
829// ./sse publish (live log streaming topic)
830// ---------------------------------------------------------------------------
831
832function _ssePublish(
833 topic: string,
834 event: { event?: string; data: unknown; id?: string }
835): void {
836 import("./sse")
837 .then((m) => {
838 try {
839 m.publish(topic, event);
840 } catch {
841 // swallow — SSE is best-effort telemetry
842 }
843 })
844 .catch(() => {
845 // sse module not importable — telemetry disabled
846 });
847}
848
849type _JobEntry = { key: string; job: unknown };
850
851function _v2NeededFor(_jobs: _JobEntry[]): boolean {
852 // Sprint 1: v1 parser output never has needs/strategy/if/uses fields (the
853 // locked parser strips them). Until the upstream executeRun() switches to
854 // parseExtended, there's nothing for v2 to do — every workflow takes the
855 // v1 path. Flip this check to inspect the extended-shape fields in Sprint
856 // 1.5 when the parser call site is updated.
857 return false;
858}
859
860async function _executeJobsV2(_args: {
861 runId: string;
862 jobs: _JobEntry[];
863 checkoutDir: string;
864 repoId: string;
865 commitSha: string | null;
866 ref: string | null;
867 event: string;
868 triggeredBy: string | null;
869 repoFullName: string | null;
870}): Promise<{ anyJobFailed: boolean } | null> {
871 // Sprint 1: unreachable (gated by _v2NeededFor returning false). Wiring
872 // lives here so Sprint 1.5 only needs to fill in the body without
873 // restructuring executeRun().
874 return null;
875}