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