feat: integrate automatic game resource delivery
This commit is contained in:
471
electron/services/game-resource-delivery.ts
Normal file
471
electron/services/game-resource-delivery.ts
Normal file
@@ -0,0 +1,471 @@
|
||||
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<GameResourceGeneration['status']>([
|
||||
'reserved',
|
||||
'accepted',
|
||||
'running',
|
||||
]);
|
||||
|
||||
type HostedBillingReceipt = Extract<CapabilityBillingReceiptV1, { mode: 'platform_metered' }>;
|
||||
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<void>;
|
||||
type RecordTouchedPaths = (
|
||||
conversationId: string,
|
||||
runId: string,
|
||||
paths: readonly string[],
|
||||
) => Promise<void>;
|
||||
type Sleep = (milliseconds: number, signal?: AbortSignal) => Promise<void>;
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
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<Record<string, string>> = {
|
||||
'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<void> {
|
||||
await writeFile(filePath, bytes, { flag: 'wx' });
|
||||
}
|
||||
|
||||
async function defaultSleep(milliseconds: number, signal?: AbortSignal): Promise<void> {
|
||||
await new Promise<void>((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<void> = Promise.resolve();
|
||||
|
||||
constructor(private readonly filePath: string) {}
|
||||
|
||||
async get(logicalOperationId: string): Promise<DeliveryReceipt | null> {
|
||||
return (await this.read()).receipts.find(
|
||||
(receipt) => receipt.logicalOperationId === logicalOperationId,
|
||||
) ?? null;
|
||||
}
|
||||
|
||||
async list(): Promise<readonly DeliveryReceipt[]> {
|
||||
return (await this.read()).receipts;
|
||||
}
|
||||
|
||||
async put(receipt: DeliveryReceipt): Promise<void> {
|
||||
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<ReceiptFile> {
|
||||
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<GameResourceClient, 'generate' | 'get' | 'download'>;
|
||||
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<string, Promise<GameResourceDeliveryResult>>();
|
||||
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<GameResourceDeliveryResult> {
|
||||
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<readonly GameResourceDeliveryResult[]> {
|
||||
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<GameResourceDeliveryResult> {
|
||||
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<GameResourceDeliveryResult> {
|
||||
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<GameResourceDeliveryResult> {
|
||||
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<GameResourceDeliveryResult> {
|
||||
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<ReturnType<GameResourceDeliveryCoordinatorOptions['client']['download']>>[];
|
||||
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<GameResourceDeliveryResult> {
|
||||
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.
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user