269 lines
14 KiB
TypeScript
269 lines
14 KiB
TypeScript
import { mkdir, readFile, stat, writeFile } from 'node:fs/promises';
|
|
import path from 'node:path';
|
|
import { setTimeout as delay } from 'node:timers/promises';
|
|
import { atomicWriteJson, readJsonFile } from '../coding-projects/atomic-json';
|
|
import type { PiProjectWriteLeaseCoordinator } from '../coding-runtime/pi/write-lease';
|
|
import { audioRecord, GameAudioClientError, parseAudioGeneration,
|
|
type AudioCommand, type AudioGeneration, type GameAudioClient, type AudioBilling } from './game-audio-client';
|
|
import { getWorksSquareAccountBinding } from './works-square-session';
|
|
import { readMediaDataUrl } from '../coding-projects/game-asset-browser';
|
|
|
|
export interface AudioDeliveryContext {
|
|
accountKey: string;
|
|
conversationId: string;
|
|
runId: string;
|
|
localProjectId: string;
|
|
durableProjectId: string;
|
|
projectPath: string;
|
|
logicalOperationId: string;
|
|
}
|
|
export interface AudioDeliveredFile { index: number; path: string; bytes: number; mimeType: string }
|
|
export interface AudioDeliveryResult {
|
|
executionId: string | null;
|
|
logicalOperationId: string;
|
|
providerStatus: AudioGeneration['provider_status'];
|
|
deliveryStatus: 'waiting' | 'saving' | 'saved' | 'delivery_failed' | 'needs_reconciliation';
|
|
outputCount: number;
|
|
files: AudioDeliveredFile[];
|
|
billing: AudioBilling;
|
|
phase: 'submitted' | 'generating' | 'saving' | 'saved' | 'paused';
|
|
progressPercent: number | null;
|
|
errorCode?: string;
|
|
}
|
|
interface Receipt {
|
|
context: AudioDeliveryContext;
|
|
generation: AudioGeneration | null;
|
|
deliveryStatus: AudioDeliveryResult['deliveryStatus'];
|
|
files: AudioDeliveredFile[];
|
|
errorCode?: string;
|
|
submissionRejected?: boolean;
|
|
}
|
|
function sameIdentity(left: AudioDeliveryContext, right: AudioDeliveryContext) {
|
|
return left.accountKey === right.accountKey && left.logicalOperationId === right.logicalOperationId;
|
|
}
|
|
function validReceipt(value: unknown): value is Receipt {
|
|
if (!audioRecord(value) || !audioRecord(value.context)
|
|
|| !['accountKey', 'conversationId', 'runId', 'localProjectId', 'durableProjectId', 'projectPath', 'logicalOperationId']
|
|
.every((key) => typeof value.context[key] === 'string' && value.context[key].length > 0)
|
|
|| !path.isAbsolute(String(value.context.projectPath))
|
|
|| !['waiting', 'saving', 'saved', 'delivery_failed', 'needs_reconciliation'].includes(String(value.deliveryStatus))
|
|
|| !Array.isArray(value.files) || value.files.length > 10) return false;
|
|
try {
|
|
if (value.generation !== null) parseAudioGeneration(value.generation);
|
|
} catch { return false; }
|
|
return value.files.every((file) => audioRecord(file) && Number.isInteger(file.index)
|
|
&& typeof file.bytes === 'number' && file.bytes > 0 && typeof file.path === 'string'
|
|
&& /^assets\/generated\/game-audio\/[0-9a-f-]{36}\/(?:music|sound)-\d+\.(?:mp3|wav|ogg|flac|m4a)$/u.test(file.path));
|
|
}
|
|
/** One atomic userData file; no project file serves as the job database. */
|
|
export class GameAudioReceiptStore {
|
|
private writes: Promise<void> = Promise.resolve();
|
|
constructor(private readonly filePath: string) {}
|
|
async list(): Promise<Receipt[]> {
|
|
let value: unknown;
|
|
try { value = await readJsonFile(this.filePath); } catch (error) {
|
|
if (audioRecord(error) && error.code === 'ENOENT') return [];
|
|
throw error;
|
|
}
|
|
if (!audioRecord(value) || value.schemaVersion !== 1 || !Array.isArray(value.receipts)
|
|
|| !value.receipts.every(validReceipt)) throw new Error('Audio receipt file is invalid');
|
|
return value.receipts;
|
|
}
|
|
async put(receipt: Receipt): Promise<void> {
|
|
const operation = this.writes.then(async () => {
|
|
const rows = (await this.list()).filter((row) => !sameIdentity(row.context, receipt.context));
|
|
rows.push(receipt);
|
|
await atomicWriteJson(this.filePath, { schemaVersion: 1, receipts: rows });
|
|
});
|
|
this.writes = operation.catch(() => undefined);
|
|
await operation;
|
|
}
|
|
}
|
|
interface Options {
|
|
client: Pick<GameAudioClient, 'generate' | 'get' | 'byOperation' | 'download'>;
|
|
receipts: GameAudioReceiptStore;
|
|
leases: PiProjectWriteLeaseCoordinator;
|
|
getAccountKey?: () => string | null;
|
|
sleep?: (ms: number, signal?: AbortSignal) => Promise<void>;
|
|
recordTouchedPaths?: (conversationId: string, runId: string, paths: readonly string[]) => Promise<void>;
|
|
acquireBackgroundLease?: (lease: { id: string; kind: 'coding-run' }) => () => void;
|
|
}
|
|
type Progress = (value: AudioDeliveryResult) => void;
|
|
export class GameAudioDeliveryCoordinator {
|
|
private readonly flights = new Map<string, Promise<AudioDeliveryResult>>();
|
|
private readonly shutdown = new AbortController();
|
|
private recoveryTimer: ReturnType<typeof setTimeout> | undefined;
|
|
constructor(private readonly options: Options) {}
|
|
dispose(): void {
|
|
this.shutdown.abort();
|
|
clearTimeout(this.recoveryTimer);
|
|
}
|
|
private scheduleRecovery(): void {
|
|
if (this.shutdown.signal.aborted || this.recoveryTimer) return;
|
|
this.recoveryTimer = setTimeout(() => {
|
|
this.recoveryTimer = undefined;
|
|
void this.resumePending().catch(() => this.scheduleRecovery());
|
|
}, 30_000);
|
|
this.recoveryTimer.unref?.();
|
|
}
|
|
async readSavedOutput(executionId: string, index: number): Promise<{ path: string; dataUrl: string }> {
|
|
const account = this.account();
|
|
const receipt = (await this.options.receipts.list()).find((row) =>
|
|
row.context.accountKey === account && row.generation?.execution_id === executionId);
|
|
const file = receipt?.files.find((item) => item.index === index);
|
|
if (!receipt || !file) throw new GameAudioClientError('game_audio_output_not_found', 404, 'Saved audio is unavailable');
|
|
this.assertAccount(receipt.context);
|
|
const dataUrl = await readMediaDataUrl(receipt.context.projectPath, file.path, 32 * 1024 * 1024);
|
|
this.assertAccount(receipt.context);
|
|
if (!dataUrl?.startsWith('data:audio/')) throw new GameAudioClientError('game_audio_content_invalid', 422, 'Saved audio cannot be read');
|
|
return { path: file.path, dataUrl };
|
|
}
|
|
async generateAndMaterialize(context: AudioDeliveryContext, command: AudioCommand, progress?: Progress) {
|
|
return await this.singleFlight(context, async () => {
|
|
const old = (await this.options.receipts.list()).find((row) => sameIdentity(row.context, context));
|
|
if (old) return await this.advance(old, progress);
|
|
this.assertAccount(context);
|
|
const receipt: Receipt = { context: { ...context }, generation: null, deliveryStatus: 'waiting', files: [] };
|
|
// The durable identity must exist BEFORE the first possibly charged request.
|
|
await this.options.receipts.put(receipt);
|
|
try {
|
|
receipt.generation = await this.options.client.generate(command, context.accountKey);
|
|
this.checkGeneration(receipt);
|
|
await this.options.receipts.put(receipt);
|
|
} catch (error) {
|
|
if (error instanceof GameAudioClientError && (error.status === 401 || error.status === 422
|
|
|| error.code === 'plugin_pricing_changed')) {
|
|
receipt.deliveryStatus = 'needs_reconciliation';
|
|
receipt.submissionRejected = true;
|
|
receipt.errorCode = error.code;
|
|
await this.options.receipts.put(receipt);
|
|
return this.result(receipt, 'paused');
|
|
}
|
|
// Response loss is not permission to submit again.
|
|
}
|
|
return await this.advance(receipt, progress);
|
|
});
|
|
}
|
|
async resumePending(): Promise<void> {
|
|
const account = this.account();
|
|
if (!account || this.shutdown.signal.aborted) return;
|
|
const rows = await this.options.receipts.list();
|
|
await Promise.allSettled(rows.filter((row) => row.context.accountKey === account
|
|
&& !row.submissionRejected
|
|
&& row.deliveryStatus !== 'saved'
|
|
&& !['failed', 'cancelled'].includes(row.generation?.provider_status ?? ''))
|
|
.map((row) => this.singleFlight(row.context, () => this.advance(row))));
|
|
}
|
|
private async singleFlight(context: AudioDeliveryContext, run: () => Promise<AudioDeliveryResult>) {
|
|
const key = JSON.stringify([context.accountKey, context.logicalOperationId]);
|
|
const old = this.flights.get(key);
|
|
if (old) return await old;
|
|
const release = this.options.acquireBackgroundLease?.({ id: 'audio:' + key, kind: 'coding-run' });
|
|
const flight = run().finally(() => { this.flights.delete(key); release?.(); });
|
|
this.flights.set(key, flight);
|
|
return await flight;
|
|
}
|
|
private account(): string | null {
|
|
return this.options.getAccountKey ? this.options.getAccountKey() : getWorksSquareAccountBinding()?.accountKey ?? null;
|
|
}
|
|
private assertAccount(context: AudioDeliveryContext): void {
|
|
if (this.shutdown.signal.aborted || this.account() !== context.accountKey) {
|
|
throw new GameAudioClientError('plugin_account_changed', 409, 'Audio recovery waits for the original account');
|
|
}
|
|
}
|
|
private checkGeneration(receipt: Receipt): void {
|
|
const job = receipt.generation;
|
|
if (job && (job.project_id !== receipt.context.durableProjectId
|
|
|| job.logical_operation_id !== receipt.context.logicalOperationId)) {
|
|
throw new GameAudioClientError('plugin_backend_invalid', 502, 'Audio execution identity changed');
|
|
}
|
|
}
|
|
private async sleep(ms: number): Promise<void> {
|
|
if (this.options.sleep) await this.options.sleep(ms, this.shutdown.signal);
|
|
else await delay(ms, undefined, { signal: this.shutdown.signal });
|
|
}
|
|
private async advance(receipt: Receipt, progress?: Progress): Promise<AudioDeliveryResult> {
|
|
if (receipt.submissionRejected) return this.result(receipt, 'paused');
|
|
if (receipt.deliveryStatus === 'saved') return this.result(receipt, 'saved');
|
|
const report = (phase: AudioDeliveryResult['phase']) => {
|
|
try { progress?.(this.result(receipt, phase)); } catch { /* Observational only. */ }
|
|
};
|
|
try {
|
|
this.assertAccount(receipt.context);
|
|
let errors = 0;
|
|
while (true) {
|
|
this.assertAccount(receipt.context);
|
|
const job = receipt.generation;
|
|
if (job && !['accepted', 'running'].includes(job.provider_status)) break;
|
|
report(job ? 'generating' : 'submitted');
|
|
try {
|
|
receipt.generation = job
|
|
? await this.options.client.get(job.execution_id, receipt.context.accountKey)
|
|
: await this.options.client.byOperation(receipt.context.logicalOperationId, receipt.context.accountKey);
|
|
this.checkGeneration(receipt);
|
|
await this.options.receipts.put(receipt);
|
|
errors = 0;
|
|
} catch (error) {
|
|
errors++;
|
|
if (errors >= 3 || (error instanceof GameAudioClientError && error.code === 'plugin_account_changed')) throw error;
|
|
await this.sleep(1000 * errors);
|
|
continue;
|
|
}
|
|
if (['accepted', 'running'].includes(receipt.generation.provider_status)) {
|
|
await this.sleep(receipt.generation.poll_interval_seconds * 1000);
|
|
}
|
|
}
|
|
const job = receipt.generation;
|
|
if (!job?.output_ready) return this.result(receipt, 'paused');
|
|
this.assertAccount(receipt.context);
|
|
const project = await stat(receipt.context.projectPath);
|
|
if (!project.isDirectory()) throw new Error('Original project is unavailable');
|
|
receipt.deliveryStatus = 'saving';
|
|
report('saving');
|
|
for (const output of job.outputs) {
|
|
if (receipt.files.some((file) => file.index === output.index)) continue;
|
|
this.assertAccount(receipt.context);
|
|
const content = await this.options.client.download(job.execution_id, output.index, receipt.context.accountKey);
|
|
const relative = 'assets/generated/game-audio/' + job.execution_id + '/' + job.kind + '-' + (output.index + 1) + '.' + content.extension;
|
|
const target = path.join(receipt.context.projectPath, relative);
|
|
const lease = await this.options.leases.acquire(receipt.context.localProjectId, 'audio:' + job.execution_id, this.shutdown.signal);
|
|
try {
|
|
this.assertAccount(receipt.context);
|
|
await mkdir(path.dirname(target), { recursive: true });
|
|
try { await writeFile(target, content.bytes, { flag: 'wx' }); } catch (error) {
|
|
if (!audioRecord(error) || error.code !== 'EEXIST') throw error;
|
|
const existing = await readFile(target);
|
|
if (!existing.equals(Buffer.from(content.bytes))) throw new Error('Audio target already contains a different file', { cause: error });
|
|
}
|
|
receipt.files.push({ index: output.index, path: relative, bytes: content.bytes.byteLength, mimeType: content.mimeType });
|
|
await this.options.receipts.put(receipt);
|
|
try {
|
|
await this.options.recordTouchedPaths?.(receipt.context.conversationId, receipt.context.runId, [relative]);
|
|
} catch { /* Restored runs may no longer have an in-memory change tracker. Saved paths remain durable. */ }
|
|
} finally { lease.release(); }
|
|
}
|
|
receipt.deliveryStatus = 'saved';
|
|
delete receipt.errorCode;
|
|
await this.options.receipts.put(receipt);
|
|
report('saved');
|
|
return this.result(receipt, 'saved');
|
|
} catch (error) {
|
|
receipt.deliveryStatus = receipt.generation?.output_ready ? 'delivery_failed' : 'needs_reconciliation';
|
|
receipt.errorCode = error instanceof GameAudioClientError ? error.code : 'game_audio_delivery_paused';
|
|
await this.options.receipts.put(receipt);
|
|
report('paused');
|
|
if (!(error instanceof GameAudioClientError && error.code === 'plugin_account_changed')) this.scheduleRecovery();
|
|
return this.result(receipt, 'paused');
|
|
}
|
|
}
|
|
private result(receipt: Receipt, phase: AudioDeliveryResult['phase']): AudioDeliveryResult {
|
|
const job = receipt.generation;
|
|
return { executionId: job?.execution_id ?? null, logicalOperationId: receipt.context.logicalOperationId,
|
|
providerStatus: job?.provider_status ?? (receipt.submissionRejected ? 'failed' : 'submission_unknown'), deliveryStatus: receipt.deliveryStatus,
|
|
outputCount: job?.outputs.length ?? 0, files: [...receipt.files], phase,
|
|
progressPercent: job?.progress_percent ?? null,
|
|
billing: job?.billing ?? { mode: 'platform_metered', status: receipt.submissionRejected ? 'not_started' : 'receipt_unavailable' },
|
|
...(receipt.errorCode || job?.error_code ? { errorCode: receipt.errorCode ?? job?.error_code ?? undefined } : {}) };
|
|
}
|
|
}
|