Repository Public MIT
1import { createHash } from "node:crypto";2import { createReadStream, createWriteStream, existsSync, mkdirSync, renameSync, rmSync, statSync, chmodSync, readFileSync } from "node:fs";3import { join } from "node:path";4import { env } from "./env";5import { ApiError } from "./errors";67/**8 * Filesystem object store (ADR-0003). Release blobs are content-addressed by SHA-256, written once9 * through an atomic rename and then made read-only. The StorageDriver boundary is what an10 * S3-compatible driver would implement.11 */12const HEX64 = /^[0-9a-f]{64}$/;13const UUID = /^[0-9a-f-]{36}$/;1415function dir(kind: "blobs" | "staging"): string {16 const d = join(env.dataDir, kind);17 if (!existsSync(d)) mkdirSync(d, { recursive: true, mode: 0o700 });18 return d;19}2021export function blobKey(sha256: string): string {22 if (!HEX64.test(sha256)) throw new Error("invalid sha256");23 return `sha256/${sha256.slice(0, 2)}/${sha256}.tar.gz`;24}25function blobPath(key: string): string {26 if (!/^sha256\/[0-9a-f]{2}\/[0-9a-f]{64}\.tar\.gz$/.test(key)) throw new Error("invalid object key");27 return join(dir("blobs"), key);28}29function stagingPath(uploadId: string): string {30 if (!UUID.test(uploadId)) throw new Error("invalid upload id");31 return join(dir("staging"), `${uploadId}.part`);32}3334/** Stream an upload body to staging with a hard byte cap; returns the digest of what was received. */35export async function receiveToStaging(uploadId: string, body: ReadableStream<Uint8Array> | null, maxBytes: number): Promise<{ sha256: string; size: number }> {36 if (!body) throw new ApiError(400, "empty_body", "Upload body is empty.");37 const path = stagingPath(uploadId);38 const hash = createHash("sha256");39 const out = createWriteStream(path, { mode: 0o600, flags: "w" });40 let size = 0;41 const reader = body.getReader();42 try {43 for (;;) {44 const { done, value } = await reader.read();45 if (done) break;46 size += value.byteLength;47 if (size > maxBytes) throw new ApiError(413, "payload_too_large", `Upload exceeds the declared size of ${maxBytes} bytes.`);48 hash.update(value);49 if (!out.write(value)) await new Promise<void>((r) => out.once("drain", () => r()));50 }51 await new Promise<void>((res, rej) => out.end((e?: Error | null) => (e ? rej(e) : res())));52 } catch (e) {53 out.destroy();54 rmSync(path, { force: true });55 await reader.cancel().catch(() => {});56 throw e;57 }58 return { sha256: hash.digest("hex"), size };59}6061export function readStaging(uploadId: string): Buffer {62 return readFileSync(stagingPath(uploadId));63}64export function discardStaging(uploadId: string) {65 rmSync(stagingPath(uploadId), { force: true });66}6768/** Promote a staged upload to an immutable blob. Idempotent: identical content maps to the same key. */69export function promoteStaging(uploadId: string, sha256: string): string {70 const key = blobKey(sha256);71 const dest = blobPath(key);72 if (existsSync(dest)) {73 discardStaging(uploadId);74 return key;75 }76 mkdirSync(join(dir("blobs"), "sha256", sha256.slice(0, 2)), { recursive: true, mode: 0o700 });77 renameSync(stagingPath(uploadId), dest);78 chmodSync(dest, 0o440);79 return key;80}8182export function openBlob(key: string): { stream: ReadableStream<Uint8Array>; size: number } {83 const path = blobPath(key);84 const size = statSync(path).size;85 const rs = createReadStream(path);86 const stream = new ReadableStream<Uint8Array>({87 start(c) {88 rs.on("data", (chunk) => c.enqueue(new Uint8Array(chunk as Buffer)));89 rs.on("end", () => c.close());90 rs.on("error", (e) => c.error(e));91 },92 cancel() {93 rs.destroy();94 },95 });96 return { stream, size };97}98export function readBlob(key: string): Buffer {99 return readFileSync(blobPath(key));100}101export function storageWritable(): boolean {102 try {103 dir("blobs");104 dir("staging");105 return true;106 } catch {107 return false;108 }109}110