Repository Public
1/** Content-addressed artifacts: <dataDir>/artifacts/<sha256[0:2]>/<sha256>, written atomically. */2import { createHash, randomBytes } from 'node:crypto';3import fs from 'node:fs';4import path from 'node:path';5import { Readable } from 'node:stream';6import { getKernel, num, withTx, type Kernel } from './db.js';7import { KernelError } from './errors.js';89const SHA_RE = /^[0-9a-f]{64}$/;1011export interface PutArtifactMeta {12 name?: string;13 mime?: string;14 /** When set, the artifact is linked to the run (run_artifacts) and `artifact.created` is emitted once. */15 runId?: string;16 stepSeq?: number | null;17}1819export function artifactPath(sha256: string, k: Kernel = getKernel()): string {20 if (!SHA_RE.test(sha256)) throw new KernelError('invalid artifact sha256');21 return path.join(k.dataDir, 'artifacts', sha256.slice(0, 2), sha256);22}2324/**25 * Stores bytes (Buffer / string / Readable). Atomic: data is streamed to a temp file in the same26 * filesystem while hashing, fsynced, then renamed into place. Idempotent for identical content, and27 * not fenced: an artifact is immutable and harmless if its step is later re-executed.28 */29export async function putArtifact(30 data: Buffer | Uint8Array | string | Readable,31 meta: PutArtifactMeta = {},32 k: Kernel = getKernel(),33): Promise<{ sha256: string; size: number }> {34 const root = path.join(k.dataDir, 'artifacts');35 const tmpDir = path.join(root, '.tmp');36 await fs.promises.mkdir(tmpDir, { recursive: true });37 const tmp = path.join(tmpDir, `${process.pid}-${randomBytes(8).toString('hex')}`);38 const hash = createHash('sha256');39 let size = 0;40 try {41 const src = data instanceof Readable ? data : Readable.from([typeof data === 'string' ? Buffer.from(data, 'utf8') : Buffer.from(data)]);42 const fh = await fs.promises.open(tmp, 'wx', 0o600);43 try {44 for await (const chunk of src as AsyncIterable<Buffer | string>) {45 const buf = typeof chunk === 'string' ? Buffer.from(chunk, 'utf8') : chunk;46 hash.update(buf);47 size += buf.length;48 await fh.write(buf);49 }50 await fh.sync();51 } finally {52 await fh.close();53 }54 const sha256 = hash.digest('hex');55 const dest = artifactPath(sha256, k);56 await fs.promises.mkdir(path.dirname(dest), { recursive: true });57 await fs.promises.rename(tmp, dest); // atomic; same content if it already existed5859 await withTx(k, async (tx) => {60 await tx.q(`INSERT INTO artifacts (sha256, size, mime) VALUES ($1,$2,$3) ON CONFLICT (sha256) DO NOTHING`, [61 sha256,62 size,63 meta.mime ?? 'application/octet-stream',64 ]);65 if (!meta.runId) return;66 const run = await tx.one<{ mission_id: string }>('SELECT mission_id FROM runs WHERE id = $1', [meta.runId]);67 if (!run) throw new KernelError(`putArtifact: run ${meta.runId} not found`);68 const name = meta.name ?? sha256.slice(0, 12);69 const ins = await tx.q(70 `INSERT INTO run_artifacts (run_id, sha256, name, step_seq) VALUES ($1,$2,$3,$4) ON CONFLICT DO NOTHING RETURNING sha256`,71 [meta.runId, sha256, name, meta.stepSeq ?? null],72 );73 if (ins.length > 0) {74 tx.emit({75 missionId: run.mission_id,76 runId: meta.runId,77 seq: meta.stepSeq ?? null,78 kind: 'artifact.created',79 payload: { sha256, name, size, mime: meta.mime ?? 'application/octet-stream' },80 });81 }82 });83 return { sha256, size };84 } finally {85 await fs.promises.rm(tmp, { force: true });86 }87}8889export interface OpenedArtifact {90 sha256: string;91 size: number;92 mime: string;93 path: string;94 stream(): Readable;95}9697/** Returns null when the artifact is unknown or its file is missing. */98export async function openArtifact(sha256: string, k: Kernel = getKernel()): Promise<OpenedArtifact | null> {99 if (!SHA_RE.test(sha256)) return null;100 const p = artifactPath(sha256, k);101 const r = await k.pool.query('SELECT size, mime FROM artifacts WHERE sha256 = $1', [sha256]);102 const row = r.rows[0] as { size: unknown; mime: string } | undefined;103 let st: fs.Stats;104 try {105 st = await fs.promises.stat(p);106 } catch {107 return null;108 }109 return {110 sha256,111 size: row ? num(row.size) : st.size,112 mime: row?.mime ?? 'application/octet-stream',113 path: p,114 stream: () => fs.createReadStream(p),115 };116}117