1199 lines
56 KiB
TypeScript
1199 lines
56 KiB
TypeScript
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 {
|
|
auditRetentionMilliseconds,
|
|
ensureAdminOperationAuditSchema,
|
|
isSafeAuditRef,
|
|
isSafeAuditSummaryJson,
|
|
serializeAuditSummary,
|
|
} from "./audit-policy.js";
|
|
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 interface AssetCleanupCandidateView {
|
|
byte_size: number;
|
|
file_id: string;
|
|
file_kind: "original" | "thumbnail";
|
|
hash_prefix: string;
|
|
reference_count: 0;
|
|
resource_version: string;
|
|
stable_id: string;
|
|
}
|
|
|
|
export interface AssetCleanupCandidatesView {
|
|
candidate_snapshot_version: string;
|
|
expires_at: string;
|
|
items: AssetCleanupCandidateView[];
|
|
}
|
|
|
|
export interface AssetCleanupIntentView {
|
|
confirmation_token: string;
|
|
expires_at: string;
|
|
file_count: number;
|
|
request_id: string;
|
|
status: "pending_confirmation" | "denied" | "queued" | "completed";
|
|
total_bytes: number;
|
|
}
|
|
|
|
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 auditExpiry(occurredAt: number) {
|
|
return occurredAt + auditRetentionMilliseconds;
|
|
}
|
|
|
|
function digest(value: string) {
|
|
return createHash("sha256").update(value, "utf8").digest("hex");
|
|
}
|
|
|
|
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";
|
|
if (prefix.length >= 12 && prefix.subarray(0, 4).toString("ascii") === "RIFF" && prefix.subarray(8, 12).toString("ascii") === "WEBP") {
|
|
return "image/webp";
|
|
}
|
|
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" | "image/webp" | "application/octet-stream";
|
|
expectedSha256?: string;
|
|
failurePoint?: CommitFailurePoint;
|
|
fileKind: ManagedFileKind;
|
|
fileName: string;
|
|
operationId: string;
|
|
ownerRef: string;
|
|
projectedWriteBytes: number;
|
|
}
|
|
|
|
export interface StagedManagedFile {
|
|
bytes: number;
|
|
destinationPath: string;
|
|
fileId: string;
|
|
fileKind: ManagedFileKind;
|
|
mimeType: "image/png" | "image/jpeg" | "image/webp";
|
|
operationId: string;
|
|
ownerRef: string;
|
|
relativePath: string;
|
|
sha256: string;
|
|
stagingDirectory: string;
|
|
stagingPath: string;
|
|
}
|
|
|
|
export class ManagedStorage {
|
|
readonly dataRoot: string;
|
|
readonly databasePath: string;
|
|
private readonly database: BetterSqlite3.Database;
|
|
private dataRootWritable = true;
|
|
private diskSpaceAvailable = true;
|
|
private logWritable = 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 });
|
|
const nativeBinding = process.env.DADA_SQLITE_NATIVE_BINDING;
|
|
this.database = new Database(this.databasePath, nativeBinding ? { nativeBinding } : undefined);
|
|
this.database.pragma("journal_mode = WAL");
|
|
this.database.pragma("foreign_keys = ON");
|
|
this.database.pragma("busy_timeout = 5000");
|
|
this.database.function("dada_audit_ref_is_safe", { deterministic: true }, isSafeAuditRef);
|
|
this.database.function("dada_audit_summary_is_safe", { deterministic: true }, isSafeAuditSummaryJson);
|
|
this.database.function("dada_allow_privacy_purge", { deterministic: false }, () => 0);
|
|
this.database.function("dada_privacy_purge_subject", { deterministic: false }, () => "");
|
|
this.database.function("dada_allow_retention_purge", { deterministic: false }, () => 0);
|
|
this.database.function("dada_retention_purge_now", { deterministic: false }, () => 0);
|
|
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')),
|
|
owner_ref TEXT,
|
|
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 asset_cleanup_candidate_snapshots (
|
|
snapshot_version TEXT PRIMARY KEY,
|
|
items_json TEXT NOT NULL,
|
|
created_at INTEGER NOT NULL,
|
|
expires_at INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS sticker_managed_file_history (
|
|
managed_file_id TEXT NOT NULL,
|
|
stable_id TEXT NOT NULL,
|
|
resource_version TEXT NOT NULL,
|
|
file_kind TEXT NOT NULL CHECK (file_kind IN ('original', 'thumbnail')),
|
|
created_at INTEGER NOT NULL,
|
|
PRIMARY KEY (managed_file_id, file_kind),
|
|
FOREIGN KEY (managed_file_id) REFERENCES managed_files(file_id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS project_sticker_asset_refs (
|
|
reference_id TEXT PRIMARY KEY,
|
|
project_id TEXT NOT NULL,
|
|
stable_id TEXT NOT NULL,
|
|
resource_version TEXT NOT NULL,
|
|
created_at INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS admin_operation_logs (
|
|
log_id TEXT PRIMARY KEY,
|
|
actor_type TEXT NOT NULL CHECK (actor_type IN ('system', 'super_admin')),
|
|
actor_ref TEXT NOT NULL,
|
|
operation_type TEXT NOT NULL,
|
|
target_type TEXT NOT NULL,
|
|
target_ref TEXT NOT NULL,
|
|
result TEXT NOT NULL CHECK (result IN ('succeeded', 'failed')),
|
|
before_summary TEXT,
|
|
after_summary TEXT,
|
|
occurred_at TEXT NOT NULL,
|
|
expires_at TEXT NOT NULL
|
|
);
|
|
CREATE TRIGGER IF NOT EXISTS admin_operation_logs_no_update
|
|
BEFORE UPDATE ON admin_operation_logs BEGIN SELECT RAISE(ABORT, 'admin_operation_logs_immutable'); END;
|
|
CREATE TRIGGER IF NOT EXISTS admin_operation_logs_no_delete
|
|
BEFORE DELETE ON admin_operation_logs BEGIN SELECT RAISE(ABORT, 'admin_operation_logs_immutable'); END;
|
|
`);
|
|
const managedFileColumns = this.database.prepare("PRAGMA table_info(managed_files)").all() as Array<{ name: string }>;
|
|
if (!managedFileColumns.some((column) => column.name === "owner_ref")) {
|
|
this.database.exec("ALTER TABLE managed_files ADD COLUMN owner_ref TEXT");
|
|
}
|
|
if (!managedFileColumns.some((column) => column.name === "cleanup_status")) {
|
|
this.database.exec("ALTER TABLE managed_files ADD COLUMN cleanup_status TEXT");
|
|
}
|
|
const cleanupRequestColumns = this.database.prepare("PRAGMA table_info(asset_cleanup_requests)").all() as Array<{ name: string }>;
|
|
const cleanupRequestAdditions: Array<[string, string]> = [
|
|
["created_by", "TEXT"],
|
|
["confirmed_by", "TEXT"],
|
|
["snapshot_version", "TEXT"],
|
|
["expires_at", "INTEGER"],
|
|
["confirmation_token_digest", "TEXT"],
|
|
["idempotency_key_digest", "TEXT"],
|
|
["request_hash", "TEXT"],
|
|
["file_count", "INTEGER"],
|
|
["total_bytes", "INTEGER"],
|
|
["denied_reason", "TEXT"],
|
|
];
|
|
for (const [column, type] of cleanupRequestAdditions) {
|
|
if (!cleanupRequestColumns.some((item) => item.name === column)) {
|
|
this.database.exec(`ALTER TABLE asset_cleanup_requests ADD COLUMN ${column} ${type}`);
|
|
}
|
|
}
|
|
const cleanupItemColumns = this.database.prepare("PRAGMA table_info(asset_cleanup_request_items)").all() as Array<{ name: string }>;
|
|
const cleanupItemAdditions: Array<[string, string]> = [
|
|
["stable_id", "TEXT"],
|
|
["resource_version", "TEXT"],
|
|
["file_kind", "TEXT"],
|
|
["byte_size", "INTEGER"],
|
|
["sha256_prefix", "TEXT"],
|
|
];
|
|
for (const [column, type] of cleanupItemAdditions) {
|
|
if (!cleanupItemColumns.some((item) => item.name === column)) {
|
|
this.database.exec(`ALTER TABLE asset_cleanup_request_items ADD COLUMN ${column} ${type}`);
|
|
}
|
|
}
|
|
this.database.exec(`
|
|
CREATE UNIQUE INDEX IF NOT EXISTS asset_cleanup_requests_actor_idempotency
|
|
ON asset_cleanup_requests (created_by, idempotency_key_digest)
|
|
WHERE created_by IS NOT NULL AND idempotency_key_digest IS NOT NULL;
|
|
CREATE INDEX IF NOT EXISTS sticker_managed_file_history_lookup
|
|
ON sticker_managed_file_history (stable_id, resource_version, file_kind);
|
|
CREATE INDEX IF NOT EXISTS asset_cleanup_candidate_snapshots_expiry
|
|
ON asset_cleanup_candidate_snapshots (expires_at);
|
|
`);
|
|
ensureAdminOperationAuditSchema(this.database, Date.now());
|
|
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 || !this.logWritable) {
|
|
return { ...withReservations, storage_status: "unavailable" };
|
|
}
|
|
return withReservations;
|
|
}
|
|
|
|
private readAssetCleanupCandidates(): AssetCleanupCandidateView[] {
|
|
const releaseReferenceClause = this.tableExists("sticker_release_items") ? `
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM sticker_release_items release_items
|
|
WHERE release_items.original_file_id = mf.file_id OR release_items.thumbnail_file_id = mf.file_id
|
|
)` : "";
|
|
return this.database.prepare(`
|
|
SELECT
|
|
mf.file_id,
|
|
mf.byte_size,
|
|
history.stable_id,
|
|
history.resource_version,
|
|
history.file_kind,
|
|
substr(mf.sha256, 1, 12) AS hash_prefix,
|
|
0 AS reference_count
|
|
FROM sticker_managed_file_history history
|
|
JOIN managed_files mf ON mf.file_id = history.managed_file_id
|
|
WHERE mf.status = 'committed'
|
|
AND mf.cleanup_status IS NULL
|
|
AND mf.file_kind IN ('sticker_original', 'sticker_thumbnail')
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM project_asset_refs refs WHERE refs.managed_file_id = mf.file_id
|
|
)
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM project_sticker_asset_refs project_refs
|
|
WHERE project_refs.stable_id = history.stable_id
|
|
AND project_refs.resource_version = history.resource_version
|
|
)
|
|
${releaseReferenceClause}
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM asset_cleanup_request_items request_items
|
|
JOIN asset_cleanup_requests requests ON requests.request_id = request_items.request_id
|
|
WHERE request_items.managed_file_id = mf.file_id
|
|
AND requests.status IN ('pending_confirmation', 'queued')
|
|
)
|
|
ORDER BY history.stable_id, history.resource_version, history.file_kind, mf.file_id
|
|
`).all() as AssetCleanupCandidateView[];
|
|
}
|
|
|
|
private assertActiveAdmin(actorId: string) {
|
|
const admin = this.database.prepare(`
|
|
SELECT 1 AS allowed FROM users u
|
|
JOIN admin_access access ON access.user_id = u.user_id
|
|
WHERE u.user_id = ? AND u.role = 'super_admin' AND u.status = 'active' AND access.allowed = 1
|
|
`).get(actorId);
|
|
if (!admin) throw new Error("ASSET_CLEANUP_CANDIDATE_STALE");
|
|
}
|
|
|
|
private assetReferenceCount(fileId: string, requestId: string) {
|
|
const projectOrRelease = (this.database.prepare(`
|
|
SELECT COUNT(*) AS count FROM project_asset_refs WHERE managed_file_id = ?
|
|
`).get(fileId) as { count: number }).count;
|
|
const releaseItems = this.tableExists("sticker_release_items")
|
|
? (this.database.prepare(`
|
|
SELECT COUNT(*) AS count FROM sticker_release_items
|
|
WHERE original_file_id = ? OR thumbnail_file_id = ?
|
|
`).get(fileId, fileId) as { count: number }).count
|
|
: 0;
|
|
const projectStickerRefs = (this.database.prepare(`
|
|
SELECT COUNT(*) AS count
|
|
FROM sticker_managed_file_history history
|
|
JOIN project_sticker_asset_refs refs
|
|
ON refs.stable_id = history.stable_id AND refs.resource_version = history.resource_version
|
|
WHERE history.managed_file_id = ?
|
|
`).get(fileId) as { count: number }).count;
|
|
const otherCleanup = (this.database.prepare(`
|
|
SELECT COUNT(*) AS count FROM asset_cleanup_request_items items
|
|
JOIN asset_cleanup_requests requests ON requests.request_id = items.request_id
|
|
WHERE items.managed_file_id = ? AND items.request_id <> ?
|
|
AND requests.status IN ('pending_confirmation', 'queued')
|
|
`).get(fileId, requestId) as { count: number }).count;
|
|
return projectOrRelease + releaseItems + projectStickerRefs + otherCleanup;
|
|
}
|
|
|
|
private cleanupConfirmationToken(requestId: string, actorId: string, keyDigest: string) {
|
|
return digest(`Dada/P0A/asset-cleanup-confirm/v1:${requestId}:${actorId}:${keyDigest}`);
|
|
}
|
|
|
|
private tableExists(name: string) {
|
|
return Boolean(this.database.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?").get(name));
|
|
}
|
|
|
|
private insertCleanupAudit(input: {
|
|
actorRef: string;
|
|
afterSummary: Record<string, unknown>;
|
|
operationType: string;
|
|
requestId: string;
|
|
result: "failed" | "succeeded";
|
|
}, occurredAt: number) {
|
|
this.database.prepare(`
|
|
INSERT INTO admin_operation_logs (
|
|
log_id, actor_type, actor_ref, operation_type, target_type, target_ref,
|
|
result, before_summary, after_summary, occurred_at, expires_at
|
|
) VALUES (?, 'super_admin', ?, ?, 'asset_cleanup_request', ?, ?, NULL, ?, ?, ?)
|
|
`).run(
|
|
randomUUID(), input.actorRef, input.operationType, input.requestId, input.result,
|
|
serializeAuditSummary(input.afterSummary), occurredAt, auditExpiry(occurredAt),
|
|
);
|
|
}
|
|
|
|
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.logWritable || 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();
|
|
}
|
|
|
|
setLogAvailability(writable: boolean) {
|
|
this.logWritable = writable;
|
|
if (!writable) 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, owner_ref, relative_path, byte_size, mime_type, sha256, status, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, 'committed', ?)
|
|
`).run(fileId, input.fileKind, input.ownerRef, 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 stageManagedImage(input: {
|
|
content: Readable;
|
|
expectedMimeType: "image/png" | "image/jpeg" | "image/webp";
|
|
expectedSha256?: string;
|
|
fileKind: ManagedFileKind;
|
|
fileName: string;
|
|
maximumBytes: number;
|
|
operationId: string;
|
|
ownerRef: string;
|
|
projectedWriteBytes: number;
|
|
}): Promise<StagedManagedFile> {
|
|
const fileId = randomUUID();
|
|
const destination = this.destination({
|
|
content: input.content,
|
|
expectedMimeType: input.expectedMimeType,
|
|
fileKind: input.fileKind,
|
|
fileName: input.fileName,
|
|
operationId: input.operationId,
|
|
ownerRef: input.ownerRef,
|
|
projectedWriteBytes: input.projectedWriteBytes,
|
|
}, fileId);
|
|
if (!Number.isSafeInteger(input.maximumBytes) || input.maximumBytes <= 0) throw new Error("maximum_bytes_invalid");
|
|
this.reserve(input.operationId, input.projectedWriteBytes);
|
|
const stagingDirectory = resolvePathWithinRoot(this.dataRoot, `staging/${input.operationId}`);
|
|
const stagingPath = resolvePathWithinRoot(this.dataRoot, `staging/${input.operationId}/payload.tmp`);
|
|
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;
|
|
if (byteSize > input.maximumBytes) return callback(new Error("content_size_invalid"));
|
|
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");
|
|
if (sniffMime(prefix) !== input.expectedMimeType) throw new Error("content_mime_invalid");
|
|
const sha256 = hash.digest("hex");
|
|
if (input.expectedSha256 && sha256.toLowerCase() !== input.expectedSha256.toLowerCase()) throw new Error("content_hash_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();
|
|
return {
|
|
bytes: byteSize,
|
|
destinationPath: destination.absolutePath,
|
|
fileId,
|
|
fileKind: input.fileKind,
|
|
mimeType: input.expectedMimeType,
|
|
operationId: input.operationId,
|
|
ownerRef: input.ownerRef,
|
|
relativePath: destination.relativePath,
|
|
sha256,
|
|
stagingDirectory,
|
|
stagingPath,
|
|
};
|
|
} catch (error) {
|
|
rmSync(stagingDirectory, { force: true, recursive: true });
|
|
this.releaseReservation(input.operationId);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async stagePrivateImage(input: {
|
|
content: Readable;
|
|
expectedMimeType: "image/png" | "image/jpeg" | "image/webp";
|
|
fileName: string;
|
|
maximumBytes: number;
|
|
operationId: string;
|
|
ownerRef: string;
|
|
projectedWriteBytes: number;
|
|
}): Promise<StagedManagedFile> {
|
|
return this.stageManagedImage({ ...input, fileKind: "reference" });
|
|
}
|
|
|
|
moveStagedFile(file: StagedManagedFile) {
|
|
mkdirSync(dirname(file.destinationPath), { recursive: true });
|
|
renameSync(file.stagingPath, file.destinationPath);
|
|
rmSync(file.stagingDirectory, { force: true, recursive: true });
|
|
}
|
|
|
|
abandonStagedFile(file: StagedManagedFile) {
|
|
if (existsSync(file.destinationPath)) this.queueCompensation(file.relativePath, statSync(file.destinationPath).size);
|
|
else rmSync(file.stagingDirectory, { force: true, recursive: true });
|
|
this.releaseReservation(file.operationId);
|
|
}
|
|
|
|
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;
|
|
}
|
|
|
|
retireManagedFile(fileId: string, reason: "compensation" | "purge" = "compensation") {
|
|
const transaction = this.database.transaction(() => {
|
|
const file = this.database.prepare(`
|
|
SELECT file_id, relative_path, byte_size FROM managed_files
|
|
WHERE file_id = ? AND status = 'committed'
|
|
`).get(fileId) as { byte_size: number; file_id: string; relative_path: string } | undefined;
|
|
if (!file) return;
|
|
const retiredAt = now();
|
|
this.database.prepare("DELETE FROM project_asset_refs WHERE managed_file_id = ?").run(fileId);
|
|
this.database.prepare("UPDATE managed_files SET status = 'purged', purged_at = ? WHERE file_id = ?").run(retiredAt, fileId);
|
|
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, completed_at, last_error
|
|
) VALUES (?, ?, ?, ?, 1, ?, 'pending', ?, NULL, NULL)
|
|
`).run(randomUUID(), fileId, file.relative_path, file.byte_size, reason, retiredAt);
|
|
});
|
|
transaction.immediate();
|
|
}
|
|
|
|
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);
|
|
}
|
|
|
|
listAssetCleanupCandidates(): AssetCleanupCandidatesView {
|
|
const createdAt = Date.now();
|
|
const expiresAt = createdAt + 5 * 60 * 1_000;
|
|
const items = this.readAssetCleanupCandidates();
|
|
const snapshotVersion = digest(JSON.stringify({
|
|
created_at: createdAt,
|
|
nonce: randomUUID(),
|
|
items: items.map((item) => ({ byte_size: item.byte_size, file_id: item.file_id, hash_prefix: item.hash_prefix })),
|
|
}));
|
|
this.database.prepare("DELETE FROM asset_cleanup_candidate_snapshots WHERE expires_at <= ?").run(createdAt);
|
|
this.database.prepare(`
|
|
INSERT INTO asset_cleanup_candidate_snapshots (
|
|
snapshot_version, items_json, created_at, expires_at
|
|
) VALUES (?, ?, ?, ?)
|
|
`).run(snapshotVersion, JSON.stringify(items), createdAt, expiresAt);
|
|
return {
|
|
candidate_snapshot_version: snapshotVersion,
|
|
expires_at: new Date(expiresAt).toISOString(),
|
|
items,
|
|
};
|
|
}
|
|
|
|
createAssetCleanupIntent(input: {
|
|
actorId: string;
|
|
fileIds: string[];
|
|
idempotencyKey: string;
|
|
snapshotVersion: string;
|
|
}): AssetCleanupIntentView {
|
|
if (this.inspectAction("explicit_cleanup") !== "allow") throw new Error("cleanup_uncommitted");
|
|
const fileIds = [...new Set(input.fileIds)].sort();
|
|
if (!uuidPattern.test(input.actorId) || fileIds.length === 0 || fileIds.length !== input.fileIds.length
|
|
|| fileIds.length > 100 || fileIds.some((fileId) => !uuidPattern.test(fileId))
|
|
|| !/^[A-Za-z0-9_-]{32,200}$/.test(input.idempotencyKey)
|
|
|| !/^[0-9a-f]{64}$/.test(input.snapshotVersion)) {
|
|
throw new Error("cleanup_candidates_invalid");
|
|
}
|
|
const keyDigest = digest(input.idempotencyKey);
|
|
const requestHash = digest(JSON.stringify({ file_ids: fileIds, snapshot_version: input.snapshotVersion }));
|
|
const existing = this.database.prepare(`
|
|
SELECT request_id, request_hash, expires_at, file_count, total_bytes, status
|
|
FROM asset_cleanup_requests
|
|
WHERE created_by = ? AND idempotency_key_digest = ?
|
|
`).get(input.actorId, keyDigest) as {
|
|
expires_at: number; file_count: number; request_hash: string; request_id: string; status: AssetCleanupIntentView["status"]; total_bytes: number;
|
|
} | undefined;
|
|
if (existing) {
|
|
if (existing.request_hash !== requestHash) throw new Error("IDEMPOTENCY_KEY_CONFLICT");
|
|
return {
|
|
confirmation_token: this.cleanupConfirmationToken(existing.request_id, input.actorId, keyDigest),
|
|
expires_at: new Date(existing.expires_at).toISOString(),
|
|
file_count: existing.file_count,
|
|
request_id: existing.request_id,
|
|
status: existing.status,
|
|
total_bytes: existing.total_bytes,
|
|
};
|
|
}
|
|
|
|
const requestId = randomUUID();
|
|
const confirmationToken = this.cleanupConfirmationToken(requestId, input.actorId, keyDigest);
|
|
const createdAt = Date.now();
|
|
let view!: AssetCleanupIntentView;
|
|
const transaction = this.database.transaction(() => {
|
|
this.assertActiveAdmin(input.actorId);
|
|
const snapshot = this.database.prepare(`
|
|
SELECT items_json, expires_at FROM asset_cleanup_candidate_snapshots
|
|
WHERE snapshot_version = ?
|
|
`).get(input.snapshotVersion) as { expires_at: number; items_json: string } | undefined;
|
|
if (!snapshot || snapshot.expires_at <= createdAt) throw new Error("ASSET_CLEANUP_CANDIDATE_STALE");
|
|
const snapshotItems = JSON.parse(snapshot.items_json) as AssetCleanupCandidateView[];
|
|
const byId = new Map(snapshotItems.map((item) => [item.file_id, item]));
|
|
const selected = fileIds.map((fileId) => byId.get(fileId));
|
|
if (selected.some((item) => !item)) throw new Error("ASSET_CLEANUP_CANDIDATE_STALE");
|
|
const current = new Map(this.readAssetCleanupCandidates().map((item) => [item.file_id, item]));
|
|
if (fileIds.some((fileId) => !current.has(fileId))) throw new Error("ASSET_CLEANUP_CANDIDATE_STALE");
|
|
const safeItems = selected as AssetCleanupCandidateView[];
|
|
const totalBytes = safeItems.reduce((sum, item) => sum + item.byte_size, 0);
|
|
this.database.prepare(`
|
|
INSERT INTO asset_cleanup_requests (
|
|
request_id, status, created_at, confirmed_at, created_by, confirmed_by,
|
|
snapshot_version, expires_at, confirmation_token_digest,
|
|
idempotency_key_digest, request_hash, file_count, total_bytes, denied_reason
|
|
) VALUES (?, 'pending_confirmation', ?, NULL, ?, NULL, ?, ?, ?, ?, ?, ?, ?, NULL)
|
|
`).run(
|
|
requestId, new Date(createdAt).toISOString(), input.actorId, input.snapshotVersion,
|
|
snapshot.expires_at, digest(confirmationToken), keyDigest, requestHash, safeItems.length, totalBytes,
|
|
);
|
|
const insert = this.database.prepare(`
|
|
INSERT INTO asset_cleanup_request_items (
|
|
request_id, managed_file_id, stable_id, resource_version, file_kind, byte_size, sha256_prefix
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
`);
|
|
for (const item of safeItems) {
|
|
insert.run(requestId, item.file_id, item.stable_id, item.resource_version, item.file_kind, item.byte_size, item.hash_prefix);
|
|
}
|
|
this.insertCleanupAudit({
|
|
actorRef: input.actorId,
|
|
afterSummary: { file_count: safeItems.length, snapshot_version: input.snapshotVersion, total_bytes: totalBytes },
|
|
operationType: "asset_cleanup_requested",
|
|
requestId,
|
|
result: "succeeded",
|
|
}, createdAt);
|
|
view = {
|
|
confirmation_token: confirmationToken,
|
|
expires_at: new Date(snapshot.expires_at).toISOString(),
|
|
file_count: safeItems.length,
|
|
request_id: requestId,
|
|
status: "pending_confirmation",
|
|
total_bytes: totalBytes,
|
|
};
|
|
});
|
|
transaction.immediate();
|
|
return view;
|
|
}
|
|
|
|
confirmAssetCleanupIntent(input: { actorId: string; confirmationToken: string; requestId: string }) {
|
|
if (this.inspectAction("explicit_cleanup") !== "allow") throw new Error("cleanup_uncommitted");
|
|
if (!uuidPattern.test(input.actorId) || !uuidPattern.test(input.requestId) || !/^[0-9a-f]{64}$/.test(input.confirmationToken)) {
|
|
throw new Error("ASSET_CLEANUP_CANDIDATE_STALE");
|
|
}
|
|
const confirmedAt = Date.now();
|
|
const outcome = this.database.transaction(() => {
|
|
this.assertActiveAdmin(input.actorId);
|
|
const request = this.database.prepare(`
|
|
SELECT status, created_by, expires_at, confirmation_token_digest, file_count, total_bytes
|
|
FROM asset_cleanup_requests WHERE request_id = ?
|
|
`).get(input.requestId) as {
|
|
confirmation_token_digest: string | null; created_by: string | null; expires_at: number | null;
|
|
file_count: number | null; status: string; total_bytes: number | null;
|
|
} | undefined;
|
|
if (!request || request.status !== "pending_confirmation" || request.created_by !== input.actorId
|
|
|| !request.expires_at || request.expires_at <= confirmedAt
|
|
|| request.confirmation_token_digest !== digest(input.confirmationToken)) {
|
|
throw new Error("ASSET_CLEANUP_CANDIDATE_STALE");
|
|
}
|
|
const files = this.database.prepare(`
|
|
SELECT mf.file_id, mf.file_kind, mf.relative_path, mf.byte_size, mf.status
|
|
FROM asset_cleanup_request_items items
|
|
JOIN managed_files mf ON mf.file_id = items.managed_file_id
|
|
WHERE items.request_id = ? ORDER BY mf.file_id
|
|
`).all(input.requestId) as ManagedFileRow[];
|
|
if (files.length !== request.file_count) throw new Error("ASSET_CLEANUP_CANDIDATE_STALE");
|
|
const conflicted = files.some((file) => file.status !== "committed"
|
|
|| !new Set(["sticker_original", "sticker_thumbnail"]).has(file.file_kind)
|
|
|| this.assetReferenceCount(file.file_id, input.requestId) > 0);
|
|
if (conflicted) {
|
|
this.database.prepare(`
|
|
UPDATE asset_cleanup_requests
|
|
SET status = 'denied', confirmed_at = ?, confirmed_by = ?, denied_reason = 'reference_conflict'
|
|
WHERE request_id = ?
|
|
`).run(new Date(confirmedAt).toISOString(), input.actorId, input.requestId);
|
|
this.insertCleanupAudit({
|
|
actorRef: input.actorId,
|
|
afterSummary: { file_count: files.length, reason: "reference_conflict", status: "denied" },
|
|
operationType: "asset_cleanup_reference_denied",
|
|
requestId: input.requestId,
|
|
result: "failed",
|
|
}, confirmedAt);
|
|
return { conflict: true as const };
|
|
}
|
|
|
|
this.insertCleanupAudit({
|
|
actorRef: input.actorId,
|
|
afterSummary: { file_count: files.length, status: "validated" },
|
|
operationType: "asset_cleanup_validated",
|
|
requestId: input.requestId,
|
|
result: "succeeded",
|
|
}, confirmedAt);
|
|
for (const file of files) {
|
|
this.database.prepare(`
|
|
UPDATE managed_files SET status = 'purged', purged_at = ?, cleanup_status = 'pending_delete'
|
|
WHERE file_id = ? AND status = 'committed'
|
|
`).run(new Date(confirmedAt).toISOString(), 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, completed_at, last_error
|
|
) VALUES (?, ?, ?, ?, 1, 'purge', 'pending', ?, NULL, NULL)
|
|
`).run(randomUUID(), file.file_id, file.relative_path, file.byte_size, new Date(confirmedAt).toISOString());
|
|
}
|
|
this.database.prepare(`
|
|
UPDATE asset_cleanup_requests
|
|
SET status = 'queued', confirmed_at = ?, confirmed_by = ?
|
|
WHERE request_id = ?
|
|
`).run(new Date(confirmedAt).toISOString(), input.actorId, input.requestId);
|
|
this.insertCleanupAudit({
|
|
actorRef: input.actorId,
|
|
afterSummary: { file_count: files.length, status: "queued", total_bytes: request.total_bytes ?? 0 },
|
|
operationType: "asset_cleanup_scheduled",
|
|
requestId: input.requestId,
|
|
result: "succeeded",
|
|
}, confirmedAt + 1);
|
|
return { conflict: false as const, file_count: files.length, request_id: input.requestId, status: "queued" as const };
|
|
}).immediate();
|
|
if (outcome.conflict) throw new Error("ASSET_HISTORY_REFERENCE_CONFLICT");
|
|
return outcome;
|
|
}
|
|
|
|
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) {
|
|
const confirmedAt = now();
|
|
const occurredAt = Date.now();
|
|
this.database.prepare("UPDATE asset_cleanup_requests SET status = 'denied', confirmed_at = ? WHERE request_id = ?").run(confirmedAt, requestId);
|
|
this.database.prepare(`
|
|
INSERT INTO admin_operation_logs (
|
|
log_id, actor_type, actor_ref, operation_type, target_type, target_ref,
|
|
result, before_summary, after_summary, occurred_at, expires_at
|
|
) VALUES (?, 'system', 'managed_storage', 'asset_cleanup', 'cleanup_request', ?, 'failed', NULL, ?, ?, ?)
|
|
`).run(randomUUID(), requestId, serializeAuditSummary({ reason: "reference_conflict" }), occurredAt, auditExpiry(occurredAt));
|
|
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());
|
|
}
|
|
const confirmedAt = now();
|
|
const occurredAt = Date.now();
|
|
this.database.prepare("UPDATE asset_cleanup_requests SET status = 'queued', confirmed_at = ? WHERE request_id = ?").run(confirmedAt, requestId);
|
|
this.database.prepare(`
|
|
INSERT INTO admin_operation_logs (
|
|
log_id, actor_type, actor_ref, operation_type, target_type, target_ref,
|
|
result, before_summary, after_summary, occurred_at, expires_at
|
|
) VALUES (?, 'system', 'managed_storage', 'asset_cleanup', 'cleanup_request', ?, 'succeeded', NULL, ?, ?, ?)
|
|
`).run(randomUUID(), requestId, serializeAuditSummary({ status: "queued" }), occurredAt, auditExpiry(occurredAt));
|
|
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 sticker_managed_file_history 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);
|
|
const occurredAt = Date.now();
|
|
this.database.prepare(`
|
|
INSERT INTO admin_operation_logs (
|
|
log_id, actor_type, actor_ref, operation_type, target_type, target_ref,
|
|
result, before_summary, after_summary, occurred_at, expires_at
|
|
) VALUES (?, 'system', 'managed_storage', 'physical_file_cleanup', 'cleanup_queue_item', ?, 'succeeded', NULL, ?, ?, ?)
|
|
`).run(randomUUID(), row.cleanup_id, serializeAuditSummary({ status: "completed" }), occurredAt, auditExpiry(occurredAt));
|
|
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 };
|
|
}
|
|
}
|