hearthforge

hearthforge/src/services/ci.ts
ci.ts
⎇
Raw
1import { existsSync, mkdirSync, writeFileSync } from "node:fs";
2import path from "node:path";
3import { parse as parseToml } from "smol-toml";
4import config from "../config.ts";
5import { paths } from "../constants.ts";
6import { db } from "../db/index.ts";
7import { repoPath } from "./git.ts";
8
9// --- Types ---
10
11interface CiVariableDef {
12 default?: string;
13 description?: string;
14}
15
16interface CiStepConfig {
17 run_sh?: string;
18 run_if?: string;
19 clear?: boolean;
20 timeout?: number;
21 publish_file?: string | string[];
22 publish_tar?: string | string[];
23 publish_gzip?: string | string[];
24 publish_zip?: string | string[];
25 publish_zstd?: string | string[];
26}
27
28export interface CiStep extends CiStepConfig {
29 name: string;
30}
31
32export interface CiConfig {
33 image: string;
34 work_dir?: string;
35 clone_project_to?: string;
36 shell?: string[];
37 shell_setup?: string;
38 timeout?: number;
39 cpu_limit?: number;
40 memory_limit?: string;
41 cache?: string[];
42 on?: {
43 push?: string[] | boolean;
44 tag?: boolean;
45 manual?: boolean;
46 };
47 variables?: Record<string, CiVariableDef>;
48 steps: CiStep[];
49}
50
51export interface TriggerOpts {
52 triggerSource: "push" | "tag" | "manual";
53 commitSha: string;
54 commitBranch?: string;
55 commitTag?: string;
56 triggeredBy?: number;
57 variableOverrides?: Record<string, string>;
58}
59
60// Reserved TOML table names that are not steps
61const RESERVED_TABLES = new Set(["on", "variables"]);
62
63// In-memory map of running tasks for cancellation
64const runningTasks = new Map<
65 number,
66 { controller: AbortController; containerId?: string }
67>();
68
69// FIFO queue of run IDs whose DB row is `status = "queued"`. We start the
70// next one in `pumpQueue` whenever `runningTasks.size` drops below
71// `CI_MAX_CONCURRENT`. `pumpQueue` is the SOLE writer of `runningTasks.set`
72// — callers that want to start a run push to `queuedRunIds` and call
73// `pumpQueue` synchronously. This makes the size check + slot reservation
74// atomic against concurrent callers (no await between).
75const queuedRunIds: number[] = [];
76
77async function promoteToPending(runId: number): Promise<void> {
78 await db
79 .updateTable("ci_runs")
80 .set({ status: "pending" })
81 .where("id", "=", runId)
82 .execute();
83}
84
85function pumpQueue(): void {
86 while (
87 queuedRunIds.length > 0 &&
88 runningTasks.size < config.CI_MAX_CONCURRENT
89 ) {
90 const next = queuedRunIds.shift()!;
91 const controller = new AbortController();
92 runningTasks.set(next, { controller });
93 // Promote queued→pending before spawning so executeRun's failure path
94 // (which only matches pending/running) can still mark it failed if
95 // it throws very early. We await the flip inside spawnRun's wrapper
96 // so the order is: row=pending → executeRun starts → row=running.
97 spawnRun(next, controller.signal, /* needsPromote */ true);
98 }
99}
100
101/** Position (1-based) of this queued run within the queue, or null. */
102export function ciQueuePosition(runId: number): number | null {
103 const idx = queuedRunIds.indexOf(runId);
104 return idx < 0 ? null : idx + 1;
105}
106
107// --- TOML Parsing ---
108
109export function parseCiConfig(tomlStr: string): CiConfig | null {
110 let raw: Record<string, unknown>;
111 try {
112 raw = parseToml(tomlStr) as Record<string, unknown>;
113 } catch {
114 return null;
115 }
116
117 const image = raw.image;
118 if (typeof image !== "string" || !image) return null;
119
120 const steps: CiStep[] = [];
121 for (const [key, val] of Object.entries(raw)) {
122 if (RESERVED_TABLES.has(key)) continue;
123 if (typeof val !== "object" || val === null || Array.isArray(val))
124 continue;
125 // It's a table section — treat as a step
126 const stepCfg = val as Record<string, unknown>;
127 steps.push({ name: key, ...(stepCfg as CiStepConfig) });
128 }
129
130 const rawOn = raw.on as Record<string, unknown> | undefined;
131 const rawVars = raw.variables as
132 | Record<string, Record<string, unknown>>
133 | undefined;
134 const variables: Record<string, CiVariableDef> = {};
135 if (rawVars) {
136 for (const [name, def] of Object.entries(rawVars)) {
137 if (typeof def === "object" && def !== null) {
138 variables[name] = {
139 default:
140 typeof def.default === "string"
141 ? def.default
142 : undefined,
143 description:
144 typeof def.description === "string"
145 ? def.description
146 : undefined,
147 };
148 }
149 }
150 }
151
152 return {
153 image,
154 work_dir: typeof raw.work_dir === "string" ? raw.work_dir : undefined,
155 clone_project_to:
156 typeof raw.clone_project_to === "string"
157 ? raw.clone_project_to
158 : undefined,
159 shell: Array.isArray(raw.shell) ? (raw.shell as string[]) : undefined,
160 shell_setup:
161 typeof raw.shell_setup === "string" ? raw.shell_setup : undefined,
162 timeout: typeof raw.timeout === "number" ? raw.timeout : undefined,
163 cpu_limit:
164 typeof raw.cpu_limit === "number" ? raw.cpu_limit : undefined,
165 memory_limit:
166 typeof raw.memory_limit === "string" ? raw.memory_limit : undefined,
167 cache: Array.isArray(raw.cache) ? (raw.cache as string[]) : undefined,
168 on: rawOn
169 ? {
170 push: Array.isArray(rawOn.push)
171 ? (rawOn.push as string[])
172 : typeof rawOn.push === "boolean"
173 ? rawOn.push
174 : undefined,
175 tag: typeof rawOn.tag === "boolean" ? rawOn.tag : undefined,
176 manual:
177 typeof rawOn.manual === "boolean"
178 ? rawOn.manual
179 : undefined,
180 }
181 : undefined,
182 variables,
183 steps,
184 };
185}
186
187// --- Trigger matching ---
188
189export function shouldTriggerPush(cfg: CiConfig, branch: string): boolean {
190 const pushCfg = cfg.on?.push;
191 if (!pushCfg) return false;
192 if (pushCfg === true) return true;
193 if (Array.isArray(pushCfg)) {
194 return pushCfg.some((pattern) => matchGlob(pattern, branch));
195 }
196 return false;
197}
198
199export function shouldTriggerTag(cfg: CiConfig): boolean {
200 return cfg.on?.tag === true;
201}
202
203function matchGlob(pattern: string, value: string): boolean {
204 if (pattern === "*") return true;
205 const re = new RegExp(
206 `^${pattern.replace(/[.+^${}()|[\]\\]/g, "\\$&").replace(/\*/g, ".*")}$`,
207 );
208 return re.test(value);
209}
210
211// --- Docker socket ---
212
213let resolvedSocket: string | null = null;
214
215async function getSocket(): Promise<string> {
216 if (resolvedSocket) return resolvedSocket;
217 const candidates = config.CI_DOCKER_SOCKET
218 ? [config.CI_DOCKER_SOCKET]
219 : (() => {
220 const uid = process.getuid?.();
221 return [
222 "/var/run/docker.sock",
223 "/run/podman/podman.sock",
224 ...(uid !== undefined
225 ? [`/run/user/${uid}/podman/podman.sock`]
226 : []),
227 ];
228 })();
229 for (const s of candidates) {
230 if (existsSync(s)) {
231 resolvedSocket = s;
232 return s;
233 }
234 }
235 throw new Error(
236 "No Docker/Podman socket found. Set CI_DOCKER_SOCKET env var.",
237 );
238}
239
240async function dockerFetch(
241 endpoint: string,
242 init?: RequestInit,
243): Promise<Response> {
244 const socket = await getSocket();
245 return fetch(`http://localhost/v1.47${endpoint}`, {
246 ...init,
247 unix: socket,
248 });
249}
250
251// --- Docker helpers ---
252
253function splitImageRef(image: string): { name: string; tag: string } {
254 const lastColon = image.lastIndexOf(":");
255 if (lastColon < 0) return { name: image, tag: "latest" };
256 const possibleTag = image.slice(lastColon + 1);
257 if (possibleTag.includes("/")) return { name: image, tag: "latest" };
258 return { name: image.slice(0, lastColon), tag: possibleTag };
259}
260
261async function pullImage(image: string): Promise<void> {
262 const { name, tag } = splitImageRef(image);
263 const resp = await dockerFetch(
264 `/images/create?fromImage=${encodeURIComponent(name)}&tag=${encodeURIComponent(tag)}`,
265 { method: "POST" },
266 );
267 // Consume body to completion
268 await resp.body?.cancel();
269}
270
271function parseMemoryBytes(s: string): number {
272 const m = s.match(/^(\d+(?:\.\d+)?)\s*([kmgKMG]?)b?$/);
273 if (!m) return 0;
274 const n = parseFloat(m[1] ?? "0");
275 switch ((m[2] ?? "").toLowerCase()) {
276 case "k":
277 return Math.floor(n * 1024);
278 case "m":
279 return Math.floor(n * 1024 * 1024);
280 case "g":
281 return Math.floor(n * 1024 * 1024 * 1024);
282 default:
283 return Math.floor(n);
284 }
285}
286
287async function ensureVolume(volName: string, repoName: string): Promise<void> {
288 const resp = await dockerFetch("/volumes/create", {
289 method: "POST",
290 headers: { "Content-Type": "application/json" },
291 body: JSON.stringify({
292 Name: volName,
293 Labels: { "com.hearthforge.repo": repoName },
294 }),
295 });
296 await resp.body?.cancel();
297}
298
299async function createContainer(
300 runId: number,
301 repoName: string,
302 cfg: CiConfig,
303 repoAbsPath: string,
304 envVars: string[],
305): Promise<string> {
306 const binds: string[] = [`${repoAbsPath}:/hearthforge-repo.git:ro`];
307 if (cfg.cache) {
308 for (const cachePath of cfg.cache) {
309 const volName = `hearthforge-ci-cache-${Buffer.from(`${repoName}:${cachePath}`).toString("base64url").slice(0, 24)}`;
310 await ensureVolume(volName, repoName);
311 binds.push(`${volName}:${cachePath}`);
312 }
313 }
314
315 const hostConfig: Record<string, unknown> = { Binds: binds };
316 if (cfg.cpu_limit) {
317 hostConfig.NanoCpus = Math.floor(cfg.cpu_limit * 1e9);
318 }
319 if (cfg.memory_limit) {
320 hostConfig.Memory = parseMemoryBytes(cfg.memory_limit);
321 }
322
323 const body = JSON.stringify({
324 Image: cfg.image,
325 Cmd: ["sleep", "infinity"],
326 Env: envVars,
327 WorkingDir: cfg.work_dir ?? "/",
328 HostConfig: hostConfig,
329 });
330
331 const resp = await dockerFetch(
332 `/containers/create?name=hearthforge-ci-${runId}`,
333 {
334 method: "POST",
335 headers: { "Content-Type": "application/json" },
336 body,
337 },
338 );
339 if (!resp.ok) {
340 const text = await resp.text();
341 throw new Error(`Failed to create container: ${resp.status} ${text}`);
342 }
343 const data = (await resp.json()) as { Id: string };
344 return data.Id;
345}
346
347async function startContainer(containerId: string): Promise<void> {
348 const resp = await dockerFetch(`/containers/${containerId}/start`, {
349 method: "POST",
350 });
351 if (!resp.ok && resp.status !== 304) {
352 throw new Error(`Failed to start container: ${resp.status}`);
353 }
354 await resp.body?.cancel();
355}
356
357interface ExecResult {
358 log: string;
359 exitCode: number;
360}
361
362const dec = new TextDecoder();
363
364function parseMuxFrames(buf: Uint8Array): {
365 text: string;
366 remaining: Uint8Array<ArrayBuffer>;
367} {
368 const chunks: string[] = [];
369 let i = 0;
370 while (i + 8 <= buf.length) {
371 const view = new DataView(buf.buffer, buf.byteOffset + i, 8);
372 const size = view.getUint32(4, false);
373 if (i + 8 + size > buf.length) break;
374 chunks.push(dec.decode(buf.slice(i + 8, i + 8 + size)));
375 i += 8 + size;
376 }
377 const remaining = new Uint8Array(buf.length - i);
378 if (i < buf.length) remaining.set(buf.subarray(i));
379 return { text: chunks.join(""), remaining };
380}
381
382async function execInContainer(
383 containerId: string,
384 cmd: string[],
385 workDir?: string,
386 envVars?: string[],
387 signal?: AbortSignal,
388 onPartialLog?: (log: string) => Promise<void>,
389): Promise<ExecResult> {
390 // Create exec
391 const execBody = JSON.stringify({
392 Cmd: cmd,
393 AttachStdout: true,
394 AttachStderr: true,
395 ...(workDir ? { WorkingDir: workDir } : {}),
396 ...(envVars ? { Env: envVars } : {}),
397 });
398 const createResp = await dockerFetch(`/containers/${containerId}/exec`, {
399 method: "POST",
400 headers: { "Content-Type": "application/json" },
401 body: execBody,
402 signal,
403 });
404 if (!createResp.ok) {
405 const text = await createResp.text();
406 throw new Error(`Failed to create exec: ${createResp.status} ${text}`);
407 }
408 const createData = (await createResp.json()) as { Id: string };
409 const execId = createData.Id;
410
411 // Start exec and stream output
412 const startResp = await dockerFetch(`/exec/${execId}/start`, {
413 method: "POST",
414 headers: { "Content-Type": "application/json" },
415 body: JSON.stringify({ Detach: false, Tty: false }),
416 signal,
417 });
418
419 let log = "";
420 if (onPartialLog && startResp.body) {
421 const reader = startResp.body.getReader();
422 let buf = new Uint8Array(0);
423 let lastSave = Date.now();
424 while (true) {
425 const { done, value } = await reader.read();
426 if (done) break;
427 const merged = new Uint8Array(buf.length + value.length);
428 merged.set(buf);
429 merged.set(value, buf.length);
430 buf = merged;
431 const { text, remaining } = parseMuxFrames(buf);
432 buf = remaining;
433 log += text;
434 if (Date.now() - lastSave >= 2000) {
435 await onPartialLog(log);
436 lastSave = Date.now();
437 }
438 }
439 const { text } = parseMuxFrames(buf);
440 log += text;
441 } else {
442 const bodyBytes = new Uint8Array(await startResp.arrayBuffer());
443 const { text } = parseMuxFrames(bodyBytes);
444 log = text;
445 }
446
447 // Get exit code
448 const inspectResp = await dockerFetch(`/exec/${execId}/json`);
449 const inspectData = (await inspectResp.json()) as { ExitCode: number };
450
451 return { log, exitCode: inspectData.ExitCode ?? 1 };
452}
453
454async function removeContainer(containerId: string): Promise<void> {
455 try {
456 const resp = await dockerFetch(
457 `/containers/${containerId}?force=true`,
458 { method: "DELETE" },
459 );
460 await resp.body?.cancel();
461 } catch {
462 // Best-effort cleanup
463 }
464}
465
466// --- Tar extraction ---
467
468function extractSingleFileFromTar(data: Uint8Array): Uint8Array | null {
469 if (data.length < 512) return null;
470 const dec = new TextDecoder();
471 const sizeOctal = dec
472 .decode(data.slice(124, 136))
473 .replace(/\0/g, "")
474 .trim();
475 const size = parseInt(sizeOctal, 8);
476 if (Number.isNaN(size) || size < 0) return null;
477 if (data.length < 512 + size) return null;
478 return data.slice(512, 512 + size);
479}
480
481async function copyFileFromContainer(
482 containerId: string,
483 containerPath: string,
484): Promise<Uint8Array | null> {
485 const resp = await dockerFetch(
486 `/containers/${containerId}/archive?path=${encodeURIComponent(containerPath)}`,
487 );
488 if (!resp.ok) return null;
489 const tarBytes = new Uint8Array(await resp.arrayBuffer());
490 return extractSingleFileFromTar(tarBytes);
491}
492
493// --- Secret masking ---
494
495function maskSecrets(text: string, secrets: string[]): string {
496 for (const secret of secrets) {
497 if (secret) text = text.split(secret).join("[MASKED]");
498 }
499 return text;
500}
501
502// --- Env var building ---
503
504function buildEnvVars(
505 runId: number,
506 repoName: string,
507 opts: TriggerOpts,
508 cfg: CiConfig,
509 secretValues: Array<{ name: string; value: string }>,
510): { envArray: string[]; secretValues: string[] } {
511 const vars: Record<string, string> = {
512 CI: "true",
513 CI_PIPELINE_ID: String(runId),
514 CI_REPO_NAME: repoName,
515 CI_SERVER_URL: config.BASE_URL,
516 CI_TRIGGER_SOURCE: opts.triggerSource,
517 CI_COMMIT_SHA: opts.commitSha,
518 CI_COMMIT_SHORT_SHA: opts.commitSha.slice(0, 8),
519 CI_COMMIT_BRANCH: opts.commitBranch ?? "",
520 CI_COMMIT_TAG: opts.commitTag ?? "",
521 CI_COMMIT_REF_NAME: opts.commitTag ?? opts.commitBranch ?? "",
522 };
523
524 // User-defined variable defaults
525 if (cfg.variables) {
526 for (const [name, def] of Object.entries(cfg.variables)) {
527 if (def.default !== undefined) vars[name] = def.default;
528 }
529 }
530
531 // Variable overrides from manual trigger
532 if (opts.variableOverrides) {
533 for (const [name, value] of Object.entries(opts.variableOverrides)) {
534 vars[name] = value;
535 }
536 }
537
538 // Secrets (injected but values tracked for masking)
539 const secretVals: string[] = [];
540 for (const { name, value } of secretValues) {
541 vars[name] = value;
542 secretVals.push(value);
543 }
544
545 const envArray = Object.entries(vars).map(([k, v]) => `${k}=${v}`);
546 return { envArray, secretValues: secretVals };
547}
548
549// --- Artifact collection ---
550
551async function collectArtifacts(
552 runId: number,
553 containerId: string,
554 step: CiStep,
555 _shell: string[],
556 envVars: string[],
557): Promise<void> {
558 const artifactDir = path.join(paths.CI_ARTIFACTS_DIR, String(runId));
559 mkdirSync(artifactDir, { recursive: true });
560
561 const toArray = (v: string | string[] | undefined): string[] => {
562 if (!v) return [];
563 return Array.isArray(v) ? v : [v];
564 };
565
566 // publish_file: copy directly out of container
567 for (const srcPath of toArray(step.publish_file)) {
568 const fileBytes = await copyFileFromContainer(containerId, srcPath);
569 if (fileBytes) {
570 const filename = path.basename(srcPath);
571 const destPath = path.join(artifactDir, filename);
572 writeFileSync(destPath, fileBytes);
573 const stat = Bun.file(destPath);
574 await db
575 .insertInto("ci_artifacts")
576 .values({
577 run_id: runId,
578 filename,
579 size: stat.size,
580 })
581 .execute();
582 }
583 }
584
585 // Archive commands run with `path.dirname(srcPath)` as the working
586 // directory and reference the source by its basename only, so the
587 // user-controlled path never appears as part of an interpolated shell
588 // string. Previously `publish_zip` used `sh -c "cd … && zip …"` with
589 // raw interpolation — a step author who could write the CI TOML
590 // could shell-inject through the source path. Today the only TOML
591 // author is the admin, but this removes the implicit assumption.
592 type ArchiveType = "tar" | "gzip" | "zip" | "zstd";
593 const archiveFormats: Array<{
594 type: ArchiveType;
595 paths: string[];
596 ext: string;
597 cmd: (basename: string, dst: string) => string[];
598 }> = [
599 {
600 type: "tar",
601 paths: toArray(step.publish_tar),
602 ext: ".tar",
603 cmd: (basename, dst) => ["tar", "-cf", dst, basename],
604 },
605 {
606 type: "gzip",
607 paths: toArray(step.publish_gzip),
608 ext: ".tar.gz",
609 cmd: (basename, dst) => ["tar", "-czf", dst, basename],
610 },
611 {
612 type: "zstd",
613 paths: toArray(step.publish_zstd),
614 ext: ".tar.zst",
615 cmd: (basename, dst) => ["tar", "--zstd", "-cf", dst, basename],
616 },
617 {
618 type: "zip",
619 paths: toArray(step.publish_zip),
620 ext: ".zip",
621 cmd: (basename, dst) => ["zip", "-r", dst, basename],
622 },
623 ];
624
625 let archiveIndex = 0;
626 for (const { paths: archivePaths, ext, cmd } of archiveFormats) {
627 for (const srcPath of archivePaths) {
628 archiveIndex++;
629 const tmpPath = `/tmp/hf-artifact-${runId}-${archiveIndex}${ext}`;
630 // Run with the source's parent directory as the working
631 // directory so each archive tool can reference the source
632 // by its basename — no -C, no shell.
633 const execResult = await execInContainer(
634 containerId,
635 cmd(path.basename(srcPath), tmpPath),
636 path.dirname(srcPath),
637 envVars,
638 ).catch(() => null);
639 if (!execResult || execResult.exitCode !== 0) continue;
640
641 // Copy archive out
642 const fileBytes = await copyFileFromContainer(containerId, tmpPath);
643 if (!fileBytes) continue;
644
645 const filename = `${path.basename(srcPath)}${ext}`;
646 const destPath = path.join(artifactDir, filename);
647 writeFileSync(destPath, fileBytes);
648 const stat = Bun.file(destPath);
649 await db
650 .insertInto("ci_artifacts")
651 .values({
652 run_id: runId,
653 filename,
654 size: stat.size,
655 })
656 .execute();
657 }
658 }
659}
660
661// --- Main execution ---
662
663async function executeRun(runId: number, signal: AbortSignal): Promise<void> {
664 const now = () => new Date().toISOString();
665 let containerId: string | undefined;
666
667 try {
668 // Mark as running
669 await db
670 .updateTable("ci_runs")
671 .set({ status: "running", started_at: now() })
672 .where("id", "=", runId)
673 .execute();
674
675 // Load run details
676 const run = await db
677 .selectFrom("ci_runs")
678 .selectAll()
679 .where("id", "=", runId)
680 .executeTakeFirst();
681 if (!run) throw new Error("Run not found");
682
683 const repo = await db
684 .selectFrom("repositories")
685 .select(["id", "name"])
686 .where("id", "=", run.repo_id)
687 .executeTakeFirst();
688 if (!repo) throw new Error("Repo not found");
689
690 // Read .hearthforge-ci.toml at the commit
691 const tomlBuf = await import("./git.ts").then((g) =>
692 g.git.show(repo.name, run.commit_sha!, ".hearthforge-ci.toml"),
693 );
694 if (!tomlBuf)
695 throw new Error(".hearthforge-ci.toml not found at commit");
696
697 const cfg = parseCiConfig(tomlBuf.toString("utf-8"));
698 if (!cfg) throw new Error("Failed to parse .hearthforge-ci.toml");
699
700 // Load secrets for log masking
701 const secrets = await db
702 .selectFrom("ci_secrets")
703 .select(["name", "value"])
704 .where("repo_id", "=", repo.id)
705 .execute();
706
707 const variableOverrides = run.variable_overrides
708 ? (JSON.parse(run.variable_overrides) as Record<string, string>)
709 : {};
710
711 const { envArray, secretValues } = buildEnvVars(
712 runId,
713 repo.name,
714 {
715 triggerSource:
716 run.trigger_source as TriggerOpts["triggerSource"],
717 commitSha: run.commit_sha ?? "",
718 commitBranch: run.commit_branch ?? undefined,
719 commitTag: run.commit_tag ?? undefined,
720 variableOverrides,
721 },
722 cfg,
723 secrets,
724 );
725
726 // Create step rows in DB
727 for (const step of cfg.steps) {
728 await db
729 .insertInto("ci_steps")
730 .values({
731 run_id: runId,
732 name: step.name,
733 status: "pending",
734 })
735 .execute();
736 }
737
738 // Pull image
739 await pullImage(cfg.image);
740 if (signal.aborted) throw new Error("Cancelled");
741
742 // Create + start container
743 containerId = await createContainer(
744 runId,
745 repo.name,
746 cfg,
747 repoPath(repo.name),
748 envArray,
749 );
750 runningTasks.get(runId)!.containerId = containerId;
751
752 await startContainer(containerId);
753 if (signal.aborted) throw new Error("Cancelled");
754
755 // Create work_dir
756 if (cfg.work_dir) {
757 await execInContainer(containerId, ["mkdir", "-p", cfg.work_dir]);
758 }
759
760 // Clone project if requested
761 if (cfg.clone_project_to && run.commit_sha) {
762 await execInContainer(
763 containerId,
764 [
765 "sh",
766 "-c",
767 `git clone /hearthforge-repo.git ${cfg.clone_project_to} && git -C ${cfg.clone_project_to} checkout --detach ${run.commit_sha}`,
768 ],
769 cfg.work_dir,
770 envArray,
771 );
772 }
773
774 // Execute steps
775 const shell = cfg.shell ?? ["/bin/sh", "-c"];
776 let runFailed = false;
777
778 for (const step of cfg.steps) {
779 if (signal.aborted) {
780 runFailed = true;
781 break;
782 }
783
784 const stepRow = await db
785 .selectFrom("ci_steps")
786 .select("id")
787 .where("run_id", "=", runId)
788 .where("name", "=", step.name)
789 .executeTakeFirst();
790 if (!stepRow) continue;
791 const stepId = stepRow.id;
792
793 // Check run_if condition
794 if (step.run_if) {
795 const { exitCode } = await execInContainer(
796 containerId,
797 [...shell, step.run_if],
798 cfg.work_dir,
799 envArray,
800 );
801 if (exitCode !== 0) {
802 await db
803 .updateTable("ci_steps")
804 .set({
805 status: "skipped",
806 started_at: now(),
807 finished_at: now(),
808 log: "Skipped: condition not met",
809 })
810 .where("id", "=", stepId)
811 .execute();
812 continue;
813 }
814 }
815
816 // Handle clear option
817 if (step.clear && cfg.clone_project_to && run.commit_sha) {
818 await execInContainer(
819 containerId,
820 [
821 "sh",
822 "-c",
823 `git -C ${cfg.clone_project_to} reset --hard ${run.commit_sha} && git -C ${cfg.clone_project_to} clean -fdx`,
824 ],
825 cfg.work_dir,
826 envArray,
827 );
828 }
829
830 await db
831 .updateTable("ci_steps")
832 .set({ status: "running", started_at: now() })
833 .where("id", "=", stepId)
834 .execute();
835
836 let stepLog = "";
837 let stepStatus: "success" | "failure" = "success";
838
839 if (step.run_sh) {
840 const command = cfg.shell_setup
841 ? `${cfg.shell_setup}\n${step.run_sh}`
842 : step.run_sh;
843
844 const stepTimeout =
845 step.timeout ?? cfg.timeout ?? config.CI_DEFAULT_TIMEOUT;
846 const timeoutSignal = AbortSignal.timeout(stepTimeout * 1000);
847
848 try {
849 const { log, exitCode } = await execInContainer(
850 containerId,
851 [...shell, command],
852 cfg.work_dir,
853 envArray,
854 timeoutSignal,
855 async (partial) => {
856 await db
857 .updateTable("ci_steps")
858 .set({
859 log: maskSecrets(partial, secretValues),
860 })
861 .where("id", "=", stepId)
862 .execute();
863 },
864 );
865 stepLog = maskSecrets(log, secretValues);
866 if (exitCode !== 0) {
867 stepStatus = "failure";
868 runFailed = true;
869 }
870 } catch (err) {
871 stepLog = `Step failed: ${err instanceof Error ? err.message : String(err)}\n`;
872 stepStatus = "failure";
873 runFailed = true;
874 }
875 }
876
877 // Collect artifacts for this step
878 if (!runFailed || stepStatus === "success") {
879 await collectArtifacts(
880 runId,
881 containerId,
882 step,
883 shell,
884 envArray,
885 ).catch(() => {});
886 }
887
888 await db
889 .updateTable("ci_steps")
890 .set({
891 status: stepStatus,
892 finished_at: now(),
893 log: stepLog,
894 })
895 .where("id", "=", stepId)
896 .execute();
897
898 if (runFailed) break;
899 }
900
901 // Mark remaining steps as skipped
902 await db
903 .updateTable("ci_steps")
904 .set({
905 status: "skipped",
906 started_at: now(),
907 finished_at: now(),
908 log: "Skipped: previous step failed",
909 })
910 .where("run_id", "=", runId)
911 .where("status", "=", "pending")
912 .execute();
913
914 const finalStatus = runFailed ? "failure" : "success";
915 await db
916 .updateTable("ci_runs")
917 .set({ status: finalStatus, finished_at: now() })
918 .where("id", "=", runId)
919 .execute();
920 } catch (err) {
921 const errMsg = err instanceof Error ? err.message : String(err);
922 const isDockerUnavailable = errMsg.includes("No Docker/Podman socket");
923 const status = signal.aborted
924 ? "cancelled"
925 : isDockerUnavailable
926 ? "skipped"
927 : "failure";
928 const skipLog = signal.aborted
929 ? "Skipped: run was cancelled"
930 : isDockerUnavailable
931 ? "Skipped: Docker/Podman not available"
932 : "Skipped: run failed";
933
934 if (!isDockerUnavailable) {
935 // Write error to a synthetic step if we have no steps yet
936 const hasSteps = await db
937 .selectFrom("ci_steps")
938 .select("id")
939 .where("run_id", "=", runId)
940 .executeTakeFirst();
941 if (!hasSteps) {
942 await db
943 .insertInto("ci_steps")
944 .values({
945 run_id: runId,
946 name: "setup",
947 status: "failure",
948 started_at: new Date().toISOString(),
949 finished_at: new Date().toISOString(),
950 log: `Error: ${errMsg}\n`,
951 })
952 .execute();
953 }
954 }
955 await db
956 .updateTable("ci_runs")
957 .set({ status, finished_at: new Date().toISOString() })
958 .where("id", "=", runId)
959 .execute();
960 // Mark pending steps as skipped
961 await db
962 .updateTable("ci_steps")
963 .set({
964 status: "skipped",
965 finished_at: new Date().toISOString(),
966 log: skipLog,
967 })
968 .where("run_id", "=", runId)
969 .where("status", "=", "pending")
970 .execute();
971 } finally {
972 if (containerId) await removeContainer(containerId);
973 runningTasks.delete(runId);
974 // Prune old history
975 const run = await db
976 .selectFrom("ci_runs")
977 .select("repo_id")
978 .where("id", "=", runId)
979 .executeTakeFirst();
980 if (run) await pruneHistory(run.repo_id).catch(() => {});
981 // A slot just freed up — start the next queued run if any.
982 pumpQueue();
983 }
984}
985
986// --- Public API ---
987
988export async function triggerRun(
989 repoName: string,
990 opts: TriggerOpts,
991): Promise<number> {
992 const repo = await db
993 .selectFrom("repositories")
994 .select("id")
995 .where("name", "=", repoName)
996 .executeTakeFirst();
997 if (!repo) throw new Error("Repository not found");
998
999 const runId = await db
1000 .insertInto("ci_runs")
1001 .values({
1002 repo_id: repo.id,
1003 triggered_by: opts.triggeredBy ?? null,
1004 trigger_source: opts.triggerSource,
1005 commit_sha: opts.commitSha,
1006 commit_branch: opts.commitBranch ?? null,
1007 commit_tag: opts.commitTag ?? null,
1008 status: "pending",
1009 variable_overrides: opts.variableOverrides
1010 ? JSON.stringify(opts.variableOverrides)
1011 : null,
1012 })
1013 .returning("id")
1014 .executeTakeFirstOrThrow();
1015
1016 const countRow = await db
1017 .selectFrom("ci_runs")
1018 .select(db.fn.countAll<number>().as("c"))
1019 .where("repo_id", "=", repo.id)
1020 .executeTakeFirstOrThrow();
1021 await db
1022 .updateTable("ci_runs")
1023 .set({ repo_run_id: Number(countRow.c) })
1024 .where("id", "=", runId.id)
1025 .execute();
1026
1027 // Flip to "queued" BEFORE pushing onto queuedRunIds. Otherwise a
1028 // concurrent pumpQueue (from a finishing run) could observe our entry,
1029 // promoteToPending it, and start executeRun while our UPDATE is still
1030 // in flight — the late UPDATE would then clobber a running row back to
1031 // "queued". Only pumpQueue (the sole writer of runningTasks.set) may
1032 // mutate the row's status after the push. The TOCTOU concern is moot
1033 // here because pumpQueue cannot pump a runId that hasn't been pushed.
1034 if (runningTasks.size >= config.CI_MAX_CONCURRENT) {
1035 await db
1036 .updateTable("ci_runs")
1037 .set({ status: "queued" })
1038 .where("id", "=", runId.id)
1039 .execute();
1040 }
1041 queuedRunIds.push(runId.id);
1042 pumpQueue();
1043
1044 return runId.id;
1045}
1046
1047// Wraps the fire-and-forget executeRun so unhandled exceptions (e.g. a throw
1048// before/after its own try/finally) are logged and the run is reconciled to
1049// "failure" instead of staying pending forever.
1050function spawnRun(
1051 runId: number,
1052 signal: AbortSignal,
1053 needsPromote = false,
1054): void {
1055 void (async () => {
1056 try {
1057 if (needsPromote) await promoteToPending(runId);
1058 await executeRun(runId, signal);
1059 } catch (err) {
1060 console.error(`[ci] executeRun threw for run ${runId}:`, err);
1061 runningTasks.delete(runId);
1062 try {
1063 await db
1064 .updateTable("ci_runs")
1065 .set({
1066 status: "failure",
1067 finished_at: new Date().toISOString(),
1068 })
1069 .where("id", "=", runId)
1070 // Include "queued" so a row that was promoted-but-not-yet-
1071 // observed (or never promoted because promoteToPending threw)
1072 // still gets marked failed instead of stuck.
1073 .where("status", "in", ["pending", "running", "queued"])
1074 .execute();
1075 } catch (dbErr) {
1076 console.error(
1077 `[ci] failed to mark run ${runId} failed:`,
1078 dbErr,
1079 );
1080 }
1081 pumpQueue();
1082 }
1083 })();
1084}
1085
1086export async function retryRun(
1087 runId: number,
1088 retriedBy: number,
1089): Promise<void> {
1090 const run = await db
1091 .selectFrom("ci_runs")
1092 .select("repo_id")
1093 .where("id", "=", runId)
1094 .executeTakeFirst();
1095 if (!run) throw new Error("Run not found");
1096
1097 // Delete existing steps
1098 await db.deleteFrom("ci_steps").where("run_id", "=", runId).execute();
1099
1100 // Delete artifacts from disk and DB
1101 const artifactDir = path.join(paths.CI_ARTIFACTS_DIR, String(runId));
1102 if (existsSync(artifactDir)) {
1103 await Bun.$`rm -rf ${artifactDir}`.quiet().nothrow();
1104 }
1105 await db.deleteFrom("ci_artifacts").where("run_id", "=", runId).execute();
1106
1107 // Reset to "pending" first; if a slot isn't free, flip to "queued"
1108 // before enqueueing. Mirrors triggerRun: pumpQueue is the sole writer
1109 // of runningTasks.set, so we never reserve a slot directly here. Two
1110 // concurrent retryRun calls (or a retryRun racing triggerRun) all
1111 // funnel through pumpQueue, which serializes the size check against
1112 // slot reservation in a single synchronous turn.
1113 await db
1114 .updateTable("ci_runs")
1115 .set({
1116 status: "pending",
1117 triggered_by: retriedBy,
1118 started_at: null,
1119 finished_at: null,
1120 })
1121 .where("id", "=", runId)
1122 .execute();
1123
1124 if (runningTasks.size >= config.CI_MAX_CONCURRENT) {
1125 await db
1126 .updateTable("ci_runs")
1127 .set({ status: "queued" })
1128 .where("id", "=", runId)
1129 .execute();
1130 }
1131 queuedRunIds.push(runId);
1132 pumpQueue();
1133}
1134
1135export async function cancelRun(runId: number): Promise<void> {
1136 const task = runningTasks.get(runId);
1137 if (task) {
1138 const { containerId } = task;
1139 task.controller.abort();
1140 if (containerId) {
1141 await removeContainer(containerId).catch(() => {});
1142 }
1143 }
1144 const queueIdx = queuedRunIds.indexOf(runId);
1145 if (queueIdx >= 0) queuedRunIds.splice(queueIdx, 1);
1146 await db
1147 .updateTable("ci_runs")
1148 .set({ status: "cancelled", finished_at: new Date().toISOString() })
1149 .where("id", "=", runId)
1150 .where("status", "in", ["pending", "running", "queued"])
1151 .execute();
1152}
1153
1154async function pruneHistory(repoId: number): Promise<void> {
1155 const maxHistory = config.CI_MAX_HISTORY;
1156 const allRuns = await db
1157 .selectFrom("ci_runs")
1158 .select("id")
1159 .where("repo_id", "=", repoId)
1160 .orderBy("id", "desc")
1161 .execute();
1162
1163 if (allRuns.length <= maxHistory) return;
1164
1165 const toDelete = allRuns.slice(maxHistory).map((r) => r.id);
1166 for (const runId of toDelete) {
1167 // Remove artifacts from disk
1168 const artifactDir = path.join(paths.CI_ARTIFACTS_DIR, String(runId));
1169 if (existsSync(artifactDir)) {
1170 await Bun.$`rm -rf ${artifactDir}`.quiet().nothrow();
1171 }
1172 }
1173 await db.deleteFrom("ci_runs").where("id", "in", toDelete).execute();
1174}
1175
1176export async function purgeRepoCaches(repoName: string): Promise<void> {
1177 try {
1178 const filters = encodeURIComponent(
1179 JSON.stringify({ label: [`com.hearthforge.repo=${repoName}`] }),
1180 );
1181 const resp = await dockerFetch(`/volumes?filters=${filters}`);
1182 if (!resp.ok) {
1183 await resp.body?.cancel();
1184 return;
1185 }
1186 const data = (await resp.json()) as {
1187 Volumes?: Array<{ Name: string }>;
1188 };
1189 for (const vol of data.Volumes ?? []) {
1190 const delResp = await dockerFetch(`/volumes/${vol.Name}`, {
1191 method: "DELETE",
1192 });
1193 await delResp.body?.cancel();
1194 }
1195 } catch {
1196 // Best-effort
1197 }
1198}
1199
1200export async function cancelStaleRuns(): Promise<void> {
1201 const now = new Date().toISOString();
1202 const stale = await db
1203 .selectFrom("ci_runs")
1204 .select("id")
1205 .where("status", "in", ["pending", "running", "queued"])
1206 .execute();
1207
1208 await Promise.allSettled(
1209 stale.map((r) =>
1210 dockerFetch(`/containers/hearthforge-ci-${r.id}?force=true`, {
1211 method: "DELETE",
1212 }).then((res) => res.body?.cancel()),
1213 ),
1214 );
1215
1216 await db
1217 .updateTable("ci_runs")
1218 .set({ status: "cancelled", finished_at: now })
1219 .where("status", "in", ["pending", "running", "queued"])
1220 .execute();
1221 await db
1222 .updateTable("ci_steps")
1223 .set({
1224 status: "cancelled",
1225 finished_at: now,
1226 log: "Skipped: run was cancelled",
1227 })
1228 .where("status", "in", ["pending", "running"])
1229 .execute();
1230}
1231
1232/** Reset the cached socket path (used in tests to switch mock sockets). */
1233export function resetDockerSocket(): void {
1234 resolvedSocket = null;
1235}
1236
1237/** Check if CI can connect to the container socket. */
1238export async function ciAvailable(): Promise<boolean> {
1239 try {
1240 const socket = await getSocket();
1241 const resp = await fetch("http://localhost/v1.47/info", {
1242 unix: socket,
1243 });
1244 return resp.ok;
1245 } catch {
1246 return false;
1247 }
1248}
1249