import { mkdir, readFile, writeFile } from 'node:fs/promises'; import path from 'node:path'; import type { CapabilityBillingReceiptV1 } from '../../shared/data-service'; import { atomicWriteJson, readJsonFile } from '../coding-projects/atomic-json'; import type { PiProjectWriteLeaseCoordinator } from '../coding-runtime/pi/write-lease'; import type { GameResourceClient, GameResourceGenerateInput, GameResourceGeneration, } from './game-resource-client'; const RECEIPT_SCHEMA_VERSION = 1; const MAX_RECEIPTS = 1_000; const MAX_LOCAL_DELIVERY_ATTEMPTS = 3; const LOCAL_DELIVERY_RETRY_DELAY_MS = 1_000; const ACTIVE_PROVIDER_STATUSES = new Set([ 'reserved', 'accepted', 'running', ]); type HostedBillingReceipt = Extract; type DeliveryStatus = 'not_started' | 'waiting' | 'saving' | 'saved' | 'delivery_failed'; export interface GameResourceDeliveryContext { readonly conversationId: string; readonly runId: string; readonly localProjectId: string; readonly durableProjectId: string; readonly projectPath: string; readonly logicalOperationId: string; } export interface GameResourceDeliveredFile { readonly path: string; readonly bytes: number; } export interface GameResourceDeliveryResult { readonly executionId: string; readonly providerStatus: GameResourceGeneration['status']; readonly deliveryStatus: DeliveryStatus; readonly outputCount: number; readonly files: readonly GameResourceDeliveredFile[]; readonly billing: HostedBillingReceipt; readonly errorCode?: string; } export interface GameResourceDeliveryProgress extends GameResourceDeliveryResult { readonly phase: 'submitted' | 'generating' | 'saving' | 'saved'; } interface DeliveryReceipt { readonly logicalOperationId: string; readonly executionId: string; readonly conversationId: string; readonly runId: string; readonly localProjectId: string; readonly durableProjectId: string; readonly projectPath: string; readonly providerStatus: GameResourceGeneration['status']; readonly outputCount: number; readonly pollIntervalSeconds: number; readonly billing: HostedBillingReceipt; readonly deliveryStatus: DeliveryStatus; readonly files: readonly GameResourceDeliveredFile[]; readonly errorCode?: string; readonly updatedAt: string; } interface ReceiptFile { readonly schemaVersion: 1; readonly receipts: readonly DeliveryReceipt[]; } type WriteOutput = (filePath: string, bytes: Uint8Array) => Promise; type RecordTouchedPaths = ( conversationId: string, runId: string, paths: readonly string[], ) => Promise; type Sleep = (milliseconds: number, signal?: AbortSignal) => Promise; function isRecord(value: unknown): value is Record { return value !== null && typeof value === 'object' && !Array.isArray(value); } function isReceipt(value: unknown): value is DeliveryReceipt { if (!isRecord(value)) return false; return typeof value.logicalOperationId === 'string' && typeof value.executionId === 'string' && typeof value.conversationId === 'string' && typeof value.runId === 'string' && typeof value.localProjectId === 'string' && typeof value.durableProjectId === 'string' && typeof value.projectPath === 'string' && typeof value.providerStatus === 'string' && Number.isSafeInteger(value.outputCount) && Number.isSafeInteger(value.pollIntervalSeconds) && isRecord(value.billing) && typeof value.deliveryStatus === 'string' && Array.isArray(value.files) && typeof value.updatedAt === 'string'; } function resultFrom(receipt: DeliveryReceipt): GameResourceDeliveryResult { return { executionId: receipt.executionId, providerStatus: receipt.providerStatus, deliveryStatus: receipt.deliveryStatus, outputCount: receipt.outputCount, files: receipt.files, billing: receipt.billing, ...(receipt.errorCode ? { errorCode: receipt.errorCode } : {}), }; } function extensionFor(fileName: string | null, contentType: string): string { const named = fileName ? path.extname(fileName).toLowerCase() : ''; if (['.png', '.webp', '.jpg', '.jpeg', '.gif'].includes(named)) return named; const byType: Readonly> = { 'image/png': '.png', 'image/webp': '.webp', 'image/jpeg': '.jpg', 'image/gif': '.gif', }; return byType[contentType.split(';', 1)[0]?.trim().toLowerCase() ?? ''] ?? '.bin'; } async function defaultWriteOutput(filePath: string, bytes: Uint8Array): Promise { await writeFile(filePath, bytes, { flag: 'wx' }); } async function defaultSleep(milliseconds: number, signal?: AbortSignal): Promise { await new Promise((resolve, reject) => { if (signal?.aborted) { reject(new Error('Game Resource delivery interrupted')); return; } const finish = () => { signal?.removeEventListener('abort', abort); resolve(); }; const timeout = setTimeout(finish, milliseconds); const abort = () => { clearTimeout(timeout); signal?.removeEventListener('abort', abort); reject(new Error('Game Resource delivery interrupted')); }; signal?.addEventListener('abort', abort, { once: true }); }); } export class GameResourceDeliveryReceiptStore { private writeQueue: Promise = Promise.resolve(); constructor(private readonly filePath: string) {} async get(logicalOperationId: string): Promise { return (await this.read()).receipts.find( (receipt) => receipt.logicalOperationId === logicalOperationId, ) ?? null; } async list(): Promise { return (await this.read()).receipts; } async put(receipt: DeliveryReceipt): Promise { const operation = this.writeQueue.then(async () => { const current = await this.read(); const remaining = current.receipts.filter( (item) => item.logicalOperationId !== receipt.logicalOperationId, ); const receipts = [...remaining, receipt] .sort((left, right) => left.updatedAt.localeCompare(right.updatedAt)) .slice(-MAX_RECEIPTS); await atomicWriteJson(this.filePath, { schemaVersion: RECEIPT_SCHEMA_VERSION, receipts, } satisfies ReceiptFile); }); this.writeQueue = operation.catch(() => undefined); await operation; } private async read(): Promise { let value: unknown; try { value = await readJsonFile(this.filePath); } catch (error) { if (isRecord(error) && error.code === 'ENOENT') { return { schemaVersion: RECEIPT_SCHEMA_VERSION, receipts: [] }; } throw error; } if (!isRecord(value) || value.schemaVersion !== RECEIPT_SCHEMA_VERSION || !Array.isArray(value.receipts) || !value.receipts.every(isReceipt)) { throw new Error('Game Resource delivery receipt file is invalid'); } return value as unknown as ReceiptFile; } } export interface GameResourceDeliveryCoordinatorOptions { readonly client: Pick; readonly receipts: GameResourceDeliveryReceiptStore; readonly leases: PiProjectWriteLeaseCoordinator; readonly sleep?: Sleep; readonly writeOutput?: WriteOutput; readonly recordTouchedPaths?: RecordTouchedPaths; readonly acquireBackgroundLease?: (lease: { id: string; kind: 'coding-run'; }) => () => void; } export class GameResourceDeliveryCoordinator { private readonly flights = new Map>(); private readonly sleep: Sleep; private readonly writeOutput: WriteOutput; private readonly shutdown = new AbortController(); constructor(private readonly options: GameResourceDeliveryCoordinatorOptions) { this.sleep = options.sleep ?? defaultSleep; this.writeOutput = options.writeOutput ?? defaultWriteOutput; } async generateAndMaterialize(input: { readonly context: GameResourceDeliveryContext; readonly request: GameResourceGenerateInput; readonly onProgress?: (progress: GameResourceDeliveryProgress) => void; }): Promise { const operationId = input.context.logicalOperationId; const existingFlight = this.flights.get(operationId); if (existingFlight) return await existingFlight; const releaseBackground = this.options.acquireBackgroundLease?.({ id: `game-resource:${operationId}`, kind: 'coding-run', }) ?? (() => undefined); const flight = this.run(input).finally(releaseBackground); this.flights.set(operationId, flight); try { return await flight; } finally { if (this.flights.get(operationId) === flight) this.flights.delete(operationId); } } async resumePending(): Promise { const receipts = await this.options.receipts.list(); const pending = receipts.filter((receipt) => receipt.deliveryStatus !== 'saved' && (receipt.providerStatus === 'succeeded' || ACTIVE_PROVIDER_STATUSES.has(receipt.providerStatus))); const results: GameResourceDeliveryResult[] = []; for (const receipt of pending) { const existingFlight = this.flights.get(receipt.logicalOperationId); if (existingFlight) { try { results.push(await existingFlight); } catch { // One unavailable reconciliation must not block other persisted deliveries. } continue; } const releaseBackground = this.options.acquireBackgroundLease?.({ id: `game-resource:${receipt.logicalOperationId}`, kind: 'coding-run', }) ?? (() => undefined); const flight = this.continueReceipt(receipt).finally(releaseBackground); this.flights.set(receipt.logicalOperationId, flight); try { results.push(await flight); } catch { // Keep the receipt for a later retry and continue with independent executions. } finally { if (this.flights.get(receipt.logicalOperationId) === flight) { this.flights.delete(receipt.logicalOperationId); } } } return results; } dispose(): void { this.shutdown.abort(); } private async run(input: { readonly context: GameResourceDeliveryContext; readonly request: GameResourceGenerateInput; readonly onProgress?: (progress: GameResourceDeliveryProgress) => void; }): Promise { const persisted = await this.options.receipts.get(input.context.logicalOperationId); if (persisted) return await this.continueReceipt(persisted, input.onProgress); const generated = await this.options.client.generate(input.request); let receipt = this.receiptFromGeneration(input.context, generated); await this.options.receipts.put(receipt); this.report(input.onProgress, 'submitted', receipt); return await this.continueReceipt(receipt, input.onProgress); } private async continueReceipt( initial: DeliveryReceipt, onProgress?: (progress: GameResourceDeliveryProgress) => void, ): Promise { let receipt = initial; if (receipt.deliveryStatus === 'saved') return resultFrom(receipt); while (ACTIVE_PROVIDER_STATUSES.has(receipt.providerStatus)) { this.report(onProgress, 'generating', receipt); await this.sleep(receipt.pollIntervalSeconds * 1_000, this.shutdown.signal); const generation = await this.options.client.get(receipt.executionId); receipt = { ...receipt, providerStatus: generation.status, outputCount: generation.outputCount, pollIntervalSeconds: generation.pollIntervalSeconds ?? receipt.pollIntervalSeconds, billing: generation.billing, ...(generation.errorCode ? { errorCode: generation.errorCode } : {}), updatedAt: new Date().toISOString(), }; await this.options.receipts.put(receipt); } if (receipt.providerStatus !== 'succeeded') { const terminal = { ...receipt, deliveryStatus: 'not_started' as const, updatedAt: new Date().toISOString(), }; await this.options.receipts.put(terminal); return resultFrom(terminal); } return await this.materializeWithRetry(receipt, onProgress); } private async materializeWithRetry( receipt: DeliveryReceipt, onProgress?: (progress: GameResourceDeliveryProgress) => void, ): Promise { let result = await this.materialize(receipt, onProgress); for (let attempt = 1; attempt < MAX_LOCAL_DELIVERY_ATTEMPTS && result.deliveryStatus === 'delivery_failed'; attempt += 1) { await this.sleep(LOCAL_DELIVERY_RETRY_DELAY_MS, this.shutdown.signal); const persisted = await this.options.receipts.get(receipt.logicalOperationId); result = await this.materialize(persisted ?? receipt, onProgress); } return result; } private async materialize( receipt: DeliveryReceipt, onProgress?: (progress: GameResourceDeliveryProgress) => void, ): Promise { const saving: DeliveryReceipt = { ...receipt, deliveryStatus: 'saving', files: [], updatedAt: new Date().toISOString(), }; await this.options.receipts.put(saving); this.report(onProgress, 'saving', saving); let downloads: Awaited>[]; try { downloads = await Promise.all(Array.from( { length: receipt.outputCount }, async (_, index) => await this.options.client.download(receipt.executionId, index), )); } catch { return await this.failDelivery(receipt); } const files = downloads.map((download, index) => ({ path: `assets/generated/game-resource/${receipt.executionId}/output-${index + 1}${extensionFor(download.fileName, download.contentType)}`, bytes: download.bytes.byteLength, content: download.bytes, })); const lease = await this.options.leases.acquire( receipt.localProjectId, `game-resource:${receipt.logicalOperationId}`, ); try { for (const file of files) { const target = path.join(receipt.projectPath, ...file.path.split('/')); await mkdir(path.dirname(target), { recursive: true }); try { await this.writeOutput(target, file.content); } catch (error) { if (!isRecord(error) || error.code !== 'EEXIST') throw error; const existing = await readFile(target); if (!existing.equals(Buffer.from(file.content))) throw error; } } const saved: DeliveryReceipt = { ...receipt, deliveryStatus: 'saved', files: files.map(({ path: filePath, bytes }) => ({ path: filePath, bytes })), errorCode: undefined, updatedAt: new Date().toISOString(), }; await this.options.receipts.put(saved); try { await this.options.recordTouchedPaths?.( receipt.conversationId, receipt.runId, saved.files.map((file) => file.path), ); } catch { // The original run may no longer exist after an app restart; saved files remain authoritative. } this.report(onProgress, 'saved', saved); return resultFrom(saved); } catch { return await this.failDelivery(receipt); } finally { lease.release(); } } private async failDelivery(receipt: DeliveryReceipt): Promise { const failed: DeliveryReceipt = { ...receipt, deliveryStatus: 'delivery_failed', files: [], errorCode: 'game_resource_delivery_failed', updatedAt: new Date().toISOString(), }; await this.options.receipts.put(failed); return resultFrom(failed); } private receiptFromGeneration( context: GameResourceDeliveryContext, generation: GameResourceGeneration, ): DeliveryReceipt { return { logicalOperationId: context.logicalOperationId, executionId: generation.executionId, conversationId: context.conversationId, runId: context.runId, localProjectId: context.localProjectId, durableProjectId: context.durableProjectId, projectPath: context.projectPath, providerStatus: generation.status, outputCount: generation.outputCount, pollIntervalSeconds: generation.pollIntervalSeconds ?? 3, billing: generation.billing, deliveryStatus: generation.status === 'succeeded' ? 'waiting' : 'not_started', files: [], ...(generation.errorCode ? { errorCode: generation.errorCode } : {}), updatedAt: new Date().toISOString(), }; } private report( callback: ((progress: GameResourceDeliveryProgress) => void) | undefined, phase: GameResourceDeliveryProgress['phase'], receipt: DeliveryReceipt, ): void { try { callback?.({ phase, ...resultFrom(receipt) }); } catch { // Progress is observational and must not alter provider or delivery state. } } }