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_RUNS_PER_PAGE, 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 if (config.CI_DOCKER_SOCKET) {
180 resolvedSocket = config.CI_DOCKER_SOCKET;
181 return resolvedSocket;
182 }
183 const uid = process.getuid?.();
184 const candidates = [
185 "/var/run/docker.sock",
186 "/run/podman/podman.sock",
187 ...(uid !== undefined ? [`/run/user/${uid}/podman/podman.sock`] : []),
188 ];
189 for (const s of candidates) {
190 if (existsSync(s)) {
191 resolvedSocket = s;
192 return s;
193 }
194 }
195 throw new Error(
196 "No Docker/Podman socket found. Set CI_DOCKER_SOCKET env var.",
197 );
198}
199
200async function dockerFetch(
201 endpoint: string,
202 init?: RequestInit,
203): Promise<Response> {
204 const socket = await getSocket();
205 return fetch(`http://localhost/v1.47${endpoint}`, {
206 ...init,
207 unix: socket,
208 });
209}
210
211// --- Docker helpers ---
212
213function splitImageRef(image: string): { name: string; tag: string } {
214 const lastColon = image.lastIndexOf(":");
215 if (lastColon < 0) return { name: image, tag: "latest" };
216 const possibleTag = image.slice(lastColon + 1);
217 if (possibleTag.includes("/")) return { name: image, tag: "latest" };
218 return { name: image.slice(0, lastColon), tag: possibleTag };
219}
220
221async function pullImage(image: string): Promise<void> {
222 const { name, tag } = splitImageRef(image);
223 const resp = await dockerFetch(
224 `/images/create?fromImage=${encodeURIComponent(name)}&tag=${encodeURIComponent(tag)}`,
225 { method: "POST" },
226 );
227 // Consume body to completion
228 await resp.body?.cancel();
229}
230
231function parseMemoryBytes(s: string): number {
232 const m = s.match(/^(\d+(?:\.\d+)?)\s*([kmgKMG]?)b?$/);
233 if (!m) return 0;
234 const n = parseFloat(m[1] ?? "0");
235 switch ((m[2] ?? "").toLowerCase()) {
236 case "k":
237 return Math.floor(n * 1024);
238 case "m":
239 return Math.floor(n * 1024 * 1024);
240 case "g":
241 return Math.floor(n * 1024 * 1024 * 1024);
242 default:
243 return Math.floor(n);
244 }
245}
246
247async function createContainer(
248 runId: number,
249 cfg: CiConfig,
250 repoAbsPath: string,
251 envVars: string[],
252): Promise<string> {
253 const binds: string[] = [`${repoAbsPath}:/hearthforge-repo.git:ro`];
254 if (cfg.cache) {
255 for (const cachePath of cfg.cache) {
256 const volName = `hearthforge-ci-cache-${Buffer.from(`${runId}-${cachePath}`).toString("base64url").slice(0, 24)}`;
257 binds.push(`${volName}:${cachePath}`);
258 }
259 }
260
261 const hostConfig: Record<string, unknown> = { Binds: binds };
262 if (cfg.cpu_limit) {
263 hostConfig.NanoCpus = Math.floor(cfg.cpu_limit * 1e9);
264 }
265 if (cfg.memory_limit) {
266 hostConfig.Memory = parseMemoryBytes(cfg.memory_limit);
267 }
268
269 const body = JSON.stringify({
270 Image: cfg.image,
271 Cmd: ["sleep", "infinity"],
272 Env: envVars,
273 WorkingDir: cfg.work_dir ?? "/",
274 HostConfig: hostConfig,
275 });
276
277 const resp = await dockerFetch(
278 `/containers/create?name=hearthforge-ci-${runId}`,
279 {
280 method: "POST",
281 headers: { "Content-Type": "application/json" },
282 body,
283 },
284 );
285 if (!resp.ok) {
286 const text = await resp.text();
287 throw new Error(`Failed to create container: ${resp.status} ${text}`);
288 }
289 const data = (await resp.json()) as { Id: string };
290 return data.Id;
291}
292
293async function startContainer(containerId: string): Promise<void> {
294 const resp = await dockerFetch(`/containers/${containerId}/start`, {
295 method: "POST",
296 });
297 if (!resp.ok && resp.status !== 304) {
298 throw new Error(`Failed to start container: ${resp.status}`);
299 }
300 await resp.body?.cancel();
301}
302
303interface ExecResult {
304 log: string;
305 exitCode: number;
306}
307
308function parseMuxStream(data: Uint8Array): string {
309 const chunks: string[] = [];
310 const dec = new TextDecoder();
311 let i = 0;
312 while (i + 8 <= data.length) {
313 const view = new DataView(data.buffer, data.byteOffset + i, 8);
314 const size = view.getUint32(4, false);
315 i += 8;
316 if (i + size > data.length) break;
317 chunks.push(dec.decode(data.slice(i, i + size)));
318 i += size;
319 }
320 return chunks.join("");
321}
322
323async function execInContainer(
324 containerId: string,
325 cmd: string[],
326 workDir?: string,
327 envVars?: string[],
328 signal?: AbortSignal,
329): Promise<ExecResult> {
330 // Create exec
331 const execBody = JSON.stringify({
332 Cmd: cmd,
333 AttachStdout: true,
334 AttachStderr: true,
335 ...(workDir ? { WorkingDir: workDir } : {}),
336 ...(envVars ? { Env: envVars } : {}),
337 });
338 const createResp = await dockerFetch(`/containers/${containerId}/exec`, {
339 method: "POST",
340 headers: { "Content-Type": "application/json" },
341 body: execBody,
342 signal,
343 });
344 if (!createResp.ok) {
345 const text = await createResp.text();
346 throw new Error(`Failed to create exec: ${createResp.status} ${text}`);
347 }
348 const createData = (await createResp.json()) as { Id: string };
349 const execId = createData.Id;
350
351 // Start exec and capture output
352 const startResp = await dockerFetch(`/exec/${execId}/start`, {
353 method: "POST",
354 headers: { "Content-Type": "application/json" },
355 body: JSON.stringify({ Detach: false, Tty: false }),
356 signal,
357 });
358 const bodyBytes = new Uint8Array(await startResp.arrayBuffer());
359 const log = parseMuxStream(bodyBytes);
360
361 // Get exit code
362 const inspectResp = await dockerFetch(`/exec/${execId}/json`);
363 const inspectData = (await inspectResp.json()) as { ExitCode: number };
364
365 return { log, exitCode: inspectData.ExitCode ?? 1 };
366}
367
368async function removeContainer(containerId: string): Promise<void> {
369 try {
370 const resp = await dockerFetch(
371 `/containers/${containerId}?force=true`,
372 { method: "DELETE" },
373 );
374 await resp.body?.cancel();
375 } catch {
376 // Best-effort cleanup
377 }
378}
379
380// --- Tar extraction ---
381
382function extractSingleFileFromTar(data: Uint8Array): Uint8Array | null {
383 if (data.length < 512) return null;
384 const dec = new TextDecoder();
385 const sizeOctal = dec
386 .decode(data.slice(124, 136))
387 .replace(/\0/g, "")
388 .trim();
389 const size = parseInt(sizeOctal, 8);
390 if (Number.isNaN(size) || size < 0) return null;
391 if (data.length < 512 + size) return null;
392 return data.slice(512, 512 + size);
393}
394
395async function copyFileFromContainer(
396 containerId: string,
397 containerPath: string,
398): Promise<Uint8Array | null> {
399 const resp = await dockerFetch(
400 `/containers/${containerId}/archive?path=${encodeURIComponent(containerPath)}`,
401 );
402 if (!resp.ok) return null;
403 const tarBytes = new Uint8Array(await resp.arrayBuffer());
404 return extractSingleFileFromTar(tarBytes);
405}
406
407// --- Secret masking ---
408
409function maskSecrets(text: string, secrets: string[]): string {
410 for (const secret of secrets) {
411 if (secret) text = text.split(secret).join("[MASKED]");
412 }
413 return text;
414}
415
416// --- Env var building ---
417
418function buildEnvVars(
419 runId: number,
420 repoName: string,
421 opts: TriggerOpts,
422 cfg: CiConfig,
423 secretValues: Array<{ name: string; value: string }>,
424): { envArray: string[]; secretValues: string[] } {
425 const vars: Record<string, string> = {
426 CI: "true",
427 CI_PIPELINE_ID: String(runId),
428 CI_REPO_NAME: repoName,
429 CI_SERVER_URL: config.BASE_URL,
430 CI_TRIGGER_SOURCE: opts.triggerSource,
431 CI_COMMIT_SHA: opts.commitSha,
432 CI_COMMIT_SHORT_SHA: opts.commitSha.slice(0, 8),
433 CI_COMMIT_BRANCH: opts.commitBranch ?? "",
434 CI_COMMIT_TAG: opts.commitTag ?? "",
435 CI_COMMIT_REF_NAME: opts.commitTag ?? opts.commitBranch ?? "",
436 };
437
438 // User-defined variable defaults
439 if (cfg.variables) {
440 for (const [name, def] of Object.entries(cfg.variables)) {
441 if (def.default !== undefined) vars[name] = def.default;
442 }
443 }
444
445 // Variable overrides from manual trigger
446 if (opts.variableOverrides) {
447 for (const [name, value] of Object.entries(opts.variableOverrides)) {
448 vars[name] = value;
449 }
450 }
451
452 // Secrets (injected but values tracked for masking)
453 const secretVals: string[] = [];
454 for (const { name, value } of secretValues) {
455 vars[name] = value;
456 secretVals.push(value);
457 }
458
459 const envArray = Object.entries(vars).map(([k, v]) => `${k}=${v}`);
460 return { envArray, secretValues: secretVals };
461}
462
463// --- Artifact collection ---
464
465async function collectArtifacts(
466 runId: number,
467 containerId: string,
468 step: CiStep,
469 shell: string[],
470 workDir: string | undefined,
471 envVars: string[],
472): Promise<void> {
473 const artifactDir = path.join(paths.CI_ARTIFACTS_DIR, String(runId));
474 mkdirSync(artifactDir, { recursive: true });
475
476 const toArray = (v: string | string[] | undefined): string[] => {
477 if (!v) return [];
478 return Array.isArray(v) ? v : [v];
479 };
480
481 // publish_file: copy directly out of container
482 for (const srcPath of toArray(step.publish_file)) {
483 const fileBytes = await copyFileFromContainer(containerId, srcPath);
484 if (fileBytes) {
485 const filename = path.basename(srcPath);
486 const destPath = path.join(artifactDir, filename);
487 writeFileSync(destPath, fileBytes);
488 const stat = Bun.file(destPath);
489 await db
490 .insertInto("ci_artifacts")
491 .values({
492 run_id: runId,
493 filename,
494 size: stat.size,
495 })
496 .execute();
497 }
498 }
499
500 type ArchiveType = "tar" | "gzip" | "zip" | "zstd";
501 const archiveFormats: Array<{
502 type: ArchiveType;
503 paths: string[];
504 ext: string;
505 cmd: (src: string, dst: string) => string[];
506 }> = [
507 {
508 type: "tar",
509 paths: toArray(step.publish_tar),
510 ext: ".tar",
511 cmd: (src, dst) => [
512 "tar",
513 "-cf",
514 dst,
515 "-C",
516 path.dirname(src),
517 path.basename(src),
518 ],
519 },
520 {
521 type: "gzip",
522 paths: toArray(step.publish_gzip),
523 ext: ".tar.gz",
524 cmd: (src, dst) => [
525 "tar",
526 "-czf",
527 dst,
528 "-C",
529 path.dirname(src),
530 path.basename(src),
531 ],
532 },
533 {
534 type: "zstd",
535 paths: toArray(step.publish_zstd),
536 ext: ".tar.zst",
537 cmd: (src, dst) => [
538 "tar",
539 "--zstd",
540 "-cf",
541 dst,
542 "-C",
543 path.dirname(src),
544 path.basename(src),
545 ],
546 },
547 {
548 type: "zip",
549 paths: toArray(step.publish_zip),
550 ext: ".zip",
551 cmd: (src, dst) => [
552 "sh",
553 "-c",
554 `cd ${path.dirname(src)} && zip -r ${dst} ${path.basename(src)}`,
555 ],
556 },
557 ];
558
559 let archiveIndex = 0;
560 for (const { paths: archivePaths, ext, cmd } of archiveFormats) {
561 for (const srcPath of archivePaths) {
562 archiveIndex++;
563 const tmpPath = `/tmp/hf-artifact-${runId}-${archiveIndex}${ext}`;
564 // Create archive inside container
565 const execResult = await execInContainer(
566 containerId,
567 cmd(srcPath, tmpPath),
568 workDir,
569 envVars,
570 ).catch(() => null);
571 if (!execResult || execResult.exitCode !== 0) continue;
572
573 // Copy archive out
574 const fileBytes = await copyFileFromContainer(containerId, tmpPath);
575 if (!fileBytes) continue;
576
577 const filename = `${path.basename(srcPath)}${ext}`;
578 const destPath = path.join(artifactDir, filename);
579 writeFileSync(destPath, fileBytes);
580 const stat = Bun.file(destPath);
581 await db
582 .insertInto("ci_artifacts")
583 .values({
584 run_id: runId,
585 filename,
586 size: stat.size,
587 })
588 .execute();
589 }
590 }
591}
592
593// --- Main execution ---
594
595async function executeRun(runId: number, signal: AbortSignal): Promise<void> {
596 const now = () => new Date().toISOString();
597 let containerId: string | undefined;
598
599 try {
600 // Mark as running
601 await db
602 .updateTable("ci_runs")
603 .set({ status: "running", started_at: now() })
604 .where("id", "=", runId)
605 .execute();
606
607 // Load run details
608 const run = await db
609 .selectFrom("ci_runs")
610 .selectAll()
611 .where("id", "=", runId)
612 .executeTakeFirst();
613 if (!run) throw new Error("Run not found");
614
615 const repo = await db
616 .selectFrom("repositories")
617 .select(["id", "name"])
618 .where("id", "=", run.repo_id)
619 .executeTakeFirst();
620 if (!repo) throw new Error("Repo not found");
621
622 // Read .hearthforge-ci.toml at the commit
623 const tomlBuf = await import("./git.ts").then((g) =>
624 g.git.show(repo.name, run.commit_sha!, ".hearthforge-ci.toml"),
625 );
626 if (!tomlBuf)
627 throw new Error(".hearthforge-ci.toml not found at commit");
628
629 const cfg = parseCiConfig(tomlBuf.toString("utf-8"));
630 if (!cfg) throw new Error("Failed to parse .hearthforge-ci.toml");
631
632 // Load secrets for log masking
633 const secrets = await db
634 .selectFrom("ci_secrets")
635 .select(["name", "value"])
636 .where("repo_id", "=", repo.id)
637 .execute();
638
639 const variableOverrides = run.variable_overrides
640 ? (JSON.parse(run.variable_overrides) as Record<string, string>)
641 : {};
642
643 const { envArray, secretValues } = buildEnvVars(
644 runId,
645 repo.name,
646 {
647 triggerSource:
648 run.trigger_source as TriggerOpts["triggerSource"],
649 commitSha: run.commit_sha ?? "",
650 commitBranch: run.commit_branch ?? undefined,
651 commitTag: run.commit_tag ?? undefined,
652 variableOverrides,
653 },
654 cfg,
655 secrets,
656 );
657
658 // Create step rows in DB
659 for (const step of cfg.steps) {
660 await db
661 .insertInto("ci_steps")
662 .values({
663 run_id: runId,
664 name: step.name,
665 status: "pending",
666 })
667 .execute();
668 }
669
670 // Pull image
671 await pullImage(cfg.image);
672 if (signal.aborted) throw new Error("Cancelled");
673
674 // Create + start container
675 containerId = await createContainer(
676 runId,
677 cfg,
678 repoPath(repo.name),
679 envArray,
680 );
681 runningTasks.get(runId)!.containerId = containerId;
682
683 await startContainer(containerId);
684 if (signal.aborted) throw new Error("Cancelled");
685
686 // Create work_dir
687 if (cfg.work_dir) {
688 await execInContainer(containerId, ["mkdir", "-p", cfg.work_dir]);
689 }
690
691 // Clone project if requested
692 if (cfg.clone_project_to && run.commit_sha) {
693 await execInContainer(
694 containerId,
695 [
696 "sh",
697 "-c",
698 `git clone /hearthforge-repo.git ${cfg.clone_project_to} && git -C ${cfg.clone_project_to} checkout --detach ${run.commit_sha}`,
699 ],
700 cfg.work_dir,
701 envArray,
702 );
703 }
704
705 // Execute steps
706 const shell = cfg.shell ?? ["/bin/sh", "-c"];
707 let runFailed = false;
708
709 for (const step of cfg.steps) {
710 if (signal.aborted) {
711 runFailed = true;
712 break;
713 }
714
715 const stepRow = await db
716 .selectFrom("ci_steps")
717 .select("id")
718 .where("run_id", "=", runId)
719 .where("name", "=", step.name)
720 .executeTakeFirst();
721 if (!stepRow) continue;
722 const stepId = stepRow.id;
723
724 // Check run_if condition
725 if (step.run_if) {
726 const { exitCode } = await execInContainer(
727 containerId,
728 [...shell, step.run_if],
729 cfg.work_dir,
730 envArray,
731 );
732 if (exitCode !== 0) {
733 await db
734 .updateTable("ci_steps")
735 .set({
736 status: "skipped",
737 started_at: now(),
738 finished_at: now(),
739 })
740 .where("id", "=", stepId)
741 .execute();
742 continue;
743 }
744 }
745
746 // Handle clear option
747 if (step.clear && cfg.clone_project_to && run.commit_sha) {
748 await execInContainer(
749 containerId,
750 [
751 "sh",
752 "-c",
753 `git -C ${cfg.clone_project_to} reset --hard ${run.commit_sha} && git -C ${cfg.clone_project_to} clean -fdx`,
754 ],
755 cfg.work_dir,
756 envArray,
757 );
758 }
759
760 await db
761 .updateTable("ci_steps")
762 .set({ status: "running", started_at: now() })
763 .where("id", "=", stepId)
764 .execute();
765
766 let stepLog = "";
767 let stepStatus: "success" | "failure" = "success";
768
769 if (step.run_sh) {
770 const command = cfg.shell_setup
771 ? `${cfg.shell_setup}\n${step.run_sh}`
772 : step.run_sh;
773
774 const stepTimeout =
775 step.timeout ?? cfg.timeout ?? config.CI_DEFAULT_TIMEOUT;
776 const timeoutSignal = AbortSignal.timeout(stepTimeout * 1000);
777
778 try {
779 const { log, exitCode } = await execInContainer(
780 containerId,
781 [...shell, command],
782 cfg.work_dir,
783 envArray,
784 timeoutSignal,
785 );
786 stepLog = maskSecrets(log, secretValues);
787 if (exitCode !== 0) {
788 stepStatus = "failure";
789 runFailed = true;
790 }
791 } catch (err) {
792 stepLog = `Step failed: ${err instanceof Error ? err.message : String(err)}\n`;
793 stepStatus = "failure";
794 runFailed = true;
795 }
796 }
797
798 // Collect artifacts for this step
799 if (!runFailed || stepStatus === "success") {
800 await collectArtifacts(
801 runId,
802 containerId,
803 step,
804 shell,
805 cfg.work_dir,
806 envArray,
807 ).catch(() => {});
808 }
809
810 await db
811 .updateTable("ci_steps")
812 .set({
813 status: stepStatus,
814 finished_at: now(),
815 log: stepLog,
816 })
817 .where("id", "=", stepId)
818 .execute();
819
820 if (runFailed) break;
821 }
822
823 // Mark remaining steps as skipped
824 await db
825 .updateTable("ci_steps")
826 .set({ status: "skipped", started_at: now(), finished_at: now() })
827 .where("run_id", "=", runId)
828 .where("status", "=", "pending")
829 .execute();
830
831 const finalStatus = runFailed ? "failure" : "success";
832 await db
833 .updateTable("ci_runs")
834 .set({ status: finalStatus, finished_at: now() })
835 .where("id", "=", runId)
836 .execute();
837 } catch (err) {
838 const status = signal.aborted ? "cancelled" : "failure";
839 const errMsg = err instanceof Error ? err.message : String(err);
840 // Write error to a synthetic step if we have no steps yet
841 const hasSteps = await db
842 .selectFrom("ci_steps")
843 .select("id")
844 .where("run_id", "=", runId)
845 .executeTakeFirst();
846 if (!hasSteps) {
847 await db
848 .insertInto("ci_steps")
849 .values({
850 run_id: runId,
851 name: "setup",
852 status: "failure",
853 started_at: new Date().toISOString(),
854 finished_at: new Date().toISOString(),
855 log: `Error: ${errMsg}\n`,
856 })
857 .execute();
858 }
859 await db
860 .updateTable("ci_runs")
861 .set({ status, finished_at: new Date().toISOString() })
862 .where("id", "=", runId)
863 .execute();
864 // Mark pending steps as skipped
865 await db
866 .updateTable("ci_steps")
867 .set({ status: "skipped", finished_at: new Date().toISOString() })
868 .where("run_id", "=", runId)
869 .where("status", "=", "pending")
870 .execute();
871 } finally {
872 if (containerId) await removeContainer(containerId);
873 runningTasks.delete(runId);
874 // Prune old history
875 const run = await db
876 .selectFrom("ci_runs")
877 .select("repo_id")
878 .where("id", "=", runId)
879 .executeTakeFirst();
880 if (run) await pruneHistory(run.repo_id).catch(() => {});
881 }
882}
883
884// --- Public API ---
885
886export async function triggerRun(
887 repoName: string,
888 opts: TriggerOpts,
889): Promise<number> {
890 const repo = await db
891 .selectFrom("repositories")
892 .select("id")
893 .where("name", "=", repoName)
894 .executeTakeFirst();
895 if (!repo) throw new Error("Repository not found");
896
897 const runId = await db
898 .insertInto("ci_runs")
899 .values({
900 repo_id: repo.id,
901 triggered_by: opts.triggeredBy ?? null,
902 trigger_source: opts.triggerSource,
903 commit_sha: opts.commitSha,
904 commit_branch: opts.commitBranch ?? null,
905 commit_tag: opts.commitTag ?? null,
906 status: "pending",
907 variable_overrides: opts.variableOverrides
908 ? JSON.stringify(opts.variableOverrides)
909 : null,
910 })
911 .returning("id")
912 .executeTakeFirstOrThrow();
913
914 const controller = new AbortController();
915 runningTasks.set(runId.id, { controller });
916
917 // Fire and forget — like release archiving
918 (async () => {
919 await executeRun(runId.id, controller.signal);
920 })();
921
922 return runId.id;
923}
924
925export async function cancelRun(runId: number): Promise<void> {
926 const task = runningTasks.get(runId);
927 if (task) {
928 const { containerId } = task;
929 task.controller.abort();
930 if (containerId) {
931 await removeContainer(containerId).catch(() => {});
932 }
933 }
934 await db
935 .updateTable("ci_runs")
936 .set({ status: "cancelled", finished_at: new Date().toISOString() })
937 .where("id", "=", runId)
938 .where("status", "in", ["pending", "running"])
939 .execute();
940}
941
942async function pruneHistory(repoId: number): Promise<void> {
943 const maxHistory = config.CI_MAX_HISTORY;
944 const allRuns = await db
945 .selectFrom("ci_runs")
946 .select("id")
947 .where("repo_id", "=", repoId)
948 .orderBy("id", "desc")
949 .execute();
950
951 if (allRuns.length <= maxHistory) return;
952
953 const toDelete = allRuns.slice(maxHistory).map((r) => r.id);
954 for (const runId of toDelete) {
955 // Remove artifacts from disk
956 const artifactDir = path.join(paths.CI_ARTIFACTS_DIR, String(runId));
957 if (existsSync(artifactDir)) {
958 await Bun.$`rm -rf ${artifactDir}`.quiet().nothrow();
959 }
960 }
961 await db.deleteFrom("ci_runs").where("id", "in", toDelete).execute();
962}
963
964/** Reset the cached socket path (used in tests to switch mock sockets). */
965export function resetDockerSocket(): void {
966 resolvedSocket = null;
967}
968
969/** Check if CI can connect to the container socket. */
970export async function ciAvailable(): Promise<boolean> {
971 try {
972 const socket = await getSocket();
973 const resp = await fetch("http://localhost/v1.47/info", {
974 unix: socket,
975 });
976 return resp.ok;
977 } catch {
978 return false;
979 }
980}
981