feat: complete TASK-WP2-07 latest exports
This commit is contained in:
suyx
@@ -0,0 +1,263 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { createRequire } from "node:module";
|
||||
import type { Readable } from "node:stream";
|
||||
|
||||
import type BetterSqlite3 from "better-sqlite3";
|
||||
|
||||
import { ManagedStorage } from "./managed-storage.js";
|
||||
import { stableJson } from "./projects.js";
|
||||
import { LatestExportError } from "./latest-export-errors.js";
|
||||
export { LatestExportError } from "./latest-export-errors.js";
|
||||
|
||||
const require = createRequire(import.meta.url);
|
||||
const Database = require("better-sqlite3") as typeof BetterSqlite3;
|
||||
|
||||
export type ExportFormat = "jpg" | "png";
|
||||
|
||||
interface ExportRow {
|
||||
byte_size: number;
|
||||
created_at: number;
|
||||
export_id: string;
|
||||
format: ExportFormat;
|
||||
managed_file_id: string;
|
||||
pixel_height: number;
|
||||
pixel_width: number;
|
||||
project_id: string;
|
||||
sha256: string;
|
||||
state_version: number;
|
||||
}
|
||||
|
||||
interface ManagedFileRow {
|
||||
byte_size: number;
|
||||
file_id: string;
|
||||
mime_type: string;
|
||||
relative_path: string;
|
||||
sha256: string;
|
||||
}
|
||||
|
||||
type LatestExportView = ReturnType<LatestExportService["view"]>;
|
||||
|
||||
function iso(timestamp: number) {
|
||||
return new Date(timestamp).toISOString();
|
||||
}
|
||||
|
||||
function validSha256(value: string) {
|
||||
return /^[0-9a-f]{64}$/.test(value);
|
||||
}
|
||||
|
||||
export class LatestExportService {
|
||||
readonly database: BetterSqlite3.Database;
|
||||
private readonly clock: () => number;
|
||||
private readonly storage: ManagedStorage;
|
||||
|
||||
constructor(input: { clock?: () => number; databasePath: string; storage: ManagedStorage }) {
|
||||
this.clock = input.clock ?? Date.now;
|
||||
this.storage = input.storage;
|
||||
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.migrate();
|
||||
}
|
||||
|
||||
close() {
|
||||
this.database.close();
|
||||
}
|
||||
|
||||
async saveLatest(input: {
|
||||
byteSize: number;
|
||||
content: Readable;
|
||||
exportId: string;
|
||||
format: ExportFormat;
|
||||
ownerId: string;
|
||||
pixelHeight: number;
|
||||
pixelWidth: number;
|
||||
projectId: string;
|
||||
sha256: string;
|
||||
stateVersion: number;
|
||||
}) {
|
||||
this.validateInput(input);
|
||||
const requestHash = createHash("sha256").update(stableJson({
|
||||
byte_size: input.byteSize, export_id: input.exportId, format: input.format,
|
||||
owner_id: input.ownerId,
|
||||
pixel_height: input.pixelHeight, pixel_width: input.pixelWidth, project_id: input.projectId,
|
||||
sha256: input.sha256, state_version: input.stateVersion,
|
||||
})).digest("hex");
|
||||
const replay = this.database.prepare("SELECT owner_id, request_hash, response_json FROM latest_export_receipts WHERE export_id = ?")
|
||||
.get(input.exportId) as { owner_id: string; request_hash: string; response_json: string | null } | undefined;
|
||||
if (replay) {
|
||||
if (replay.owner_id !== input.ownerId || replay.request_hash !== requestHash) throw new LatestExportError("conflict");
|
||||
const replayHash = createHash("sha256");
|
||||
let replayBytes = 0;
|
||||
for await (const chunk of input.content) {
|
||||
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
||||
replayBytes += bytes.byteLength;
|
||||
replayHash.update(bytes);
|
||||
}
|
||||
if (replayBytes !== input.byteSize || replayHash.digest("hex") !== input.sha256) throw new LatestExportError("conflict");
|
||||
return replay.response_json ? JSON.parse(replay.response_json) as LatestExportView : this.readByExportId(input.ownerId, input.exportId);
|
||||
}
|
||||
this.assertWritableProject(input);
|
||||
const stored = await this.storage.commitStream({
|
||||
content: input.content,
|
||||
expectedMimeType: input.format === "png" ? "image/png" : "image/jpeg",
|
||||
expectedSha256: input.sha256,
|
||||
fileKind: "export",
|
||||
fileName: `latest.${input.format === "jpg" ? "jpg" : "png"}`,
|
||||
operationId: input.exportId,
|
||||
ownerRef: input.ownerId,
|
||||
projectedWriteBytes: input.byteSize,
|
||||
});
|
||||
try {
|
||||
if (stored.bytes !== input.byteSize) throw new LatestExportError("invalid");
|
||||
const createdAt = this.clock();
|
||||
const response = this.view({
|
||||
byte_size: stored.bytes,
|
||||
created_at: createdAt,
|
||||
export_id: input.exportId,
|
||||
format: input.format,
|
||||
managed_file_id: stored.file_id,
|
||||
pixel_height: input.pixelHeight,
|
||||
pixel_width: input.pixelWidth,
|
||||
project_id: input.projectId,
|
||||
sha256: stored.sha256,
|
||||
state_version: input.stateVersion,
|
||||
});
|
||||
const transaction = this.database.transaction(() => {
|
||||
this.assertWritableProject(input);
|
||||
const previous = this.database.prepare("SELECT managed_file_id FROM latest_exports WHERE project_id = ? AND format = ?")
|
||||
.get(input.projectId, input.format) as { managed_file_id: string } | undefined;
|
||||
this.database.prepare(`
|
||||
INSERT INTO project_resource_files (project_id, managed_file_id, resource_kind, created_at)
|
||||
VALUES (?, ?, 'export', ?)
|
||||
`).run(input.projectId, stored.file_id, createdAt);
|
||||
this.database.prepare(`
|
||||
INSERT INTO project_asset_refs (reference_id, managed_file_id, reference_type, created_at)
|
||||
VALUES (?, ?, 'project', ?)
|
||||
`).run(`project:${input.projectId}:${stored.file_id}`, stored.file_id, iso(createdAt));
|
||||
this.database.prepare(`
|
||||
INSERT INTO latest_exports (
|
||||
project_id, format, export_id, managed_file_id, state_version, sha256,
|
||||
byte_size, pixel_width, pixel_height, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(project_id, format) DO UPDATE SET
|
||||
export_id = excluded.export_id, managed_file_id = excluded.managed_file_id,
|
||||
state_version = excluded.state_version, sha256 = excluded.sha256,
|
||||
byte_size = excluded.byte_size, pixel_width = excluded.pixel_width,
|
||||
pixel_height = excluded.pixel_height, created_at = excluded.created_at
|
||||
`).run(
|
||||
input.projectId, input.format, input.exportId, stored.file_id, input.stateVersion,
|
||||
stored.sha256, stored.bytes, input.pixelWidth, input.pixelHeight, createdAt,
|
||||
);
|
||||
this.database.prepare("INSERT INTO latest_export_receipts (export_id, owner_id, request_hash, response_json, created_at) VALUES (?, ?, ?, ?, ?)")
|
||||
.run(input.exportId, input.ownerId, requestHash, stableJson(response), createdAt);
|
||||
if (previous && previous.managed_file_id !== stored.file_id) this.retireReplacedFile(previous.managed_file_id, createdAt);
|
||||
});
|
||||
transaction.immediate();
|
||||
return response;
|
||||
} catch (error) {
|
||||
this.storage.retireManagedFile(stored.file_id, "compensation");
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
getLatest(ownerId: string, projectId: string, format: ExportFormat) {
|
||||
const row = this.database.prepare(`
|
||||
SELECT le.*, mf.mime_type, mf.relative_path
|
||||
FROM latest_exports le
|
||||
JOIN projects p ON p.project_id = le.project_id
|
||||
JOIN managed_files mf ON mf.file_id = le.managed_file_id AND mf.status = 'committed'
|
||||
WHERE p.owner_id = ? AND p.project_id = ? AND p.status <> 'purged' AND le.format = ?
|
||||
`).get(ownerId, projectId, format) as (ExportRow & ManagedFileRow) | undefined;
|
||||
if (!row) throw new LatestExportError("not_found");
|
||||
const path = this.storage.resolveManagedFile(row.managed_file_id);
|
||||
if (!path) throw new LatestExportError("not_found");
|
||||
return { ...this.view(row), mimeType: row.mime_type, path };
|
||||
}
|
||||
|
||||
getOriginal(ownerId: string, projectId: string, imageId: string) {
|
||||
const row = this.database.prepare(`
|
||||
SELECT mf.file_id, mf.relative_path, mf.byte_size, mf.mime_type, mf.sha256
|
||||
FROM project_images pi
|
||||
JOIN projects p ON p.project_id = pi.project_id
|
||||
JOIN managed_files mf ON mf.file_id = pi.image_id AND mf.status = 'committed'
|
||||
WHERE p.owner_id = ? AND p.project_id = ? AND p.status <> 'purged' AND pi.image_id = ?
|
||||
`).get(ownerId, projectId, imageId) as ManagedFileRow | undefined;
|
||||
if (!row) throw new LatestExportError("not_found");
|
||||
const path = this.storage.resolveManagedFile(row.file_id);
|
||||
if (!path) throw new LatestExportError("not_found");
|
||||
return { ...row, path };
|
||||
}
|
||||
|
||||
private assertWritableProject(input: Pick<Parameters<LatestExportService["saveLatest"]>[0], "ownerId" | "pixelHeight" | "pixelWidth" | "projectId" | "stateVersion">) {
|
||||
const project = this.database.prepare(`
|
||||
SELECT state_version, pixel_width, pixel_height FROM projects
|
||||
WHERE owner_id = ? AND project_id = ? AND status = 'active'
|
||||
`).get(input.ownerId, input.projectId) as { pixel_height: number; pixel_width: number; state_version: number } | undefined;
|
||||
if (!project) throw new LatestExportError("not_found");
|
||||
if (project.state_version !== input.stateVersion) throw new LatestExportError("conflict");
|
||||
if (project.pixel_width !== input.pixelWidth || project.pixel_height !== input.pixelHeight) throw new LatestExportError("invalid");
|
||||
}
|
||||
|
||||
private readByExportId(ownerId: string, exportId: string) {
|
||||
const row = this.database.prepare(`
|
||||
SELECT le.* FROM latest_exports le JOIN projects p ON p.project_id = le.project_id
|
||||
WHERE p.owner_id = ? AND le.export_id = ?
|
||||
`).get(ownerId, exportId) as ExportRow | undefined;
|
||||
if (!row) throw new LatestExportError("not_found");
|
||||
return this.view(row);
|
||||
}
|
||||
|
||||
private retireReplacedFile(fileId: string, timestamp: number) {
|
||||
const file = this.database.prepare("SELECT relative_path, byte_size FROM managed_files WHERE file_id = ? AND status = 'committed'")
|
||||
.get(fileId) as { byte_size: number; relative_path: string } | undefined;
|
||||
if (!file) return;
|
||||
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(iso(timestamp), 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, 'purge', 'pending', ?, NULL, NULL)
|
||||
`).run(randomUUID(), fileId, file.relative_path, file.byte_size, iso(timestamp));
|
||||
}
|
||||
|
||||
private validateInput(input: Parameters<LatestExportService["saveLatest"]>[0]) {
|
||||
if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(input.exportId) || !["jpg", "png"].includes(input.format)
|
||||
|| !validSha256(input.sha256) || ![input.byteSize, input.pixelHeight, input.pixelWidth, input.stateVersion].every((value) => Number.isSafeInteger(value) && value > 0)) {
|
||||
throw new LatestExportError("invalid");
|
||||
}
|
||||
}
|
||||
|
||||
private view(row: ExportRow) {
|
||||
return {
|
||||
byteSize: row.byte_size,
|
||||
createdAt: iso(row.created_at),
|
||||
downloadUrl: `/api/v1/projects/${row.project_id}/latest-exports/${row.format}`,
|
||||
exportId: row.export_id,
|
||||
format: row.format,
|
||||
pixelHeight: row.pixel_height,
|
||||
pixelWidth: row.pixel_width,
|
||||
projectId: row.project_id,
|
||||
sha256: row.sha256,
|
||||
stateVersion: row.state_version,
|
||||
};
|
||||
}
|
||||
|
||||
private migrate() {
|
||||
this.database.exec(`
|
||||
CREATE TABLE IF NOT EXISTS latest_export_receipts (
|
||||
export_id TEXT PRIMARY KEY,
|
||||
owner_id TEXT NOT NULL,
|
||||
request_hash TEXT NOT NULL CHECK (length(request_hash) = 64),
|
||||
response_json TEXT CHECK (response_json IS NULL OR json_valid(response_json)),
|
||||
created_at INTEGER NOT NULL
|
||||
);
|
||||
`);
|
||||
const columns = this.database.prepare("PRAGMA table_info(latest_export_receipts)").all() as Array<{ name: string }>;
|
||||
if (!columns.some((column) => column.name === "response_json")) {
|
||||
this.database.exec("ALTER TABLE latest_export_receipts ADD COLUMN response_json TEXT");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user