350 lines
16 KiB
TypeScript
350 lines
16 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
|
import { existsSync, rmSync, statSync } from "node:fs";
|
|
import { createRequire } from "node:module";
|
|
import { isAbsolute, relative, resolve } from "node:path";
|
|
|
|
import type BetterSqlite3 from "better-sqlite3";
|
|
|
|
const require = createRequire(import.meta.url);
|
|
const Database = require("better-sqlite3") as typeof BetterSqlite3;
|
|
|
|
interface ProjectCleanupRow {
|
|
cleanup_id: string;
|
|
project_id: string;
|
|
}
|
|
|
|
interface ExpiredProjectRow {
|
|
owner_id: string;
|
|
project_id: string;
|
|
}
|
|
|
|
interface FileCleanupRow {
|
|
byte_size: number;
|
|
cleanup_id: string;
|
|
counts_toward_managed: 0 | 1;
|
|
managed_file_id: string | null;
|
|
relative_path: string;
|
|
}
|
|
|
|
const resourceScope = JSON.stringify([
|
|
"project_state", "generation", "generated_image", "reference", "location", "latest_export",
|
|
]);
|
|
const auditRetentionMilliseconds = 180 * 24 * 60 * 60 * 1_000;
|
|
const auditRefPattern = /^[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}$/;
|
|
const forbiddenSummaryKeys = new Set([
|
|
"absolute_path", "api_key", "body", "code_hmac", "content", "credential", "email", "image",
|
|
"image_content", "password", "path", "prompt", "secret", "session_token", "verification_code", "whitelist",
|
|
]);
|
|
const forbiddenSummaryFragments = ["content", "credential", "email", "image", "password", "path", "prompt", "secret", "token"];
|
|
|
|
function isSafeAuditRef(value: unknown) {
|
|
return typeof value === "string" && auditRefPattern.test(value) ? 1 : 0;
|
|
}
|
|
|
|
function isSafeAuditSummaryJson(value: unknown) {
|
|
if (typeof value !== "string" || Buffer.byteLength(value, "utf8") > 2_048) return 0;
|
|
try {
|
|
const valid = (entry: unknown, depth: number): boolean => {
|
|
if (depth > 3) return false;
|
|
if (entry === null || typeof entry === "boolean") return true;
|
|
if (typeof entry === "number") return Number.isSafeInteger(entry);
|
|
if (typeof entry === "string") return /^[A-Za-z0-9_.:@-]{1,160}$/.test(entry) && !entry.includes("@");
|
|
if (Array.isArray(entry)) return entry.length <= 20 && entry.every((item) => valid(item, depth + 1));
|
|
if (!entry || typeof entry !== "object") return false;
|
|
return Object.entries(entry).length <= 32 && Object.entries(entry).every(([key, item]) => (
|
|
auditRefPattern.test(key)
|
|
&& !forbiddenSummaryKeys.has(key.toLowerCase())
|
|
&& !forbiddenSummaryFragments.some((fragment) => key.toLowerCase().includes(fragment))
|
|
&& valid(item, depth + 1)
|
|
));
|
|
};
|
|
return valid(JSON.parse(value), 0) ? 1 : 0;
|
|
} catch {
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
function iso(timestamp: number) {
|
|
return new Date(timestamp).toISOString();
|
|
}
|
|
|
|
export class ProjectPurgeCleanup {
|
|
private readonly clock: () => number;
|
|
private readonly dataRoot: string | undefined;
|
|
private readonly database: BetterSqlite3.Database;
|
|
|
|
constructor(input: { clock?: () => number; dataRoot?: string; databasePath: string }) {
|
|
this.clock = input.clock ?? Date.now;
|
|
this.dataRoot = input.dataRoot ? resolve(input.dataRoot) : undefined;
|
|
const nativeBinding = process.env.DADA_SQLITE_NATIVE_BINDING;
|
|
this.database = new Database(input.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);
|
|
}
|
|
|
|
close() {
|
|
this.database.close();
|
|
}
|
|
|
|
run() {
|
|
const expired = this.sweepExpired();
|
|
const relational = this.processPending();
|
|
const physical = this.processFileCleanup();
|
|
return { expired, physical, relational };
|
|
}
|
|
|
|
sweepExpired() {
|
|
if (!this.tableExists("projects") || !this.tableExists("project_cleanup_queue")) return 0;
|
|
const now = this.clock();
|
|
const rows = this.database.prepare(`
|
|
SELECT project_id, owner_id FROM projects
|
|
WHERE status = 'trashed' AND purge_at <= ?
|
|
ORDER BY purge_at, project_id
|
|
`).all(now) as ExpiredProjectRow[];
|
|
let expired = 0;
|
|
for (const row of rows) {
|
|
const transaction = this.database.transaction(() => {
|
|
const changed = this.database.prepare(`
|
|
UPDATE projects SET status = 'purged', updated_at = ?, state_version = state_version + 1
|
|
WHERE project_id = ? AND status = 'trashed'
|
|
`).run(now, row.project_id);
|
|
if (changed.changes !== 1) return false;
|
|
this.queueManagedFiles(row.project_id, now);
|
|
this.database.prepare(`
|
|
INSERT OR IGNORE INTO project_cleanup_queue (
|
|
cleanup_id, project_id, owner_id, resource_scope_json, status, created_at, completed_at, last_error
|
|
) VALUES (?, ?, ?, ?, 'pending', ?, NULL, NULL)
|
|
`).run(randomUUID(), row.project_id, row.owner_id, resourceScope, now);
|
|
return true;
|
|
});
|
|
if (transaction.immediate()) expired += 1;
|
|
}
|
|
return expired;
|
|
}
|
|
|
|
processPending() {
|
|
if (!this.tableExists("project_cleanup_queue") || !this.tableExists("projects")) return { completed: 0, failed: 0 };
|
|
const rows = this.database.prepare(`
|
|
SELECT cleanup_id, project_id FROM project_cleanup_queue
|
|
WHERE status IN ('pending', 'failed') ORDER BY created_at, project_id
|
|
`).all() as ProjectCleanupRow[];
|
|
let completed = 0;
|
|
let failed = 0;
|
|
for (const row of rows) {
|
|
try {
|
|
const transaction = this.database.transaction(() => {
|
|
this.database.prepare("DELETE FROM projects WHERE project_id = ? AND status = 'purged'").run(row.project_id);
|
|
this.database.prepare(`
|
|
UPDATE project_cleanup_queue
|
|
SET status = 'completed', completed_at = ?, last_error = NULL
|
|
WHERE cleanup_id = ?
|
|
`).run(this.clock(), row.cleanup_id);
|
|
});
|
|
transaction.immediate();
|
|
completed += 1;
|
|
} catch {
|
|
this.database.prepare(`
|
|
UPDATE project_cleanup_queue SET status = 'failed', last_error = 'project_cleanup_failed'
|
|
WHERE cleanup_id = ?
|
|
`).run(row.cleanup_id);
|
|
failed += 1;
|
|
}
|
|
}
|
|
return { completed, failed };
|
|
}
|
|
|
|
processFileCleanup() {
|
|
if (!this.dataRoot || !this.tableExists("file_cleanup_queue") || !this.tableExists("managed_files")) {
|
|
return { completed: 0, failed: 0 };
|
|
}
|
|
const rows = this.database.prepare(`
|
|
SELECT cleanup_id, managed_file_id, relative_path, byte_size, counts_toward_managed
|
|
FROM file_cleanup_queue WHERE status IN ('pending', 'failed') ORDER BY created_at, cleanup_id
|
|
`).all() as FileCleanupRow[];
|
|
let completed = 0;
|
|
let failed = 0;
|
|
for (const row of rows) {
|
|
try {
|
|
const path = this.resolveManagedPath(row.relative_path);
|
|
rmSync(path, { force: true });
|
|
const transaction = this.database.transaction(() => {
|
|
if (row.counts_toward_managed === 1 && row.managed_file_id) {
|
|
if (this.tableExists("project_asset_refs")) {
|
|
this.database.prepare("DELETE FROM project_asset_refs WHERE managed_file_id = ?").run(row.managed_file_id);
|
|
}
|
|
if (this.tableExists("project_resource_files")) {
|
|
this.database.prepare("DELETE FROM project_resource_files WHERE managed_file_id = ?").run(row.managed_file_id);
|
|
}
|
|
if (this.tableExists("asset_cleanup_request_items")) {
|
|
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);
|
|
if (this.tableExists("asset_cleanup_requests")) {
|
|
for (const request of requests) {
|
|
const pending = this.database.prepare("SELECT COUNT(*) AS count FROM asset_cleanup_request_items WHERE request_id = ?")
|
|
.get(request.request_id) as { count: number };
|
|
if (pending.count === 0) {
|
|
this.database.prepare("UPDATE asset_cleanup_requests SET status = 'completed' WHERE request_id = ? AND status = 'queued'")
|
|
.run(request.request_id);
|
|
this.insertAssetCleanupAudit(request.request_id, row.byte_size, this.clock());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if (this.tableExists("sticker_managed_file_history")) {
|
|
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);
|
|
if (this.tableExists("local_backend_storage_state")) this.remeasureManagedCapacity();
|
|
}
|
|
this.database.prepare(`
|
|
UPDATE file_cleanup_queue SET status = 'completed', completed_at = ?, last_error = NULL
|
|
WHERE cleanup_id = ?
|
|
`).run(iso(this.clock()), row.cleanup_id);
|
|
});
|
|
transaction.immediate();
|
|
completed += 1;
|
|
} catch {
|
|
const transaction = this.database.transaction(() => {
|
|
this.database.prepare(`
|
|
UPDATE file_cleanup_queue SET status = 'failed', last_error = 'physical_file_cleanup_failed'
|
|
WHERE cleanup_id = ?
|
|
`).run(row.cleanup_id);
|
|
if (row.managed_file_id && this.tableExists("asset_cleanup_request_items")) {
|
|
const request = this.database.prepare(`
|
|
SELECT request_id FROM asset_cleanup_request_items WHERE managed_file_id = ? LIMIT 1
|
|
`).get(row.managed_file_id) as { request_id: string } | undefined;
|
|
if (request) this.insertAssetCleanupFailureAudit(request.request_id, this.clock());
|
|
}
|
|
});
|
|
transaction.immediate();
|
|
failed += 1;
|
|
}
|
|
}
|
|
return { completed, failed };
|
|
}
|
|
|
|
private queueManagedFiles(projectId: string, now: number) {
|
|
if (!["project_resource_files", "managed_files", "project_asset_refs", "file_cleanup_queue"]
|
|
.every((table) => this.tableExists(table))) return;
|
|
const files = this.database.prepare(`
|
|
SELECT mf.file_id, mf.relative_path, mf.byte_size
|
|
FROM project_resource_files prf
|
|
JOIN managed_files mf ON mf.file_id = prf.managed_file_id
|
|
WHERE prf.project_id = ? AND mf.status = 'committed'
|
|
ORDER BY mf.file_id
|
|
`).all(projectId) as Array<{ byte_size: number; file_id: string; relative_path: string }>;
|
|
for (const file of files) {
|
|
this.database.prepare("DELETE FROM project_asset_refs WHERE reference_id = ?")
|
|
.run(`project:${projectId}:${file.file_id}`);
|
|
const otherProject = this.database.prepare(`
|
|
SELECT 1 FROM project_resource_files prf
|
|
JOIN projects p ON p.project_id = prf.project_id
|
|
WHERE prf.managed_file_id = ? AND prf.project_id <> ? AND p.status <> 'purged'
|
|
LIMIT 1
|
|
`).get(file.file_id, projectId);
|
|
const otherReference = this.database.prepare("SELECT 1 FROM project_asset_refs WHERE managed_file_id = ? LIMIT 1").get(file.file_id);
|
|
if (otherProject || otherReference) continue;
|
|
this.database.prepare("UPDATE managed_files SET status = 'purged', purged_at = ? WHERE file_id = ? AND status = 'committed'")
|
|
.run(iso(now), file.file_id);
|
|
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, 'purge', 'pending', ?, NULL, NULL)
|
|
`).run(randomUUID(), file.file_id, file.relative_path, file.byte_size, iso(now));
|
|
}
|
|
}
|
|
|
|
private tableExists(name: string) {
|
|
return Boolean(this.database.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?").get(name));
|
|
}
|
|
|
|
private resolveManagedPath(relativePath: string) {
|
|
const path = resolve(this.dataRoot!, relativePath);
|
|
const scoped = relative(this.dataRoot!, path);
|
|
if (!scoped || scoped.startsWith("..") || isAbsolute(scoped)) throw new Error("managed_path_outside_data_root");
|
|
return path;
|
|
}
|
|
|
|
private decrementManagedCapacity(bytes: number) {
|
|
const current = this.database.prepare(`
|
|
SELECT managed_content_bytes FROM local_backend_storage_state WHERE singleton = 1
|
|
`).get() as { managed_content_bytes: number } | undefined;
|
|
if (!current) return;
|
|
const managed = Math.max(0, current.managed_content_bytes - bytes);
|
|
const notice = managed < 4_294_967_296 ? "normal" : managed < 4_831_838_208 ? "warning" : "critical";
|
|
const reservations = this.tableExists("storage_reservations")
|
|
? (this.database.prepare("SELECT COALESCE(SUM(projected_bytes), 0) AS bytes FROM storage_reservations WHERE status = 'active'").get() as { bytes: number }).bytes
|
|
: 0;
|
|
const status = managed + reservations >= 5_368_709_120 ? "full" : "active";
|
|
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(managed, notice, status, iso(this.clock()));
|
|
}
|
|
|
|
private insertAssetCleanupAudit(requestId: string, deletedBytes: number, occurredAt: number) {
|
|
if (!this.tableExists("admin_operation_logs")) return;
|
|
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', 'project_purge_worker', 'asset_cleanup_physical_completed', 'asset_cleanup', ?, 'succeeded', NULL, ?, ?, ?)
|
|
`).run(
|
|
randomUUID(), requestId, JSON.stringify({ deleted_bytes: deletedBytes, status: "completed" }), occurredAt,
|
|
occurredAt + auditRetentionMilliseconds,
|
|
);
|
|
}
|
|
|
|
private insertAssetCleanupFailureAudit(requestId: string, occurredAt: number) {
|
|
if (!this.tableExists("admin_operation_logs")) return;
|
|
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', 'project_purge_worker', 'asset_cleanup_physical_failed', 'asset_cleanup', ?, 'failed', NULL, ?, ?, ?)
|
|
`).run(
|
|
randomUUID(), requestId, JSON.stringify({ failed_count: 1, status: "retry_pending" }), occurredAt,
|
|
occurredAt + auditRetentionMilliseconds,
|
|
);
|
|
}
|
|
|
|
private remeasureManagedCapacity() {
|
|
const state = this.database.prepare(`
|
|
SELECT managed_content_bytes FROM local_backend_storage_state WHERE singleton = 1
|
|
`).get() as { managed_content_bytes: number } | undefined;
|
|
if (!state) return;
|
|
const rows = this.database.prepare(`
|
|
SELECT relative_path FROM managed_files WHERE status = 'committed'
|
|
`).all() as Array<{ relative_path: string }>;
|
|
let managed = 0;
|
|
for (const row of rows) {
|
|
try {
|
|
const path = this.resolveManagedPath(row.relative_path);
|
|
if (existsSync(path)) managed += statSync(path).size;
|
|
} catch {
|
|
// A missing or invalid path is excluded from the measured physical total.
|
|
}
|
|
}
|
|
const reservations = this.tableExists("storage_reservations")
|
|
? (this.database.prepare("SELECT COALESCE(SUM(projected_bytes), 0) AS bytes FROM storage_reservations WHERE status = 'active'").get() as { bytes: number }).bytes
|
|
: 0;
|
|
const notice = managed < 4_294_967_296 ? "normal" : managed < 4_831_838_208 ? "warning" : "critical";
|
|
const status = managed + reservations >= 5_368_709_120 ? "full" : "active";
|
|
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(managed, notice, status, iso(this.clock()));
|
|
}
|
|
}
|