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 { CI_MAX_LOG_BYTES, 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 if (!resp.ok) {
268 const detail = (await resp.text().catch(() => "")).trim();
269 throw new Error(
270 `Failed to pull image ${image}: HTTP ${resp.status}${detail ? ` ${detail}` : ""}`,
271 );
272 }
273 // The /images/create stream must be read to the end — the pull only
274 // completes when the stream does. Cancelling it (the previous behavior)
275 // aborted the pull, so createContainer could race a not-yet-present image.
276 // Each line is a JSON progress object; a trailing {"error": …} means the
277 // pull failed despite the HTTP 200.
278 const body = await resp.text();
279 for (const line of body.split("\n")) {
280 const trimmed = line.trim();
281 if (!trimmed) continue;
282 let obj: { error?: string } | null = null;
283 try {
284 obj = JSON.parse(trimmed);
285 } catch {
286 continue; // non-JSON progress line — ignore
287 }
288 if (obj?.error) {
289 throw new Error(`Failed to pull image ${image}: ${obj.error}`);
290 }
291 }
292}
293
294function parseMemoryBytes(s: string): number {
295 const m = s.match(/^(\d+(?:\.\d+)?)\s*([kmgKMG]?)b?$/);
296 if (!m) return 0;
297 const n = parseFloat(m[1] ?? "0");
298 switch ((m[2] ?? "").toLowerCase()) {
299 case "k":
300 return Math.floor(n * 1024);
301 case "m":
302 return Math.floor(n * 1024 * 1024);
303 case "g":
304 return Math.floor(n * 1024 * 1024 * 1024);
305 default:
306 return Math.floor(n);
307 }
308}
309
310async function ensureVolume(volName: string, repoName: string): Promise<void> {
311 const resp = await dockerFetch("/volumes/create", {
312 method: "POST",
313 headers: { "Content-Type": "application/json" },
314 body: JSON.stringify({
315 Name: volName,
316 Labels: { "com.hearthforge.repo": repoName },
317 }),
318 });
319 await resp.body?.cancel();
320}
321
322async function createContainer(
323 runId: number,
324 repoName: string,
325 cfg: CiConfig,
326 repoAbsPath: string,
327 envVars: string[],
328): Promise<string> {
329 const binds: string[] = [`${repoAbsPath}:/hearthforge-repo.git:ro`];
330 if (cfg.cache) {
331 for (const cachePath of cfg.cache) {
332 const volName = `hearthforge-ci-cache-${Buffer.from(`${repoName}:${cachePath}`).toString("base64url").slice(0, 24)}`;
333 await ensureVolume(volName, repoName);
334 binds.push(`${volName}:${cachePath}`);
335 }
336 }
337
338 const hostConfig: Record<string, unknown> = { Binds: binds };
339 if (cfg.cpu_limit) {
340 hostConfig.NanoCpus = Math.floor(cfg.cpu_limit * 1e9);
341 }
342 if (cfg.memory_limit) {
343 hostConfig.Memory = parseMemoryBytes(cfg.memory_limit);
344 }
345
346 const body = JSON.stringify({
347 Image: cfg.image,
348 Cmd: ["sleep", "infinity"],
349 Env: envVars,
350 WorkingDir: cfg.work_dir ?? "/",
351 HostConfig: hostConfig,
352 });
353
354 const resp = await dockerFetch(
355 `/containers/create?name=hearthforge-ci-${runId}`,
356 {
357 method: "POST",
358 headers: { "Content-Type": "application/json" },
359 body,
360 },
361 );
362 if (!resp.ok) {
363 const text = await resp.text();
364 throw new Error(`Failed to create container: ${resp.status} ${text}`);
365 }
366 const data = (await resp.json()) as { Id: string };
367 return data.Id;
368}
369
370async function startContainer(containerId: string): Promise<void> {
371 const resp = await dockerFetch(`/containers/${containerId}/start`, {
372 method: "POST",
373 });
374 if (!resp.ok && resp.status !== 304) {
375 throw new Error(`Failed to start container: ${resp.status}`);
376 }
377 await resp.body?.cancel();
378}
379
380interface ExecResult {
381 log: string;
382 exitCode: number;
383}
384
385const dec = new TextDecoder();
386
387function parseMuxFrames(buf: Uint8Array): {
388 text: string;
389 remaining: Uint8Array<ArrayBuffer>;
390} {
391 const chunks: string[] = [];
392 let i = 0;
393 while (i + 8 <= buf.length) {
394 const view = new DataView(buf.buffer, buf.byteOffset + i, 8);
395 const size = view.getUint32(4, false);
396 if (i + 8 + size > buf.length) break;
397 chunks.push(dec.decode(buf.slice(i + 8, i + 8 + size)));
398 i += 8 + size;
399 }
400 const remaining = new Uint8Array(buf.length - i);
401 if (i < buf.length) remaining.set(buf.subarray(i));
402 return { text: chunks.join(""), remaining };
403}
404
405async function execInContainer(
406 containerId: string,
407 cmd: string[],
408 workDir?: string,
409 envVars?: string[],
410 signal?: AbortSignal,
411 onPartialLog?: (log: string) => Promise<void>,
412): Promise<ExecResult> {
413 // Create exec
414 const execBody = JSON.stringify({
415 Cmd: cmd,
416 AttachStdout: true,
417 AttachStderr: true,
418 ...(workDir ? { WorkingDir: workDir } : {}),
419 ...(envVars ? { Env: envVars } : {}),
420 });
421 const createResp = await dockerFetch(`/containers/${containerId}/exec`, {
422 method: "POST",
423 headers: { "Content-Type": "application/json" },
424 body: execBody,
425 signal,
426 });
427 if (!createResp.ok) {
428 const text = await createResp.text();
429 throw new Error(`Failed to create exec: ${createResp.status} ${text}`);
430 }
431 const createData = (await createResp.json()) as { Id: string };
432 const execId = createData.Id;
433
434 // Start exec and stream output
435 const startResp = await dockerFetch(`/exec/${execId}/start`, {
436 method: "POST",
437 headers: { "Content-Type": "application/json" },
438 body: JSON.stringify({ Detach: false, Tty: false }),
439 signal,
440 });
441
442 let log = "";
443 let truncated = false;
444 // Bound the buffered log: stop appending once we hit the cap (and note it
445 // once) so a runaway step can't exhaust RAM or make each partial-flush
446 // rewrite an ever-growing row.
447 const appendLog = (text: string) => {
448 if (truncated || !text) return;
449 const room = CI_MAX_LOG_BYTES - log.length;
450 if (text.length <= room) {
451 log += text;
452 } else {
453 log += text.slice(0, Math.max(0, room));
454 log += `\n[log truncated at ${CI_MAX_LOG_BYTES} bytes]\n`;
455 truncated = true;
456 }
457 };
458
459 if (onPartialLog && startResp.body) {
460 const reader = startResp.body.getReader();
461 let buf = new Uint8Array(0);
462 let lastSave = Date.now();
463 while (true) {
464 const { done, value } = await reader.read();
465 if (done) break;
466 const merged = new Uint8Array(buf.length + value.length);
467 merged.set(buf);
468 merged.set(value, buf.length);
469 buf = merged;
470 const { text, remaining } = parseMuxFrames(buf);
471 buf = remaining;
472 appendLog(text);
473 // Once truncated the log no longer changes, so stop re-flushing it.
474 if (!truncated && Date.now() - lastSave >= 2000) {
475 await onPartialLog(log);
476 lastSave = Date.now();
477 }
478 }
479 const { text } = parseMuxFrames(buf);
480 appendLog(text);
481 } else {
482 const bodyBytes = new Uint8Array(await startResp.arrayBuffer());
483 const { text } = parseMuxFrames(bodyBytes);
484 appendLog(text);
485 }
486
487 // Get exit code
488 const inspectResp = await dockerFetch(`/exec/${execId}/json`);
489 const inspectData = (await inspectResp.json()) as { ExitCode: number };
490
491 return { log, exitCode: inspectData.ExitCode ?? 1 };
492}
493
494async function removeContainer(containerId: string): Promise<void> {
495 try {
496 const resp = await dockerFetch(
497 `/containers/${containerId}?force=true`,
498 { method: "DELETE" },
499 );
500 await resp.body?.cancel();
501 } catch {
502 // Best-effort cleanup
503 }
504}
505
506// --- Tar extraction ---
507
508function extractSingleFileFromTar(data: Uint8Array): Uint8Array | null {
509 if (data.length < 512) return null;
510 const dec = new TextDecoder();
511 const sizeOctal = dec
512 .decode(data.slice(124, 136))
513 .replace(/\0/g, "")
514 .trim();
515 const size = parseInt(sizeOctal, 8);
516 if (Number.isNaN(size) || size < 0) return null;
517 if (data.length < 512 + size) return null;
518 return data.slice(512, 512 + size);
519}
520
521async function copyFileFromContainer(
522 containerId: string,
523 containerPath: string,
524): Promise<Uint8Array | null> {
525 const resp = await dockerFetch(
526 `/containers/${containerId}/archive?path=${encodeURIComponent(containerPath)}`,
527 );
528 if (!resp.ok) return null;
529 const tarBytes = new Uint8Array(await resp.arrayBuffer());
530 return extractSingleFileFromTar(tarBytes);
531}
532
533// --- Secret masking ---
534
535function maskSecrets(text: string, secrets: string[]): string {
536 for (const secret of secrets) {
537 if (secret) text = text.split(secret).join("[MASKED]");
538 }
539 return text;
540}
541
542// --- Env var building ---
543
544function buildEnvVars(
545 runId: number,
546 repoName: string,
547 opts: TriggerOpts,
548 cfg: CiConfig,
549 secretValues: Array<{ name: string; value: string }>,
550): { envArray: string[]; secretValues: string[] } {
551 const vars: Record<string, string> = {
552 CI: "true",
553 CI_PIPELINE_ID: String(runId),
554 CI_REPO_NAME: repoName,
555 CI_SERVER_URL: config.BASE_URL,
556 CI_TRIGGER_SOURCE: opts.triggerSource,
557 CI_COMMIT_SHA: opts.commitSha,
558 CI_COMMIT_SHORT_SHA: opts.commitSha.slice(0, 8),
559 CI_COMMIT_BRANCH: opts.commitBranch ?? "",
560 CI_COMMIT_TAG: opts.commitTag ?? "",
561 CI_COMMIT_REF_NAME: opts.commitTag ?? opts.commitBranch ?? "",
562 };
563
564 // User-defined variable defaults
565 if (cfg.variables) {
566 for (const [name, def] of Object.entries(cfg.variables)) {
567 if (def.default !== undefined) vars[name] = def.default;
568 }
569 }
570
571 // Variable overrides from manual trigger
572 if (opts.variableOverrides) {
573 for (const [name, value] of Object.entries(opts.variableOverrides)) {
574 vars[name] = value;
575 }
576 }
577
578 // Secrets (injected but values tracked for masking)
579 const secretVals: string[] = [];
580 for (const { name, value } of secretValues) {
581 vars[name] = value;
582 secretVals.push(value);
583 }
584
585 const envArray = Object.entries(vars).map(([k, v]) => `${k}=${v}`);
586 return { envArray, secretValues: secretVals };
587}
588
589// --- Artifact collection ---
590
591async function collectArtifacts(
592 runId: number,
593 containerId: string,
594 step: CiStep,
595 _shell: string[],
596 envVars: string[],
597): Promise<void> {
598 const artifactDir = path.join(paths.CI_ARTIFACTS_DIR, String(runId));
599 mkdirSync(artifactDir, { recursive: true });
600
601 const toArray = (v: string | string[] | undefined): string[] => {
602 if (!v) return [];
603 return Array.isArray(v) ? v : [v];
604 };
605
606 // publish_file: copy directly out of container
607 for (const srcPath of toArray(step.publish_file)) {
608 const fileBytes = await copyFileFromContainer(containerId, srcPath);
609 if (fileBytes) {
610 const filename = path.basename(srcPath);
611 const destPath = path.join(artifactDir, filename);
612 writeFileSync(destPath, fileBytes);
613 const stat = Bun.file(destPath);
614 await db
615 .insertInto("ci_artifacts")
616 .values({
617 run_id: runId,
618 filename,
619 size: stat.size,
620 })
621 .execute();
622 }
623 }
624
625 // Archive commands run with `path.dirname(srcPath)` as the working
626 // directory and reference the source by its basename only, so the
627 // user-controlled path never appears as part of an interpolated shell
628 // string. Previously `publish_zip` used `sh -c "cd … && zip …"` with
629 // raw interpolation — a step author who could write the CI TOML
630 // could shell-inject through the source path. Today the only TOML
631 // author is the admin, but this removes the implicit assumption.
632 type ArchiveType = "tar" | "gzip" | "zip" | "zstd";
633 const archiveFormats: Array<{
634 type: ArchiveType;
635 paths: string[];
636 ext: string;
637 cmd: (basename: string, dst: string) => string[];
638 }> = [
639 {
640 type: "tar",
641 paths: toArray(step.publish_tar),
642 ext: ".tar",
643 cmd: (basename, dst) => ["tar", "-cf", dst, basename],
644 },
645 {
646 type: "gzip",
647 paths: toArray(step.publish_gzip),
648 ext: ".tar.gz",
649 cmd: (basename, dst) => ["tar", "-czf", dst, basename],
650 },
651 {
652 type: "zstd",
653 paths: toArray(step.publish_zstd),
654 ext: ".tar.zst",
655 cmd: (basename, dst) => ["tar", "--zstd", "-cf", dst, basename],
656 },
657 {
658 type: "zip",
659 paths: toArray(step.publish_zip),
660 ext: ".zip",
661 cmd: (basename, dst) => ["zip", "-r", dst, basename],
662 },
663 ];
664
665 let archiveIndex = 0;
666 for (const { paths: archivePaths, ext, cmd } of archiveFormats) {
667 for (const srcPath of archivePaths) {
668 archiveIndex++;
669 const tmpPath = `/tmp/hf-artifact-${runId}-${archiveIndex}${ext}`;
670 // Run with the source's parent directory as the working
671 // directory so each archive tool can reference the source
672 // by its basename — no -C, no shell.
673 const execResult = await execInContainer(
674 containerId,
675 cmd(path.basename(srcPath), tmpPath),
676 path.dirname(srcPath),
677 envVars,
678 ).catch(() => null);
679 if (!execResult || execResult.exitCode !== 0) continue;
680
681 // Copy archive out
682 const fileBytes = await copyFileFromContainer(containerId, tmpPath);
683 if (!fileBytes) continue;
684
685 const filename = `${path.basename(srcPath)}${ext}`;
686 const destPath = path.join(artifactDir, filename);
687 writeFileSync(destPath, fileBytes);
688 const stat = Bun.file(destPath);
689 await db
690 .insertInto("ci_artifacts")
691 .values({
692 run_id: runId,
693 filename,
694 size: stat.size,
695 })
696 .execute();
697 }
698 }
699}
700
701// --- Main execution ---
702
703async function executeRun(runId: number, signal: AbortSignal): Promise<void> {
704 const now = () => new Date().toISOString();
705 let containerId: string | undefined;
706
707 try {
708 // Mark as running
709 await db
710 .updateTable("ci_runs")
711 .set({ status: "running", started_at: now() })
712 .where("id", "=", runId)
713 .execute();
714
715 // Load run details
716 const run = await db
717 .selectFrom("ci_runs")
718 .selectAll()
719 .where("id", "=", runId)
720 .executeTakeFirst();
721 if (!run) throw new Error("Run not found");
722
723 const repo = await db
724 .selectFrom("repositories")
725 .select(["id", "name"])
726 .where("id", "=", run.repo_id)
727 .executeTakeFirst();
728 if (!repo) throw new Error("Repo not found");
729
730 // Read .hearthforge-ci.toml at the commit
731 const tomlBuf = await import("./git.ts").then((g) =>
732 g.git.show(repo.name, run.commit_sha!, ".hearthforge-ci.toml"),
733 );
734 if (!tomlBuf)
735 throw new Error(".hearthforge-ci.toml not found at commit");
736
737 const cfg = parseCiConfig(tomlBuf.toString("utf-8"));
738 if (!cfg) throw new Error("Failed to parse .hearthforge-ci.toml");
739
740 // Load secrets for log masking
741 const secrets = await db
742 .selectFrom("ci_secrets")
743 .select(["name", "value"])
744 .where("repo_id", "=", repo.id)
745 .execute();
746
747 const variableOverrides = run.variable_overrides
748 ? (JSON.parse(run.variable_overrides) as Record<string, string>)
749 : {};
750
751 const { envArray, secretValues } = buildEnvVars(
752 runId,
753 repo.name,
754 {
755 triggerSource:
756 run.trigger_source as TriggerOpts["triggerSource"],
757 commitSha: run.commit_sha ?? "",
758 commitBranch: run.commit_branch ?? undefined,
759 commitTag: run.commit_tag ?? undefined,
760 variableOverrides,
761 },
762 cfg,
763 secrets,
764 );
765
766 // Create step rows in DB
767 for (const step of cfg.steps) {
768 await db
769 .insertInto("ci_steps")
770 .values({
771 run_id: runId,
772 name: step.name,
773 status: "pending",
774 })
775 .execute();
776 }
777
778 // Pull image
779 await pullImage(cfg.image);
780 if (signal.aborted) throw new Error("Cancelled");
781
782 // Create + start container
783 containerId = await createContainer(
784 runId,
785 repo.name,
786 cfg,
787 repoPath(repo.name),
788 envArray,
789 );
790 runningTasks.get(runId)!.containerId = containerId;
791
792 await startContainer(containerId);
793 if (signal.aborted) throw new Error("Cancelled");
794
795 // Create work_dir
796 if (cfg.work_dir) {
797 await execInContainer(containerId, ["mkdir", "-p", cfg.work_dir]);
798 }
799
800 // Clone project if requested
801 if (cfg.clone_project_to && run.commit_sha) {
802 await execInContainer(
803 containerId,
804 [
805 "sh",
806 "-c",
807 `git clone /hearthforge-repo.git ${cfg.clone_project_to} && git -C ${cfg.clone_project_to} checkout --detach ${run.commit_sha}`,
808 ],
809 cfg.work_dir,
810 envArray,
811 );
812 }
813
814 // Execute steps
815 const shell = cfg.shell ?? ["/bin/sh", "-c"];
816 let runFailed = false;
817
818 for (const step of cfg.steps) {
819 if (signal.aborted) {
820 runFailed = true;
821 break;
822 }
823
824 const stepRow = await db
825 .selectFrom("ci_steps")
826 .select("id")
827 .where("run_id", "=", runId)
828 .where("name", "=", step.name)
829 .executeTakeFirst();
830 if (!stepRow) continue;
831 const stepId = stepRow.id;
832
833 // Check run_if condition
834 if (step.run_if) {
835 const { exitCode } = await execInContainer(
836 containerId,
837 [...shell, step.run_if],
838 cfg.work_dir,
839 envArray,
840 );
841 if (exitCode !== 0) {
842 await db
843 .updateTable("ci_steps")
844 .set({
845 status: "skipped",
846 started_at: now(),
847 finished_at: now(),
848 log: "Skipped: condition not met",
849 })
850 .where("id", "=", stepId)
851 .execute();
852 continue;
853 }
854 }
855
856 // Handle clear option
857 if (step.clear && cfg.clone_project_to && run.commit_sha) {
858 await execInContainer(
859 containerId,
860 [
861 "sh",
862 "-c",
863 `git -C ${cfg.clone_project_to} reset --hard ${run.commit_sha} && git -C ${cfg.clone_project_to} clean -fdx`,
864 ],
865 cfg.work_dir,
866 envArray,
867 );
868 }
869
870 await db
871 .updateTable("ci_steps")
872 .set({ status: "running", started_at: now() })
873 .where("id", "=", stepId)
874 .execute();
875
876 let stepLog = "";
877 let stepStatus: "success" | "failure" = "success";
878
879 if (step.run_sh) {
880 const command = cfg.shell_setup
881 ? `${cfg.shell_setup}\n${step.run_sh}`
882 : step.run_sh;
883
884 const stepTimeout =
885 step.timeout ?? cfg.timeout ?? config.CI_DEFAULT_TIMEOUT;
886 const timeoutSignal = AbortSignal.timeout(stepTimeout * 1000);
887 // Abort the exec stream on either a run cancellation or the
888 // per-step timeout. Docker has no per-exec kill, so on abort we
889 // force-remove the container (below), which kills the command
890 // still running inside it.
891 const stepSignal = AbortSignal.any([signal, timeoutSignal]);
892
893 try {
894 const { log, exitCode } = await execInContainer(
895 containerId,
896 [...shell, command],
897 cfg.work_dir,
898 envArray,
899 stepSignal,
900 async (partial) => {
901 await db
902 .updateTable("ci_steps")
903 .set({
904 log: maskSecrets(partial, secretValues),
905 })
906 .where("id", "=", stepId)
907 .execute();
908 },
909 );
910 stepLog = maskSecrets(log, secretValues);
911 if (exitCode !== 0) {
912 stepStatus = "failure";
913 runFailed = true;
914 }
915 } catch (err) {
916 // A run cancellation is reported as "cancelled" by the outer
917 // catch — don't relabel it as a step failure here.
918 if (signal.aborted) throw err;
919 stepStatus = "failure";
920 runFailed = true;
921 if (timeoutSignal.aborted) {
922 stepLog = `Step timed out after ${stepTimeout}s\n`;
923 // Kill the container now so the timed-out command stops
924 // immediately rather than lingering until cleanup.
925 await removeContainer(containerId);
926 containerId = "";
927 } else {
928 stepLog = `Step failed: ${err instanceof Error ? err.message : String(err)}\n`;
929 }
930 }
931 }
932
933 // Collect artifacts for this step
934 if (!runFailed || stepStatus === "success") {
935 await collectArtifacts(
936 runId,
937 containerId,
938 step,
939 shell,
940 envArray,
941 ).catch(() => {});
942 }
943
944 await db
945 .updateTable("ci_steps")
946 .set({
947 status: stepStatus,
948 finished_at: now(),
949 log: stepLog,
950 })
951 .where("id", "=", stepId)
952 .execute();
953
954 if (runFailed) break;
955 }
956
957 // Mark remaining steps as skipped
958 await db
959 .updateTable("ci_steps")
960 .set({
961 status: "skipped",
962 started_at: now(),
963 finished_at: now(),
964 log: "Skipped: previous step failed",
965 })
966 .where("run_id", "=", runId)
967 .where("status", "=", "pending")
968 .execute();
969
970 const finalStatus = runFailed ? "failure" : "success";
971 await db
972 .updateTable("ci_runs")
973 .set({ status: finalStatus, finished_at: now() })
974 .where("id", "=", runId)
975 .execute();
976 } catch (err) {
977 const errMsg = err instanceof Error ? err.message : String(err);
978 const isDockerUnavailable = errMsg.includes("No Docker/Podman socket");
979 const status = signal.aborted
980 ? "cancelled"
981 : isDockerUnavailable
982 ? "skipped"
983 : "failure";
984 const skipLog = signal.aborted
985 ? "Skipped: run was cancelled"
986 : isDockerUnavailable
987 ? "Skipped: Docker/Podman not available"
988 : "Skipped: run failed";
989
990 if (!isDockerUnavailable) {
991 // Write error to a synthetic step if we have no steps yet
992 const hasSteps = await db
993 .selectFrom("ci_steps")
994 .select("id")
995 .where("run_id", "=", runId)
996 .executeTakeFirst();
997 if (!hasSteps) {
998 await db
999 .insertInto("ci_steps")
1000 .values({
1001 run_id: runId,
1002 name: "setup",
1003 status: "failure",
1004 started_at: new Date().toISOString(),
1005 finished_at: new Date().toISOString(),
1006 log: `Error: ${errMsg}\n`,
1007 })
1008 .execute();
1009 }
1010 }
1011 await db
1012 .updateTable("ci_runs")
1013 .set({ status, finished_at: new Date().toISOString() })
1014 .where("id", "=", runId)
1015 .execute();
1016 // Mark pending steps as skipped
1017 await db
1018 .updateTable("ci_steps")
1019 .set({
1020 status: "skipped",
1021 finished_at: new Date().toISOString(),
1022 log: skipLog,
1023 })
1024 .where("run_id", "=", runId)
1025 .where("status", "=", "pending")
1026 .execute();
1027 } finally {
1028 if (containerId) await removeContainer(containerId);
1029 runningTasks.delete(runId);
1030 // Prune old history
1031 const run = await db
1032 .selectFrom("ci_runs")
1033 .select("repo_id")
1034 .where("id", "=", runId)
1035 .executeTakeFirst();
1036 if (run) await pruneHistory(run.repo_id).catch(() => {});
1037 // A slot just freed up — start the next queued run if any.
1038 pumpQueue();
1039 }
1040}
1041
1042// --- Public API ---
1043
1044export async function triggerRun(
1045 repoName: string,
1046 opts: TriggerOpts,
1047): Promise<number> {
1048 const repo = await db
1049 .selectFrom("repositories")
1050 .select("id")
1051 .where("name", "=", repoName)
1052 .executeTakeFirst();
1053 if (!repo) throw new Error("Repository not found");
1054
1055 const runId = await db
1056 .insertInto("ci_runs")
1057 .values({
1058 repo_id: repo.id,
1059 triggered_by: opts.triggeredBy ?? null,
1060 trigger_source: opts.triggerSource,
1061 commit_sha: opts.commitSha,
1062 commit_branch: opts.commitBranch ?? null,
1063 commit_tag: opts.commitTag ?? null,
1064 status: "pending",
1065 variable_overrides: opts.variableOverrides
1066 ? JSON.stringify(opts.variableOverrides)
1067 : null,
1068 })
1069 .returning("id")
1070 .executeTakeFirstOrThrow();
1071
1072 // Allocate the human-facing run number atomically from a per-repo counter.
1073 // A single upsert-and-increment can't collide under concurrent triggers and
1074 // never reuses a number after pruneHistory shrinks the run table — both of
1075 // which a COUNT(*)-based scheme suffered from.
1076 const counter = await db
1077 .insertInto("ci_run_counters")
1078 .values({ repo_id: repo.id, last_run_id: 1 })
1079 .onConflict((oc) =>
1080 oc.column("repo_id").doUpdateSet((eb) => ({
1081 last_run_id: eb("ci_run_counters.last_run_id", "+", 1),
1082 })),
1083 )
1084 .returning("last_run_id")
1085 .executeTakeFirstOrThrow();
1086 await db
1087 .updateTable("ci_runs")
1088 .set({ repo_run_id: counter.last_run_id })
1089 .where("id", "=", runId.id)
1090 .execute();
1091
1092 // Flip to "queued" BEFORE pushing onto queuedRunIds. Otherwise a
1093 // concurrent pumpQueue (from a finishing run) could observe our entry,
1094 // promoteToPending it, and start executeRun while our UPDATE is still
1095 // in flight — the late UPDATE would then clobber a running row back to
1096 // "queued". Only pumpQueue (the sole writer of runningTasks.set) may
1097 // mutate the row's status after the push. The TOCTOU concern is moot
1098 // here because pumpQueue cannot pump a runId that hasn't been pushed.
1099 if (runningTasks.size >= config.CI_MAX_CONCURRENT) {
1100 await db
1101 .updateTable("ci_runs")
1102 .set({ status: "queued" })
1103 .where("id", "=", runId.id)
1104 .execute();
1105 }
1106 queuedRunIds.push(runId.id);
1107 pumpQueue();
1108
1109 return runId.id;
1110}
1111
1112// Wraps the fire-and-forget executeRun so unhandled exceptions (e.g. a throw
1113// before/after its own try/finally) are logged and the run is reconciled to
1114// "failure" instead of staying pending forever.
1115function spawnRun(
1116 runId: number,
1117 signal: AbortSignal,
1118 needsPromote = false,
1119): void {
1120 void (async () => {
1121 try {
1122 if (needsPromote) await promoteToPending(runId);
1123 await executeRun(runId, signal);
1124 } catch (err) {
1125 console.error(`[ci] executeRun threw for run ${runId}:`, err);
1126 runningTasks.delete(runId);
1127 try {
1128 await db
1129 .updateTable("ci_runs")
1130 .set({
1131 status: "failure",
1132 finished_at: new Date().toISOString(),
1133 })
1134 .where("id", "=", runId)
1135 // Include "queued" so a row that was promoted-but-not-yet-
1136 // observed (or never promoted because promoteToPending threw)
1137 // still gets marked failed instead of stuck.
1138 .where("status", "in", ["pending", "running", "queued"])
1139 .execute();
1140 } catch (dbErr) {
1141 console.error(
1142 `[ci] failed to mark run ${runId} failed:`,
1143 dbErr,
1144 );
1145 }
1146 pumpQueue();
1147 }
1148 })();
1149}
1150
1151export async function retryRun(
1152 runId: number,
1153 retriedBy: number,
1154): Promise<void> {
1155 const run = await db
1156 .selectFrom("ci_runs")
1157 .select("repo_id")
1158 .where("id", "=", runId)
1159 .executeTakeFirst();
1160 if (!run) throw new Error("Run not found");
1161
1162 // Delete existing steps
1163 await db.deleteFrom("ci_steps").where("run_id", "=", runId).execute();
1164
1165 // Delete artifacts from disk and DB
1166 const artifactDir = path.join(paths.CI_ARTIFACTS_DIR, String(runId));
1167 if (existsSync(artifactDir)) {
1168 await Bun.$`rm -rf ${artifactDir}`.quiet().nothrow();
1169 }
1170 await db.deleteFrom("ci_artifacts").where("run_id", "=", runId).execute();
1171
1172 // Reset to "pending" first; if a slot isn't free, flip to "queued"
1173 // before enqueueing. Mirrors triggerRun: pumpQueue is the sole writer
1174 // of runningTasks.set, so we never reserve a slot directly here. Two
1175 // concurrent retryRun calls (or a retryRun racing triggerRun) all
1176 // funnel through pumpQueue, which serializes the size check against
1177 // slot reservation in a single synchronous turn.
1178 await db
1179 .updateTable("ci_runs")
1180 .set({
1181 status: "pending",
1182 triggered_by: retriedBy,
1183 started_at: null,
1184 finished_at: null,
1185 })
1186 .where("id", "=", runId)
1187 .execute();
1188
1189 if (runningTasks.size >= config.CI_MAX_CONCURRENT) {
1190 await db
1191 .updateTable("ci_runs")
1192 .set({ status: "queued" })
1193 .where("id", "=", runId)
1194 .execute();
1195 }
1196 queuedRunIds.push(runId);
1197 pumpQueue();
1198}
1199
1200export async function cancelRun(runId: number): Promise<void> {
1201 const task = runningTasks.get(runId);
1202 if (task) {
1203 const { containerId } = task;
1204 task.controller.abort();
1205 if (containerId) {
1206 await removeContainer(containerId).catch(() => {});
1207 }
1208 }
1209 const queueIdx = queuedRunIds.indexOf(runId);
1210 if (queueIdx >= 0) queuedRunIds.splice(queueIdx, 1);
1211 await db
1212 .updateTable("ci_runs")
1213 .set({ status: "cancelled", finished_at: new Date().toISOString() })
1214 .where("id", "=", runId)
1215 .where("status", "in", ["pending", "running", "queued"])
1216 .execute();
1217}
1218
1219async function pruneHistory(repoId: number): Promise<void> {
1220 const maxHistory = config.CI_MAX_HISTORY;
1221 const allRuns = await db
1222 .selectFrom("ci_runs")
1223 .select("id")
1224 .where("repo_id", "=", repoId)
1225 .orderBy("id", "desc")
1226 .execute();
1227
1228 if (allRuns.length <= maxHistory) return;
1229
1230 const toDelete = allRuns.slice(maxHistory).map((r) => r.id);
1231 for (const runId of toDelete) {
1232 // Remove artifacts from disk
1233 const artifactDir = path.join(paths.CI_ARTIFACTS_DIR, String(runId));
1234 if (existsSync(artifactDir)) {
1235 await Bun.$`rm -rf ${artifactDir}`.quiet().nothrow();
1236 }
1237 }
1238 await db.deleteFrom("ci_runs").where("id", "in", toDelete).execute();
1239}
1240
1241export async function purgeRepoCaches(repoName: string): Promise<void> {
1242 try {
1243 const filters = encodeURIComponent(
1244 JSON.stringify({ label: [`com.hearthforge.repo=${repoName}`] }),
1245 );
1246 const resp = await dockerFetch(`/volumes?filters=${filters}`);
1247 if (!resp.ok) {
1248 await resp.body?.cancel();
1249 return;
1250 }
1251 const data = (await resp.json()) as {
1252 Volumes?: Array<{ Name: string }>;
1253 };
1254 for (const vol of data.Volumes ?? []) {
1255 const delResp = await dockerFetch(`/volumes/${vol.Name}`, {
1256 method: "DELETE",
1257 });
1258 await delResp.body?.cancel();
1259 }
1260 } catch {
1261 // Best-effort
1262 }
1263}
1264
1265export async function cancelStaleRuns(): Promise<void> {
1266 const now = new Date().toISOString();
1267 const stale = await db
1268 .selectFrom("ci_runs")
1269 .select("id")
1270 .where("status", "in", ["pending", "running", "queued"])
1271 .execute();
1272
1273 await Promise.allSettled(
1274 stale.map((r) =>
1275 dockerFetch(`/containers/hearthforge-ci-${r.id}?force=true`, {
1276 method: "DELETE",
1277 }).then((res) => res.body?.cancel()),
1278 ),
1279 );
1280
1281 await db
1282 .updateTable("ci_runs")
1283 .set({ status: "cancelled", finished_at: now })
1284 .where("status", "in", ["pending", "running", "queued"])
1285 .execute();
1286 await db
1287 .updateTable("ci_steps")
1288 .set({
1289 status: "cancelled",
1290 finished_at: now,
1291 log: "Skipped: run was cancelled",
1292 })
1293 .where("status", "in", ["pending", "running"])
1294 .execute();
1295}
1296
1297/** Reset the cached socket path (used in tests to switch mock sockets). */
1298export function resetDockerSocket(): void {
1299 resolvedSocket = null;
1300}
1301
1302/** Check if CI can connect to the container socket. */
1303export async function ciAvailable(): Promise<boolean> {
1304 try {
1305 const socket = await getSocket();
1306 const resp = await fetch("http://localhost/v1.47/info", {
1307 unix: socket,
1308 });
1309 return resp.ok;
1310 } catch {
1311 return false;
1312 }
1313}
1314