feat: add durable CMS-free image video subsystem and UI components
This commit is contained in:
Wenxin Duan
@@ -0,0 +1,147 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { lstat, mkdir, open, readFile, readdir, rename, stat } from "node:fs/promises";
|
||||
import { join, resolve } from "node:path";
|
||||
import type { AttachmentStore } from "../attachments/attachment-store.js";
|
||||
import { HarnessError } from "../errors/harness-error.js";
|
||||
import { ImageVideoDraftSchema, recordSchema, taskIdSchema, validationSchema, type ImageVideoDraft, type ImageVideoPublicTask, type ImageVideoRecord, type ImageVideoRemote, type ImageVideoValidation } from "./contracts.js";
|
||||
|
||||
const locks = new Map<string, Promise<unknown>>();
|
||||
export const sha256 = (bytes: Buffer | string) => createHash("sha256").update(bytes).digest("hex");
|
||||
const fail = (code: string, message: string): never => { throw new HarnessError(code, message); };
|
||||
export type ImageVideoOptions = { root: string; attachments: Pick<AttachmentStore, "get" | "openNormalized">; remote: ImageVideoRemote; collect: (url: string, directory: string, duration: number) => Promise<ImageVideoValidation> };
|
||||
|
||||
/** One process owns this private store. Per-task locks are shared by service instances. */
|
||||
export class ImageVideoService {
|
||||
readonly root: string;
|
||||
constructor(private readonly options: ImageVideoOptions) { this.root = resolve(options.root); }
|
||||
|
||||
async create(sessionId: string, input: ImageVideoDraft): Promise<ImageVideoPublicTask> {
|
||||
const draft = ImageVideoDraftSchema.parse(input);
|
||||
const attachment = this.options.attachments.get(sessionId, draft.attachmentId);
|
||||
if (!attachment || attachment.sessionId !== sessionId) return fail("I2V_TASK_NOT_FOUND", "Reference image not found in this Session");
|
||||
const bytes = await this.options.attachments.openNormalized(sessionId, draft.attachmentId);
|
||||
const now = new Date().toISOString(); const id = `i2v_${randomUUID()}`;
|
||||
const record = recordSchema.parse({ version: 1, id, ownerSessionId: sessionId, revision: 0, draft, referenceHash: sha256(bytes), requestHash: sha256(JSON.stringify(draft)), referenceType: attachment.mediaType, state: "draft", createdAt: now, updatedAt: now });
|
||||
await mkdir(join(this.root, id), { recursive: true, mode: 0o700 });
|
||||
await this.atomic(join(this.root, id, "reference"), bytes);
|
||||
await this.atomic(join(this.root, id, "record.json"), JSON.stringify(record));
|
||||
return this.public(record);
|
||||
}
|
||||
|
||||
async list(): Promise<ImageVideoPublicTask[]> {
|
||||
const entries = await readdir(this.root).catch(() => []);
|
||||
const tasks = await Promise.all(entries.filter(id => taskIdSchema.safeParse(id).success).map(id => this.view(id).catch(() => undefined)));
|
||||
return tasks.filter((task): task is ImageVideoPublicTask => task !== undefined).sort((a, b) => b.createdAt.localeCompare(a.createdAt));
|
||||
}
|
||||
async view(id: string): Promise<ImageVideoPublicTask> { return this.lock(id, async () => this.public(await this.read(id))); }
|
||||
async owned(sessionId: string, id: string): Promise<ImageVideoPublicTask> { return this.lock(id, async () => { const r = await this.read(id); this.owner(r, sessionId); return this.public(r); }); }
|
||||
|
||||
async upload(sessionId: string, id: string): Promise<ImageVideoPublicTask> {
|
||||
return this.lock(id, async () => {
|
||||
const r = await this.read(id); this.owner(r, sessionId);
|
||||
if (r.state === "unknown_upload") return fail("I2V_UNKNOWN_UPLOAD", "Upload outcome unknown; inspect the saved receipt before any new import");
|
||||
if (r.state !== "draft") return this.public(r);
|
||||
// Read-only discovery and identity validation happen before any remote mutation intent.
|
||||
r.identity = await this.options.remote.preflight();
|
||||
const bytes = await this.reference(r);
|
||||
r.state = "initializing_upload"; await this.save(r);
|
||||
try {
|
||||
r.upload = await this.options.remote.initialize(r, bytes); r.state = "uploading"; await this.save(r);
|
||||
await this.options.remote.put(r, bytes); r.state = "finalizing"; await this.save(r);
|
||||
await this.options.remote.finalize(r); r.state = "importing"; await this.save(r);
|
||||
} catch { r.state = "unknown_upload"; r.error = "参考图上传或入库结果未知；禁止重复导入。请核实本任务收据。"; await this.save(r); }
|
||||
return this.public(r);
|
||||
});
|
||||
}
|
||||
|
||||
async submit(sessionId: string, id: string): Promise<ImageVideoPublicTask> {
|
||||
return this.lock(id, async () => {
|
||||
const r = await this.read(id); this.owner(r, sessionId);
|
||||
if (r.jobId) return this.public(r);
|
||||
if (r.state === "unknown_submission") return fail("I2V_UNKNOWN_SUBMISSION", "Submission outcome unknown; no safe retry or new billing request is allowed");
|
||||
if (r.state !== "awaiting_submission" || !r.assetId) return fail("I2V_NOT_READY", "Reference import must be ready before generation approval");
|
||||
await this.options.remote.preflight(); await this.reference(r);
|
||||
r.state = "submitting"; await this.save(r);
|
||||
try {
|
||||
// Deliberately independent of chat cancellation: retain any returned receipt.
|
||||
const receipt = await this.options.remote.submit(r);
|
||||
if (!receipt.jobId) throw new Error("Missing job ID");
|
||||
await this.atomic(join(this.root, id, "submission-receipt.json"), JSON.stringify({ version: 1, id, revision: r.revision, requestHash: r.requestHash, jobId: receipt.jobId }));
|
||||
r.jobId = receipt.jobId; r.state = "queued"; delete r.lastPollAt; await this.save(r);
|
||||
} catch { r.state = "unknown_submission"; r.error = "生成提交结果未知；禁止重投。请人工核实，避免重复计费。"; await this.save(r); }
|
||||
return this.public(r);
|
||||
});
|
||||
}
|
||||
|
||||
async refresh(id: string): Promise<ImageVideoPublicTask> {
|
||||
return this.lock(id, async () => {
|
||||
const r = await this.read(id);
|
||||
if (r.state === "ready" || r.state === "failed" || Date.now() - (r.lastPollAt ?? 0) < 30_000) return this.public(r);
|
||||
if (!r.jobId && !(r.upload && ["importing", "unknown_upload"].includes(r.state))) return this.public(r);
|
||||
r.lastPollAt = Date.now(); await this.save(r);
|
||||
try {
|
||||
if (!r.jobId) {
|
||||
const result = await this.options.remote.importStatus(r);
|
||||
if (result.state === "ready" && result.assetId) { r.assetId = result.assetId; r.state = "awaiting_submission"; delete r.error; }
|
||||
else if (result.state === "failed") { r.state = "failed"; r.error = "参考图入库失败。"; }
|
||||
} else {
|
||||
const result = await this.options.remote.status(r);
|
||||
if (result.state === "succeeded") {
|
||||
if (!result.artifactUrl) throw new Error("No unambiguous MP4 artifact");
|
||||
r.artifactUrl = result.artifactUrl; r.state = "validating"; await this.save(r);
|
||||
r.validation = validationSchema.parse(await this.options.collect(result.artifactUrl, join(this.root, id), r.draft.durationSeconds));
|
||||
if (Math.abs(r.validation.videoSeconds - r.draft.durationSeconds) > 0.002 || Math.abs(r.validation.audioSeconds - r.draft.durationSeconds) > 0.05) throw new Error("Duration mismatch");
|
||||
r.state = "ready"; delete r.error;
|
||||
} else { r.state = result.state; if (result.state === "failed") r.error = "H3 生成失败；未自动重投。"; }
|
||||
}
|
||||
} catch {
|
||||
if (r.state === "validating") r.state = "failed";
|
||||
r.error = r.state === "failed" ? "成片技术验证失败：请核实 MP4、精确时长、尺寸、帧率和音轨；未自动重投。" : "状态响应暂不可用或契约不明确；保留原任务，下次只读查询。";
|
||||
}
|
||||
await this.save(r); return this.public(r);
|
||||
});
|
||||
}
|
||||
|
||||
async media(id: string, kind: "reference" | "video") {
|
||||
return this.lock(id, async () => {
|
||||
const r = await this.read(id);
|
||||
if (kind === "video" && (r.state !== "ready" || !r.validation)) return fail("I2V_MEDIA_NOT_FOUND", "Validated video not ready");
|
||||
const path = join(this.root, id, kind === "reference" ? "reference" : "video.mp4");
|
||||
await this.regular(path); const bytes = await readFile(path);
|
||||
if (sha256(bytes) !== (kind === "reference" ? r.referenceHash : r.validation?.sha256)) return fail("I2V_MEDIA_INVALID", "Registered media hash mismatch");
|
||||
return { path, size: (await stat(path)).size, contentType: kind === "reference" ? r.referenceType : "video/mp4" };
|
||||
});
|
||||
}
|
||||
private owner(r: ImageVideoRecord, sessionId: string) { if (r.ownerSessionId !== sessionId) fail("I2V_TASK_NOT_FOUND", "Task not found in this Session"); }
|
||||
private public(r: ImageVideoRecord): ImageVideoPublicTask {
|
||||
return { id: r.id, title: r.draft.title, prompt: r.draft.prompt, durationSeconds: r.draft.durationSeconds, state: r.state, createdAt: r.createdAt, updatedAt: r.updatedAt, referenceUrl: `/api/image-video/tasks/${r.id}/reference`, ...(r.jobId ? { jobId: r.jobId } : {}), ...(r.error ? { error: r.error } : {}), ...(r.validation ? { validation: r.validation } : {}), ...(r.state === "ready" ? { videoUrl: `/api/image-video/tasks/${r.id}/video` } : {}) };
|
||||
}
|
||||
private async reference(r: ImageVideoRecord) { await this.regular(join(this.root, r.id, "reference")); const bytes = await readFile(join(this.root, r.id, "reference")); if (sha256(bytes) !== r.referenceHash) fail("I2V_MEDIA_INVALID", "Reference changed after draft"); return bytes; }
|
||||
private async regular(path: string) { const info = await lstat(path); if (!info.isFile() || info.isSymbolicLink()) fail("I2V_RECORD_INVALID", "Unsafe registered file"); }
|
||||
private async read(id: string): Promise<ImageVideoRecord> {
|
||||
if (!taskIdSchema.safeParse(id).success) return fail("I2V_TASK_NOT_FOUND", "Task not found");
|
||||
const path = join(this.root, id, "record.json");
|
||||
const dir = await lstat(join(this.root, id)).catch(() => undefined);
|
||||
if (!dir || !dir.isDirectory() || dir.isSymbolicLink()) return fail("I2V_TASK_NOT_FOUND", "Task not found");
|
||||
try {
|
||||
await this.regular(path); const r = recordSchema.parse(JSON.parse(await readFile(path, "utf8")));
|
||||
if (r.id !== id || r.requestHash !== sha256(JSON.stringify(r.draft))) throw new Error("Binding mismatch");
|
||||
if (r.state === "submitting") { r.state = "unknown_submission"; await this.save(r); }
|
||||
if (["initializing_upload", "uploading", "finalizing"].includes(r.state)) { r.state = "unknown_upload"; await this.save(r); }
|
||||
return r;
|
||||
} catch { return fail("I2V_RECORD_INVALID", "Task record is invalid; manual recovery required"); }
|
||||
}
|
||||
private async save(r: ImageVideoRecord) {
|
||||
const path = join(this.root, r.id, "record.json");
|
||||
const previous = recordSchema.parse(JSON.parse(await readFile(path, "utf8")));
|
||||
if (previous.revision !== r.revision) return fail("I2V_STALE_REVISION", "Stale task result rejected");
|
||||
r.revision++; r.updatedAt = new Date().toISOString();
|
||||
await this.atomic(path, JSON.stringify(recordSchema.parse(r)));
|
||||
}
|
||||
private async atomic(path: string, content: Buffer | string) { const temp = `${path}.${randomUUID()}.tmp`; const file = await open(temp, "wx", 0o600); try { await file.writeFile(content); await file.sync(); } finally { await file.close(); } await rename(temp, path); }
|
||||
private async lock<T>(id: string, operation: () => Promise<T>): Promise<T> {
|
||||
const key = `${this.root}/${id}`; const previous = locks.get(key) ?? Promise.resolve();
|
||||
const work = previous.catch(() => undefined).then(operation); locks.set(key, work);
|
||||
try { return await work; } finally { if (locks.get(key) === work) locks.delete(key); }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user