spb/autoscholar
Public
1import { createHash } from "node:crypto";2import { createReadStream, createWriteStream } from "node:fs";3import { access, mkdir, readdir, readFile, rename, rm, stat, writeFile } from "node:fs/promises";4import { dirname, join, normalize, resolve, sep } from "node:path";5import type { Readable } from "node:stream";6import { pipeline } from "node:stream/promises";78export interface StoredObject {9 key: string;10 sha256: string;11 bytes: number;12}1314/** Write-once, checksum-verified object storage. Filesystem backend by default; S3 can implement the same interface. */15export interface StorageBackend {16 put(key: string, data: Buffer | Readable, expectedSha256?: string): Promise<StoredObject>;17 get(key: string): Promise<Buffer>;18 stream(key: string): Promise<Readable>;19 head(key: string): Promise<StoredObject | null>;20 exists(key: string): Promise<boolean>;21 copy(fromKey: string, toKey: string): Promise<StoredObject>;22 list(prefix: string): Promise<string[]>;23 remove(key: string): Promise<void>;24 absolutePath(key: string): string;25}2627export function sha256Hex(data: Buffer | string): string {28 return createHash("sha256").update(data).digest("hex");29}3031export async function sha256File(path: string): Promise<string> {32 const h = createHash("sha256");33 for await (const chunk of createReadStream(path)) h.update(chunk as Buffer);34 return h.digest("hex");35}3637export function assertSafeKey(key: string): string {38 const n = normalize(key).replace(/\\/g, "/");39 if (!n || n.startsWith("/") || n.startsWith("..") || n.split("/").includes("..") || n.includes("\0")) {40 throw new Error(`unsafe storage key: ${key}`);41 }42 return n;43}4445export class FilesystemStorage implements StorageBackend {46 readonly root: string;47 constructor(root: string) {48 this.root = resolve(root);49 }5051 absolutePath(key: string): string {52 const safe = assertSafeKey(key);53 const abs = resolve(this.root, safe);54 if (!abs.startsWith(this.root + sep) && abs !== this.root) throw new Error(`key escapes root: ${key}`);55 return abs;56 }5758 /** Digest sidecar. The suffix cannot collide with user artifact names (research code often writes `*.sha256`). */59 private sidecar(abs: string): string {60 return `${abs}.~digest`;61 }6263 async put(key: string, data: Buffer | Readable, expectedSha256?: string): Promise<StoredObject> {64 const abs = this.absolutePath(key);65 if (await this.exists(key)) {66 const existing = await this.head(key);67 if (existing && expectedSha256 && existing.sha256 !== expectedSha256) {68 throw new Error(`object exists with different digest: ${key}`);69 }70 if (existing) return existing; // write-once: identical content is idempotent71 }72 await mkdir(dirname(abs), { recursive: true });73 const tmp = `${abs}.tmp-${process.pid}-${Date.now()}`;74 const h = createHash("sha256");75 let bytes = 0;76 if (Buffer.isBuffer(data)) {77 h.update(data);78 bytes = data.length;79 await writeFile(tmp, data, { flag: "wx" });80 } else {81 await pipeline(82 data,83 async function* (src) {84 for await (const chunk of src) {85 const b = chunk as Buffer;86 h.update(b);87 bytes += b.length;88 yield b;89 }90 },91 createWriteStream(tmp, { flags: "wx" }),92 );93 }94 const sha256 = h.digest("hex");95 if (expectedSha256 && expectedSha256 !== sha256) {96 await rm(tmp, { force: true });97 throw new Error(`digest mismatch for ${key}: expected ${expectedSha256}, got ${sha256}`);98 }99 await rename(tmp, abs);100 await writeFile(this.sidecar(abs), `${sha256} ${key}\n`);101 return { key: assertSafeKey(key), sha256, bytes };102 }103104 async get(key: string): Promise<Buffer> {105 const abs = this.absolutePath(key);106 const buf = await readFile(abs);107 const meta = await this.head(key);108 if (meta && sha256Hex(buf) !== meta.sha256) throw new Error(`integrity failure reading ${key}`);109 return buf;110 }111112 async stream(key: string): Promise<Readable> {113 return createReadStream(this.absolutePath(key));114 }115116 async head(key: string): Promise<StoredObject | null> {117 const abs = this.absolutePath(key);118 try {119 const s = await stat(abs);120 let sha256: string;121 try {122 const side = await readFile(this.sidecar(abs), "utf8");123 sha256 = side.slice(0, 64);124 } catch {125 sha256 = await sha256File(abs);126 }127 return { key: assertSafeKey(key), sha256, bytes: s.size };128 } catch {129 return null;130 }131 }132133 async exists(key: string): Promise<boolean> {134 try {135 await access(this.absolutePath(key));136 return true;137 } catch {138 return false;139 }140 }141142 async copy(fromKey: string, toKey: string): Promise<StoredObject> {143 const buf = await this.get(fromKey);144 return this.put(toKey, buf);145 }146147 async list(prefix: string): Promise<string[]> {148 const abs = this.absolutePath(prefix);149 const out: string[] = [];150 const walk = async (dir: string, rel: string) => {151 let entries: import("node:fs").Dirent[];152 try {153 entries = (await readdir(dir, { withFileTypes: true })) as import("node:fs").Dirent[];154 } catch {155 return;156 }157 for (const e of entries) {158 const name = String(e.name);159 if (name.endsWith(".~digest") || name.includes(".tmp-")) continue;160 const r = rel ? `${rel}/${name}` : name;161 if (e.isDirectory()) await walk(join(dir, name), r);162 else out.push(`${assertSafeKey(prefix)}/${r}`.replace(/^\.\//, ""));163 }164 };165 await walk(abs, "");166 return out.sort();167 }168169 async remove(key: string): Promise<void> {170 const abs = this.absolutePath(key);171 await rm(abs, { force: true });172 await rm(this.sidecar(abs), { force: true });173 }174}175176export function runArtifactKey(runId: string, attemptNo: number, path: string): string {177 return assertSafeKey(`runs/${runId}/a${attemptNo}/${path}`);178}179180export function publicArtifactKey(publicId: string, versionNo: number, path: string): string {181 return assertSafeKey(`public/${publicId}/v${versionNo}/${path}`);182}183184export function getStorage(root = process.env.ARTIFACT_ROOT ?? "./data/artifacts"): FilesystemStorage {185 return new FilesystemStorage(root);186}187