feat: complete TASK-WP0-05 storage capacity
This commit is contained in:
suyx
@@ -0,0 +1,614 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import {
|
||||
createWriteStream,
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
readdirSync,
|
||||
renameSync,
|
||||
rmSync,
|
||||
statSync,
|
||||
} from "node:fs";
|
||||
import { createRequire } from "node:module";
|
||||
import { basename, dirname, extname, join, relative, resolve } from "node:path";
|
||||
import { Readable, Transform } from "node:stream";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
|
||||
import type BetterSqlite3 from "better-sqlite3";
|
||||
|
||||
import { resolvePathWithinRoot } from "./local-data-root.js";
|
||||
import {
|
||||
HARD_LIMIT_BYTES,
|
||||
classifyCapacity,
|
||||
decideStorageAction,
|
||||
storageCapacityErrorDetails,
|
||||
type StorageAction,
|
||||
} from "./storage-policy.js";
|
||||
|
||||
const require = createRequire(import.meta.url);
|
||||
const Database = require("better-sqlite3") as typeof BetterSqlite3;
|
||||
const uuidPattern = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
|
||||
const safeFileNamePattern = /^[A-Za-z0-9][A-Za-z0-9._-]{0,159}$/;
|
||||
|
||||
export { HARD_LIMIT_BYTES } from "./storage-policy.js";
|
||||
|
||||
export type ManagedFileKind =
|
||||
| "reference"
|
||||
| "generated"
|
||||
| "export"
|
||||
| "derived"
|
||||
| "sticker_original"
|
||||
| "sticker_thumbnail";
|
||||
export type CommitFailurePoint = "after_staging" | "after_rename" | "after_database_commit";
|
||||
|
||||
interface StorageStateRow {
|
||||
active_storage_reservations_bytes: number;
|
||||
capacity_notice_level: "normal" | "warning" | "critical";
|
||||
data_root_ref: "configured_local_data_root";
|
||||
hard_limit_bytes: number;
|
||||
managed_content_bytes: number;
|
||||
measured_at: string;
|
||||
storage_backend: "local_data_root";
|
||||
storage_status: "active" | "full" | "unavailable";
|
||||
version: number;
|
||||
}
|
||||
|
||||
interface ManagedFileRow {
|
||||
byte_size: number;
|
||||
file_id: string;
|
||||
file_kind: ManagedFileKind;
|
||||
relative_path: string;
|
||||
status: "committed" | "purged";
|
||||
}
|
||||
|
||||
interface CleanupQueueRow {
|
||||
byte_size: number;
|
||||
cleanup_id: string;
|
||||
counts_toward_managed: 0 | 1;
|
||||
managed_file_id: string | null;
|
||||
relative_path: string;
|
||||
}
|
||||
|
||||
export class StorageCapacityError extends Error {
|
||||
readonly code = "STORAGE_CAPACITY_EXCEEDED";
|
||||
readonly httpStatus = 507;
|
||||
readonly details: ReturnType<typeof storageCapacityErrorDetails>;
|
||||
|
||||
constructor(input: { activeReservationBytes: number; managedContentBytes: number; projectedWriteBytes: number }) {
|
||||
super("STORAGE_CAPACITY_EXCEEDED");
|
||||
this.details = storageCapacityErrorDetails(input);
|
||||
}
|
||||
}
|
||||
|
||||
export class StorageUnavailableError extends Error {
|
||||
readonly code = "storage_unavailable";
|
||||
|
||||
constructor() {
|
||||
super("storage_unavailable");
|
||||
}
|
||||
}
|
||||
|
||||
class InjectedCrashError extends Error {}
|
||||
|
||||
function now() {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
function validatePositiveBytes(value: number, name: string) {
|
||||
if (!Number.isSafeInteger(value) || value <= 0) throw new Error(`${name}_invalid`);
|
||||
}
|
||||
|
||||
function sniffMime(prefix: Buffer) {
|
||||
if (prefix.length >= 8 && prefix.subarray(0, 8).equals(Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]))) {
|
||||
return "image/png";
|
||||
}
|
||||
if (prefix.length >= 3 && prefix[0] === 0xff && prefix[1] === 0xd8 && prefix[2] === 0xff) return "image/jpeg";
|
||||
return "application/octet-stream";
|
||||
}
|
||||
|
||||
function listFiles(root: string): string[] {
|
||||
if (!existsSync(root)) return [];
|
||||
const files: string[] = [];
|
||||
for (const entry of readdirSync(root, { withFileTypes: true })) {
|
||||
const child = join(root, entry.name);
|
||||
if (entry.isDirectory()) files.push(...listFiles(child));
|
||||
else if (entry.isFile()) files.push(child);
|
||||
}
|
||||
return files;
|
||||
}
|
||||
|
||||
export interface CommitStreamInput {
|
||||
content: Readable;
|
||||
expectedMimeType: "image/png" | "image/jpeg" | "application/octet-stream";
|
||||
expectedSha256?: string;
|
||||
failurePoint?: CommitFailurePoint;
|
||||
fileKind: ManagedFileKind;
|
||||
fileName: string;
|
||||
operationId: string;
|
||||
ownerRef: string;
|
||||
projectedWriteBytes: number;
|
||||
}
|
||||
|
||||
export class ManagedStorage {
|
||||
readonly dataRoot: string;
|
||||
readonly databasePath: string;
|
||||
private readonly database: BetterSqlite3.Database;
|
||||
private dataRootWritable = true;
|
||||
private diskSpaceAvailable = true;
|
||||
private sqliteWritable = true;
|
||||
private measurementBaselineBytes = 0;
|
||||
private recoveryRequiresRemeasure = false;
|
||||
|
||||
constructor(input: { dataRoot: string; databasePath: string }) {
|
||||
this.dataRoot = resolve(input.dataRoot);
|
||||
this.databasePath = resolve(input.databasePath);
|
||||
if (!this.databasePath.startsWith(`${this.dataRoot}${process.platform === "win32" ? "\\" : "/"}`)) {
|
||||
throw new Error("database_outside_data_root");
|
||||
}
|
||||
mkdirSync(dirname(this.databasePath), { recursive: true });
|
||||
this.database = new Database(this.databasePath);
|
||||
this.database.pragma("journal_mode = WAL");
|
||||
this.database.pragma("foreign_keys = ON");
|
||||
this.database.pragma("busy_timeout = 5000");
|
||||
this.migrate();
|
||||
const state = this.database.prepare("SELECT managed_content_bytes FROM local_backend_storage_state WHERE singleton = 1").get() as { managed_content_bytes: number };
|
||||
this.measurementBaselineBytes = Math.max(0, state.managed_content_bytes - this.physicalManagedBytes());
|
||||
}
|
||||
|
||||
private migrate() {
|
||||
this.database.exec(`
|
||||
CREATE TABLE IF NOT EXISTS local_backend_storage_state (
|
||||
singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
|
||||
storage_backend TEXT NOT NULL CHECK (storage_backend = 'local_data_root'),
|
||||
data_root_ref TEXT NOT NULL CHECK (data_root_ref = 'configured_local_data_root'),
|
||||
hard_limit_bytes INTEGER NOT NULL,
|
||||
managed_content_bytes INTEGER NOT NULL CHECK (managed_content_bytes >= 0),
|
||||
capacity_notice_level TEXT NOT NULL CHECK (capacity_notice_level IN ('normal', 'warning', 'critical')),
|
||||
storage_status TEXT NOT NULL CHECK (storage_status IN ('active', 'full', 'unavailable')),
|
||||
measured_at TEXT NOT NULL,
|
||||
version INTEGER NOT NULL CHECK (version >= 1)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS storage_reservations (
|
||||
reservation_id TEXT PRIMARY KEY,
|
||||
operation_id TEXT NOT NULL UNIQUE,
|
||||
projected_bytes INTEGER NOT NULL CHECK (projected_bytes > 0),
|
||||
status TEXT NOT NULL CHECK (status IN ('active', 'consumed', 'released')),
|
||||
created_at TEXT NOT NULL,
|
||||
resolved_at TEXT
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS managed_files (
|
||||
file_id TEXT PRIMARY KEY,
|
||||
file_kind TEXT NOT NULL CHECK (file_kind IN ('reference', 'generated', 'export', 'derived', 'sticker_original', 'sticker_thumbnail')),
|
||||
relative_path TEXT NOT NULL UNIQUE,
|
||||
byte_size INTEGER NOT NULL CHECK (byte_size > 0),
|
||||
mime_type TEXT NOT NULL,
|
||||
sha256 TEXT NOT NULL,
|
||||
status TEXT NOT NULL CHECK (status IN ('committed', 'purged')),
|
||||
created_at TEXT NOT NULL,
|
||||
purged_at TEXT
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS file_cleanup_queue (
|
||||
cleanup_id TEXT PRIMARY KEY,
|
||||
managed_file_id TEXT,
|
||||
relative_path TEXT NOT NULL UNIQUE,
|
||||
byte_size INTEGER NOT NULL CHECK (byte_size >= 0),
|
||||
counts_toward_managed INTEGER NOT NULL CHECK (counts_toward_managed IN (0, 1)),
|
||||
reason TEXT NOT NULL CHECK (reason IN ('compensation', 'purge')),
|
||||
status TEXT NOT NULL CHECK (status IN ('pending', 'completed', 'failed')),
|
||||
created_at TEXT NOT NULL,
|
||||
completed_at TEXT,
|
||||
last_error TEXT
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS project_asset_refs (
|
||||
reference_id TEXT PRIMARY KEY,
|
||||
managed_file_id TEXT NOT NULL,
|
||||
reference_type TEXT NOT NULL CHECK (reference_type IN ('project', 'release')),
|
||||
created_at TEXT NOT NULL,
|
||||
FOREIGN KEY (managed_file_id) REFERENCES managed_files(file_id)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS asset_cleanup_requests (
|
||||
request_id TEXT PRIMARY KEY,
|
||||
status TEXT NOT NULL CHECK (status IN ('pending_confirmation', 'denied', 'queued', 'completed')),
|
||||
created_at TEXT NOT NULL,
|
||||
confirmed_at TEXT
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS asset_cleanup_request_items (
|
||||
request_id TEXT NOT NULL,
|
||||
managed_file_id TEXT NOT NULL,
|
||||
PRIMARY KEY (request_id, managed_file_id),
|
||||
FOREIGN KEY (request_id) REFERENCES asset_cleanup_requests(request_id),
|
||||
FOREIGN KEY (managed_file_id) REFERENCES managed_files(file_id)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS admin_operation_logs (
|
||||
log_id TEXT PRIMARY KEY,
|
||||
operation TEXT NOT NULL,
|
||||
outcome TEXT NOT NULL,
|
||||
target_ref TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
`);
|
||||
const initial = classifyCapacity(0, 0);
|
||||
this.database.prepare(`
|
||||
INSERT OR IGNORE INTO local_backend_storage_state
|
||||
(singleton, storage_backend, data_root_ref, hard_limit_bytes, managed_content_bytes, capacity_notice_level, storage_status, measured_at, version)
|
||||
VALUES (1, 'local_data_root', 'configured_local_data_root', ?, 0, ?, ?, ?, 1)
|
||||
`).run(HARD_LIMIT_BYTES, initial.capacity_notice_level, initial.storage_status, now());
|
||||
}
|
||||
|
||||
close() {
|
||||
this.database.close();
|
||||
}
|
||||
|
||||
getState(): StorageStateRow {
|
||||
const state = this.database.prepare("SELECT * FROM local_backend_storage_state WHERE singleton = 1").get() as Omit<StorageStateRow, "active_storage_reservations_bytes">;
|
||||
const withReservations = { ...state, active_storage_reservations_bytes: this.activeReservationBytes() };
|
||||
if (!this.dataRootWritable || !this.diskSpaceAvailable || !this.sqliteWritable) {
|
||||
return { ...withReservations, storage_status: "unavailable" };
|
||||
}
|
||||
return withReservations;
|
||||
}
|
||||
|
||||
private activeReservationBytes(excludingOperationId?: string) {
|
||||
const row = this.database.prepare(`
|
||||
SELECT COALESCE(SUM(projected_bytes), 0) AS bytes
|
||||
FROM storage_reservations
|
||||
WHERE status = 'active' AND (? IS NULL OR operation_id <> ?)
|
||||
`).get(excludingOperationId ?? null, excludingOperationId ?? null) as { bytes: number };
|
||||
return row.bytes;
|
||||
}
|
||||
|
||||
private refreshState() {
|
||||
if (!this.sqliteWritable) return;
|
||||
const state = this.database.prepare("SELECT managed_content_bytes FROM local_backend_storage_state WHERE singleton = 1").get() as { managed_content_bytes: number };
|
||||
const classification = classifyCapacity(state.managed_content_bytes, this.activeReservationBytes());
|
||||
const storageStatus = !this.dataRootWritable || !this.diskSpaceAvailable || this.recoveryRequiresRemeasure
|
||||
? "unavailable"
|
||||
: classification.storage_status;
|
||||
this.database.prepare(`
|
||||
UPDATE local_backend_storage_state
|
||||
SET capacity_notice_level = ?, storage_status = ?, measured_at = ?, version = version + 1
|
||||
WHERE singleton = 1
|
||||
`).run(classification.capacity_notice_level, storageStatus, now());
|
||||
}
|
||||
|
||||
applyControlledMeasurement(bytes: number) {
|
||||
if (!Number.isSafeInteger(bytes) || bytes < 0) throw new Error("managed_content_bytes_invalid");
|
||||
this.measurementBaselineBytes = Math.max(0, bytes - this.physicalManagedBytes());
|
||||
this.recoveryRequiresRemeasure = false;
|
||||
const classification = classifyCapacity(bytes, this.activeReservationBytes());
|
||||
this.database.prepare(`
|
||||
UPDATE local_backend_storage_state
|
||||
SET managed_content_bytes = ?, capacity_notice_level = ?, storage_status = ?, measured_at = ?, version = version + 1
|
||||
WHERE singleton = 1
|
||||
`).run(bytes, classification.capacity_notice_level, classification.storage_status, now());
|
||||
return this.getState();
|
||||
}
|
||||
|
||||
setAvailability(input: { dataRootWritable: boolean; diskSpaceAvailable: boolean; sqliteWritable: boolean }) {
|
||||
this.dataRootWritable = input.dataRootWritable;
|
||||
this.diskSpaceAvailable = input.diskSpaceAvailable;
|
||||
this.sqliteWritable = input.sqliteWritable;
|
||||
if (!input.dataRootWritable || !input.diskSpaceAvailable || !input.sqliteWritable) this.recoveryRequiresRemeasure = true;
|
||||
this.refreshState();
|
||||
}
|
||||
|
||||
private physicalManagedBytes() {
|
||||
return ["content", "managed-assets", "derived-assets"]
|
||||
.flatMap((root) => listFiles(resolvePathWithinRoot(this.dataRoot, root)))
|
||||
.reduce((sum, path) => sum + statSync(path).size, 0);
|
||||
}
|
||||
|
||||
private recordPhysicalMeasurement() {
|
||||
const bytes = this.measurementBaselineBytes + this.physicalManagedBytes();
|
||||
this.recoveryRequiresRemeasure = false;
|
||||
const classification = classifyCapacity(bytes, this.activeReservationBytes());
|
||||
this.database.prepare(`
|
||||
UPDATE local_backend_storage_state
|
||||
SET managed_content_bytes = ?, capacity_notice_level = ?, storage_status = ?, measured_at = ?, version = version + 1
|
||||
WHERE singleton = 1
|
||||
`).run(bytes, classification.capacity_notice_level, classification.storage_status, now());
|
||||
}
|
||||
|
||||
inspectAction(action: StorageAction) {
|
||||
return decideStorageAction(this.getState().storage_status, this.sqliteWritable, action);
|
||||
}
|
||||
|
||||
private assertWritable() {
|
||||
if (this.inspectAction("binary_write") === "reject_unavailable") throw new StorageUnavailableError();
|
||||
}
|
||||
|
||||
private reserve(operationId: string, projectedWriteBytes: number) {
|
||||
this.assertWritable();
|
||||
if (!uuidPattern.test(operationId)) throw new Error("operation_id_invalid");
|
||||
validatePositiveBytes(projectedWriteBytes, "projected_write_bytes");
|
||||
const transaction = this.database.transaction(() => {
|
||||
const state = this.getState();
|
||||
const activeReservationBytes = this.activeReservationBytes();
|
||||
if (state.managed_content_bytes + activeReservationBytes + projectedWriteBytes > HARD_LIMIT_BYTES) {
|
||||
throw new StorageCapacityError({ activeReservationBytes, managedContentBytes: state.managed_content_bytes, projectedWriteBytes });
|
||||
}
|
||||
const reservationId = randomUUID();
|
||||
this.database.prepare(`
|
||||
INSERT INTO storage_reservations (reservation_id, operation_id, projected_bytes, status, created_at)
|
||||
VALUES (?, ?, ?, 'active', ?)
|
||||
`).run(reservationId, operationId, projectedWriteBytes, now());
|
||||
this.refreshState();
|
||||
return reservationId;
|
||||
});
|
||||
return transaction();
|
||||
}
|
||||
|
||||
private releaseReservation(operationId: string) {
|
||||
this.database.prepare(`
|
||||
UPDATE storage_reservations SET status = 'released', resolved_at = ?
|
||||
WHERE operation_id = ? AND status = 'active'
|
||||
`).run(now(), operationId);
|
||||
this.refreshState();
|
||||
}
|
||||
|
||||
private destination(input: CommitStreamInput, fileId: string) {
|
||||
if (!safeFileNamePattern.test(input.fileName) || basename(input.fileName) !== input.fileName || input.fileName.includes("..")) {
|
||||
throw new Error("file_name_invalid");
|
||||
}
|
||||
if (!uuidPattern.test(input.ownerRef)) throw new Error("owner_ref_invalid");
|
||||
const group = input.fileKind === "reference"
|
||||
? "content/references"
|
||||
: input.fileKind === "generated"
|
||||
? "content/generated"
|
||||
: input.fileKind === "export"
|
||||
? "content/exports"
|
||||
: input.fileKind === "derived"
|
||||
? "derived-assets"
|
||||
: input.fileKind === "sticker_original"
|
||||
? "managed-assets/stickers/original"
|
||||
: "managed-assets/stickers/thumbnail";
|
||||
const extension = extname(input.fileName).toLowerCase();
|
||||
const relativePath = `${group}/${input.ownerRef}/${fileId}${extension}`;
|
||||
return { absolutePath: resolvePathWithinRoot(this.dataRoot, relativePath), relativePath };
|
||||
}
|
||||
|
||||
async commitStream(input: CommitStreamInput) {
|
||||
const fileId = randomUUID();
|
||||
const destination = this.destination(input, fileId);
|
||||
this.reserve(input.operationId, input.projectedWriteBytes);
|
||||
const stagingDirectory = resolvePathWithinRoot(this.dataRoot, `staging/${input.operationId}`);
|
||||
const stagingPath = resolvePathWithinRoot(this.dataRoot, `staging/${input.operationId}/payload.tmp`);
|
||||
let renamed = false;
|
||||
try {
|
||||
mkdirSync(stagingDirectory, { recursive: true });
|
||||
const hash = createHash("sha256");
|
||||
let byteSize = 0;
|
||||
let prefix = Buffer.alloc(0);
|
||||
const inspect = new Transform({
|
||||
transform(chunk: Buffer | string, encoding, callback) {
|
||||
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding);
|
||||
byteSize += bytes.byteLength;
|
||||
hash.update(bytes);
|
||||
if (prefix.byteLength < 16) prefix = Buffer.concat([prefix, bytes.subarray(0, 16 - prefix.byteLength)]);
|
||||
callback(null, bytes);
|
||||
},
|
||||
});
|
||||
await pipeline(input.content, inspect, createWriteStream(stagingPath, { flags: "wx" }));
|
||||
validatePositiveBytes(byteSize, "actual_write_bytes");
|
||||
const sha256 = hash.digest("hex");
|
||||
if (input.expectedSha256 && sha256.toLowerCase() !== input.expectedSha256.toLowerCase()) throw new Error("content_hash_invalid");
|
||||
if (sniffMime(prefix) !== input.expectedMimeType) throw new Error("content_mime_invalid");
|
||||
const state = this.getState();
|
||||
const otherReservations = this.activeReservationBytes(input.operationId);
|
||||
if (state.managed_content_bytes + otherReservations + byteSize > HARD_LIMIT_BYTES) {
|
||||
throw new StorageCapacityError({ activeReservationBytes: otherReservations, managedContentBytes: state.managed_content_bytes, projectedWriteBytes: byteSize });
|
||||
}
|
||||
this.database.prepare(`UPDATE storage_reservations SET projected_bytes = ? WHERE operation_id = ? AND status = 'active'`).run(byteSize, input.operationId);
|
||||
this.refreshState();
|
||||
if (input.failurePoint === "after_staging") throw new InjectedCrashError("injected_crash_after_staging");
|
||||
mkdirSync(dirname(destination.absolutePath), { recursive: true });
|
||||
renameSync(stagingPath, destination.absolutePath);
|
||||
renamed = true;
|
||||
rmSync(stagingDirectory, { force: true, recursive: true });
|
||||
if (input.failurePoint === "after_rename") throw new InjectedCrashError("injected_crash_after_rename");
|
||||
|
||||
const commit = this.database.transaction(() => {
|
||||
this.database.prepare(`
|
||||
INSERT INTO managed_files (file_id, file_kind, relative_path, byte_size, mime_type, sha256, status, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, 'committed', ?)
|
||||
`).run(fileId, input.fileKind, destination.relativePath, byteSize, input.expectedMimeType, sha256, now());
|
||||
this.database.prepare(`
|
||||
UPDATE local_backend_storage_state SET managed_content_bytes = managed_content_bytes + ? WHERE singleton = 1
|
||||
`).run(byteSize);
|
||||
this.database.prepare(`
|
||||
UPDATE storage_reservations SET status = 'consumed', resolved_at = ? WHERE operation_id = ? AND status = 'active'
|
||||
`).run(now(), input.operationId);
|
||||
this.refreshState();
|
||||
});
|
||||
commit();
|
||||
if (input.failurePoint === "after_database_commit") throw new InjectedCrashError("injected_crash_after_database_commit");
|
||||
return { bytes: byteSize, file_id: fileId, relative_path: destination.relativePath, sha256 };
|
||||
} catch (error) {
|
||||
if (error instanceof InjectedCrashError) throw error;
|
||||
if (renamed && existsSync(destination.absolutePath)) {
|
||||
try {
|
||||
this.queueCompensation(destination.relativePath, statSync(destination.absolutePath).size);
|
||||
} catch {
|
||||
// Startup reconciliation is the final fallback if SQLite itself is unavailable.
|
||||
}
|
||||
} else {
|
||||
rmSync(stagingDirectory, { force: true, recursive: true });
|
||||
}
|
||||
this.releaseReservation(input.operationId);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async commitBufferFixture(fileKind: ManagedFileKind, fileName: string, bytes: Buffer) {
|
||||
return this.commitStream({
|
||||
content: Readable.from(bytes),
|
||||
expectedMimeType: sniffMime(bytes.subarray(0, 16)),
|
||||
fileKind,
|
||||
fileName,
|
||||
operationId: randomUUID(),
|
||||
ownerRef: randomUUID(),
|
||||
projectedWriteBytes: bytes.byteLength,
|
||||
});
|
||||
}
|
||||
|
||||
private queueCompensation(relativePath: string, byteSize: number) {
|
||||
this.database.prepare(`
|
||||
INSERT OR IGNORE INTO file_cleanup_queue
|
||||
(cleanup_id, managed_file_id, relative_path, byte_size, counts_toward_managed, reason, status, created_at)
|
||||
VALUES (?, NULL, ?, ?, 0, 'compensation', 'pending', ?)
|
||||
`).run(randomUUID(), relativePath, byteSize, now());
|
||||
}
|
||||
|
||||
async reconcileStartup() {
|
||||
let releasedReservations = 0;
|
||||
const active = this.database.prepare("SELECT operation_id FROM storage_reservations WHERE status = 'active'").all() as Array<{ operation_id: string }>;
|
||||
for (const reservation of active) {
|
||||
this.database.prepare(`UPDATE storage_reservations SET status = 'released', resolved_at = ? WHERE operation_id = ?`).run(now(), reservation.operation_id);
|
||||
releasedReservations += 1;
|
||||
}
|
||||
const stagingRoot = resolvePathWithinRoot(this.dataRoot, "staging");
|
||||
for (const entry of existsSync(stagingRoot) ? readdirSync(stagingRoot) : []) {
|
||||
rmSync(join(stagingRoot, entry), { force: true, recursive: true });
|
||||
}
|
||||
|
||||
const managedPaths = new Set((this.database.prepare("SELECT relative_path FROM managed_files").all() as Array<{ relative_path: string }>).map((row) => row.relative_path));
|
||||
const queuedPaths = new Set((this.database.prepare("SELECT relative_path FROM file_cleanup_queue WHERE status IN ('pending', 'failed')").all() as Array<{ relative_path: string }>).map((row) => row.relative_path));
|
||||
let orphaned = 0;
|
||||
for (const root of ["content", "managed-assets", "derived-assets"]) {
|
||||
for (const path of listFiles(resolvePathWithinRoot(this.dataRoot, root))) {
|
||||
const relativePath = relative(this.dataRoot, path).replaceAll("\\", "/");
|
||||
if (!managedPaths.has(relativePath) && !queuedPaths.has(relativePath)) {
|
||||
this.queueCompensation(relativePath, statSync(path).size);
|
||||
orphaned += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
const missingManagedFile = (this.database.prepare("SELECT relative_path FROM managed_files WHERE status = 'committed'").all() as Array<{ relative_path: string }>).some(
|
||||
(row) => !existsSync(resolvePathWithinRoot(this.dataRoot, row.relative_path)),
|
||||
);
|
||||
if (missingManagedFile) this.setAvailability({ dataRootWritable: false, diskSpaceAvailable: true, sqliteWritable: true });
|
||||
else this.recordPhysicalMeasurement();
|
||||
return { orphaned, released_reservations: releasedReservations };
|
||||
}
|
||||
|
||||
inspectCounts() {
|
||||
const count = (table: string, where = "1 = 1") => (this.database.prepare(`SELECT COUNT(*) AS count FROM ${table} WHERE ${where}`).get() as { count: number }).count;
|
||||
return {
|
||||
active_reservations: count("storage_reservations", "status = 'active'"),
|
||||
managed_files: count("managed_files"),
|
||||
pending_cleanup: count("file_cleanup_queue", "status = 'pending'"),
|
||||
cleanup_requests: count("asset_cleanup_requests"),
|
||||
admin_logs: count("admin_operation_logs"),
|
||||
};
|
||||
}
|
||||
|
||||
resolveManagedFile(fileId: string) {
|
||||
const row = this.database.prepare("SELECT relative_path FROM managed_files WHERE file_id = ? AND status = 'committed'").get(fileId) as { relative_path: string } | undefined;
|
||||
return row ? resolvePathWithinRoot(this.dataRoot, row.relative_path) : undefined;
|
||||
}
|
||||
|
||||
addAssetReference(fileId: string, referenceType: "project" | "release") {
|
||||
if (this.inspectAction("project_json_write") !== "allow") throw new StorageUnavailableError();
|
||||
this.database.prepare(`INSERT INTO project_asset_refs (reference_id, managed_file_id, reference_type, created_at) VALUES (?, ?, ?, ?)`)
|
||||
.run(randomUUID(), fileId, referenceType, now());
|
||||
}
|
||||
|
||||
removeAssetReferences(fileId: string) {
|
||||
if (this.inspectAction("project_json_write") !== "allow") throw new StorageUnavailableError();
|
||||
this.database.prepare("DELETE FROM project_asset_refs WHERE managed_file_id = ?").run(fileId);
|
||||
}
|
||||
|
||||
createCleanupIntent(fileIds: string[]) {
|
||||
if (this.inspectAction("explicit_cleanup") !== "allow") throw new Error("cleanup_uncommitted");
|
||||
if (fileIds.length === 0 || new Set(fileIds).size !== fileIds.length) throw new Error("cleanup_candidates_invalid");
|
||||
const requestId = randomUUID();
|
||||
const transaction = this.database.transaction(() => {
|
||||
for (const fileId of fileIds) {
|
||||
const candidate = this.database.prepare(`
|
||||
SELECT file_id FROM managed_files
|
||||
WHERE file_id = ? AND status = 'committed' AND file_kind IN ('sticker_original', 'sticker_thumbnail')
|
||||
`).get(fileId);
|
||||
if (!candidate) throw new Error("ASSET_CLEANUP_CANDIDATE_STALE");
|
||||
}
|
||||
this.database.prepare("INSERT INTO asset_cleanup_requests (request_id, status, created_at) VALUES (?, 'pending_confirmation', ?)").run(requestId, now());
|
||||
const insert = this.database.prepare("INSERT INTO asset_cleanup_request_items (request_id, managed_file_id) VALUES (?, ?)");
|
||||
for (const fileId of fileIds) insert.run(requestId, fileId);
|
||||
});
|
||||
transaction();
|
||||
return { request_id: requestId };
|
||||
}
|
||||
|
||||
confirmCleanupIntent(requestId: string) {
|
||||
if (this.inspectAction("explicit_cleanup") !== "allow") throw new Error("cleanup_uncommitted");
|
||||
const transaction = this.database.transaction(() => {
|
||||
const request = this.database.prepare("SELECT status FROM asset_cleanup_requests WHERE request_id = ?").get(requestId) as { status: string } | undefined;
|
||||
if (request?.status !== "pending_confirmation") throw new Error("ASSET_CLEANUP_CANDIDATE_STALE");
|
||||
const files = this.database.prepare(`
|
||||
SELECT mf.* FROM asset_cleanup_request_items items
|
||||
JOIN managed_files mf ON mf.file_id = items.managed_file_id
|
||||
WHERE items.request_id = ?
|
||||
`).all(requestId) as ManagedFileRow[];
|
||||
const conflict = files.some((file) => {
|
||||
if (file.status !== "committed" || (file.file_kind !== "sticker_original" && file.file_kind !== "sticker_thumbnail")) return true;
|
||||
const row = this.database.prepare("SELECT COUNT(*) AS count FROM project_asset_refs WHERE managed_file_id = ?").get(file.file_id) as { count: number };
|
||||
return row.count > 0;
|
||||
});
|
||||
if (conflict) {
|
||||
this.database.prepare("UPDATE asset_cleanup_requests SET status = 'denied', confirmed_at = ? WHERE request_id = ?").run(now(), requestId);
|
||||
this.database.prepare("INSERT INTO admin_operation_logs (log_id, operation, outcome, target_ref, created_at) VALUES (?, 'asset_cleanup', 'denied_reference_conflict', ?, ?)")
|
||||
.run(randomUUID(), requestId, now());
|
||||
return false;
|
||||
}
|
||||
for (const file of files) {
|
||||
this.database.prepare("UPDATE managed_files SET status = 'purged', purged_at = ? WHERE file_id = ?").run(now(), file.file_id);
|
||||
this.database.prepare(`
|
||||
INSERT INTO file_cleanup_queue
|
||||
(cleanup_id, managed_file_id, relative_path, byte_size, counts_toward_managed, reason, status, created_at)
|
||||
VALUES (?, ?, ?, ?, 1, 'purge', 'pending', ?)
|
||||
`).run(randomUUID(), file.file_id, file.relative_path, file.byte_size, now());
|
||||
}
|
||||
this.database.prepare("UPDATE asset_cleanup_requests SET status = 'queued', confirmed_at = ? WHERE request_id = ?").run(now(), requestId);
|
||||
this.database.prepare("INSERT INTO admin_operation_logs (log_id, operation, outcome, target_ref, created_at) VALUES (?, 'asset_cleanup', 'queued', ?, ?)")
|
||||
.run(randomUUID(), requestId, now());
|
||||
return true;
|
||||
});
|
||||
if (!transaction()) throw new Error("ASSET_HISTORY_REFERENCE_CONFLICT");
|
||||
}
|
||||
|
||||
async processCleanupQueue() {
|
||||
if (this.inspectAction("explicit_cleanup") !== "allow" || !this.dataRootWritable || !this.diskSpaceAvailable) {
|
||||
return { completed: 0, failed: 0 };
|
||||
}
|
||||
const rows = this.database.prepare("SELECT * FROM file_cleanup_queue WHERE status IN ('pending', 'failed') ORDER BY created_at").all() as CleanupQueueRow[];
|
||||
let completed = 0;
|
||||
let failed = 0;
|
||||
for (const row of rows) {
|
||||
try {
|
||||
const path = resolvePathWithinRoot(this.dataRoot, row.relative_path);
|
||||
rmSync(path, { force: true });
|
||||
const finish = this.database.transaction(() => {
|
||||
if (row.counts_toward_managed === 1 && row.managed_file_id) {
|
||||
this.database.prepare("DELETE FROM project_asset_refs WHERE managed_file_id = ?").run(row.managed_file_id);
|
||||
const requests = this.database.prepare("SELECT request_id FROM asset_cleanup_request_items WHERE managed_file_id = ?").all(row.managed_file_id) as Array<{ request_id: string }>;
|
||||
this.database.prepare("DELETE FROM asset_cleanup_request_items WHERE managed_file_id = ?").run(row.managed_file_id);
|
||||
this.database.prepare("DELETE FROM managed_files WHERE file_id = ?").run(row.managed_file_id);
|
||||
for (const request of requests) {
|
||||
const pendingItems = this.database.prepare("SELECT COUNT(*) AS count FROM asset_cleanup_request_items WHERE request_id = ?").get(request.request_id) as { count: number };
|
||||
if (pendingItems.count === 0) this.database.prepare("UPDATE asset_cleanup_requests SET status = 'completed' WHERE request_id = ? AND status = 'queued'").run(request.request_id);
|
||||
}
|
||||
}
|
||||
this.database.prepare("UPDATE file_cleanup_queue SET status = 'completed', completed_at = ?, last_error = NULL WHERE cleanup_id = ?").run(now(), row.cleanup_id);
|
||||
this.database.prepare("INSERT INTO admin_operation_logs (log_id, operation, outcome, target_ref, created_at) VALUES (?, 'physical_file_cleanup', 'completed', ?, ?)")
|
||||
.run(randomUUID(), row.cleanup_id, now());
|
||||
this.recordPhysicalMeasurement();
|
||||
});
|
||||
finish();
|
||||
completed += 1;
|
||||
} catch (error) {
|
||||
this.database.prepare("UPDATE file_cleanup_queue SET status = 'failed', last_error = ? WHERE cleanup_id = ?")
|
||||
.run(error instanceof Error ? error.message : "cleanup_failed", row.cleanup_id);
|
||||
failed += 1;
|
||||
}
|
||||
}
|
||||
return { completed, failed };
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
export const WARNING_LIMIT_BYTES = 4_294_967_296;
|
||||
export const CRITICAL_LIMIT_BYTES = 4_831_838_208;
|
||||
export const HARD_LIMIT_BYTES = 5_368_709_120;
|
||||
export const CAPACITY_COUNTED_FILE_KINDS = [
|
||||
"reference",
|
||||
"generated",
|
||||
"export",
|
||||
"derived",
|
||||
"sticker_original",
|
||||
"sticker_thumbnail",
|
||||
] as const;
|
||||
export const CAPACITY_EXCLUDED_STORAGE = [
|
||||
"read_only_assets",
|
||||
"application_files",
|
||||
"database",
|
||||
"logs",
|
||||
"audit",
|
||||
"browser_cache",
|
||||
"user_downloads",
|
||||
] as const;
|
||||
|
||||
export type CapacityNoticeLevel = "normal" | "warning" | "critical";
|
||||
export type StorageStatus = "active" | "full" | "unavailable";
|
||||
export type StorageAction =
|
||||
| "project_json_write"
|
||||
| "binary_write"
|
||||
| "ai_call"
|
||||
| "latest_export_write"
|
||||
| "read"
|
||||
| "download"
|
||||
| "client_only_download"
|
||||
| "permanent_delete"
|
||||
| "explicit_cleanup";
|
||||
export type StorageActionDecision = "allow" | "reject_capacity" | "reject_unavailable" | "reject_uncommitted";
|
||||
|
||||
export function classifyCapacity(managedContentBytes: number, activeReservationBytes: number) {
|
||||
if (!Number.isSafeInteger(managedContentBytes) || managedContentBytes < 0) throw new Error("managed_content_bytes_invalid");
|
||||
if (!Number.isSafeInteger(activeReservationBytes) || activeReservationBytes < 0) throw new Error("active_reservation_bytes_invalid");
|
||||
const capacity_notice_level: CapacityNoticeLevel = managedContentBytes < WARNING_LIMIT_BYTES
|
||||
? "normal"
|
||||
: managedContentBytes < CRITICAL_LIMIT_BYTES
|
||||
? "warning"
|
||||
: "critical";
|
||||
const storage_status: StorageStatus = managedContentBytes + activeReservationBytes >= HARD_LIMIT_BYTES
|
||||
? "full"
|
||||
: "active";
|
||||
return { capacity_notice_level, storage_status };
|
||||
}
|
||||
|
||||
export function decideStorageAction(
|
||||
status: StorageStatus,
|
||||
sqliteWritable: boolean,
|
||||
action: StorageAction,
|
||||
): StorageActionDecision {
|
||||
if (status === "unavailable") {
|
||||
if (action === "read" || action === "download" || action === "client_only_download") return "allow";
|
||||
if (action === "permanent_delete" || action === "explicit_cleanup") {
|
||||
return sqliteWritable ? "allow" : "reject_uncommitted";
|
||||
}
|
||||
return "reject_unavailable";
|
||||
}
|
||||
if (status === "full" && (action === "binary_write" || action === "ai_call" || action === "latest_export_write")) {
|
||||
return "reject_capacity";
|
||||
}
|
||||
return "allow";
|
||||
}
|
||||
|
||||
export function storageCapacityErrorDetails(input: {
|
||||
activeReservationBytes: number;
|
||||
managedContentBytes: number;
|
||||
projectedWriteBytes: number;
|
||||
}) {
|
||||
const { capacity_notice_level, storage_status } = classifyCapacity(
|
||||
input.managedContentBytes,
|
||||
input.activeReservationBytes,
|
||||
);
|
||||
return {
|
||||
capacity_status: storage_status === "full" ? "full" as const : capacity_notice_level,
|
||||
remaining_bytes: Math.max(0, HARD_LIMIT_BYTES - input.managedContentBytes - input.activeReservationBytes),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
export interface StorageMaintenanceTarget {
|
||||
processCleanupQueue(): Promise<{ completed: number; failed: number }>;
|
||||
reconcileStartup(): Promise<{ orphaned: number; released_reservations: number }>;
|
||||
}
|
||||
|
||||
export async function runStorageMaintenance(target: StorageMaintenanceTarget) {
|
||||
const reconciliation = await target.reconcileStartup();
|
||||
const cleanup = await target.processCleanupQueue();
|
||||
return { cleanup, reconciliation };
|
||||
}
|
||||
Reference in New Issue
Block a user