501 lines
38 KiB
TypeScript
501 lines
38 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
import { mkdir, rename, stat, rm, lstat } from 'node:fs/promises';
|
|
import { dirname, join, resolve } from 'node:path';
|
|
import { db } from '../db/index.js';
|
|
import { config } from '../config.js';
|
|
import { getHost, getHostOperation, listHosts, queueHostOperation } from '../hosts/index.js';
|
|
import { ensureGameFileSchema } from './schema.js';
|
|
import { ensureHostSchema } from '../hosts/schema.js';
|
|
import { manifestChecksum, receiveGameFile, validateGamePath, validateManifest, verifyGameFile } from './storage.js';
|
|
import { MAX_UPLOAD_CHUNK_BYTES, validateUploadPlan, getUploadFileProgress, receiveUploadChunk, hashUploadedFiles, withUploadLock } from './uploads.js';
|
|
import { GAME_FILE_GAMES } from '../../game-files.js';
|
|
import type { GameFileEntry, GameLibrary, GameLibraryGame, GameLibraryVersion, HostLibraryCache, LibraryCacheState, LibraryTransferProgress } from '../../game-files.js';
|
|
|
|
function database() { const handle = db(); ensureGameFileSchema(handle); ensureHostSchema(handle); return handle; }
|
|
export const gameFilesRoot = () => resolve(process.env.GAME_FILES_ROOT ?? join(dirname(config.dbPath), 'game-files'));
|
|
const stagingRoot = (id: string) => join(gameFilesRoot(), '.imports', id);
|
|
type VersionRow = { id: string; library_id: string; game: GameLibraryGame; version: string; description: string; dlc: string; state: GameLibraryVersion['state']; size_bytes: number; manifest: string; manifest_sha256: string | null; created_at: string; published_at: string | null; storage_location: string; source_host_id: string; source_volume: string; error: string | null; operation_id: string | null };
|
|
type CacheRow = { host_id: string; version_id: string; status: LibraryCacheState; bytes_downloaded: number; bytes_total: number; manifest_sha256: string | null; error: string | null; updated_at: string; operation_id: string | null; bytes_verified: number | null; transfer_progress: string | null };
|
|
const cacheFromRow = (row: CacheRow): HostLibraryCache => ({ hostId: row.host_id, libraryVersionId: row.version_id, status: row.status, bytesDownloaded: row.bytes_downloaded, bytesTotal: row.bytes_total, manifestSha256: row.manifest_sha256, error: row.error, updatedAt: row.updated_at, operationId: row.operation_id,
|
|
...(row.bytes_verified == null ? {} : { bytesVerified: row.bytes_verified }),
|
|
...(row.transfer_progress ? { transfer: JSON.parse(row.transfer_progress) as LibraryTransferProgress } : {}) });
|
|
function versionFromRow(row: VersionRow & { file_count?: number }, summaryCaches?: Map<string, HostLibraryCache[]>): GameLibraryVersion {
|
|
const files = summaryCaches ? [] : JSON.parse(row.manifest) as GameFileEntry[];
|
|
return { id: row.id, libraryId: row.library_id, game: row.game, version: row.version,
|
|
description: row.description, state: row.state, sizeBytes: row.size_bytes,
|
|
filesRemoved: !!database().prepare('SELECT 1 FROM game_file_cleanup WHERE version_id=? AND removed_at IS NOT NULL').get(row.id),
|
|
filesRemoving: !!database().prepare('SELECT 1 FROM game_file_cleanup WHERE version_id=? AND removed_at IS NULL').get(row.id),
|
|
files, fileCount: row.file_count ?? (files.length || (database().prepare('SELECT COUNT(*) AS count FROM game_file_upload_entries WHERE version_id=?').get(row.id) as { count: number }).count), manifestSha256: row.manifest_sha256, createdAt: row.created_at,
|
|
publishedAt: row.published_at, storageLocation: row.storage_location, sourceHostId: row.source_host_id || null,
|
|
sourceVolume: row.source_volume || null, error: row.error, operationId: row.operation_id,
|
|
caches: summaryCaches ? summaryCaches.get(row.id) ?? [] : (database().prepare('SELECT * FROM host_library_caches WHERE version_id = ?').all(row.id) as CacheRow[]).map(cacheFromRow) };
|
|
}
|
|
export function getLibraryVersion(id: string): GameLibraryVersion | null {
|
|
const row = database().prepare('SELECT v.*, l.game FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id WHERE v.id=?').get(id) as VersionRow | undefined;
|
|
return row ? versionFromRow(row) : null;
|
|
}
|
|
export function libraryFilesRetired(versionId: string): boolean {
|
|
return !!database().prepare('SELECT 1 FROM game_file_cleanup WHERE version_id=?').get(versionId);
|
|
}
|
|
export function getLibraryFile(versionId: string, path: string) {
|
|
validateGamePath(path);
|
|
return database().prepare(`SELECT f.path,f.size,f.sha256,v.state,v.source_host_id AS sourceHostId,v.storage_location AS storageLocation
|
|
FROM game_file_entries f JOIN game_file_versions v ON v.id=f.version_id WHERE f.version_id=? AND f.path=?`)
|
|
.get(versionId, path) as (GameFileEntry & { state: GameLibraryVersion['state']; sourceHostId: string; storageLocation: string }) | undefined ?? null;
|
|
}
|
|
/** Launch selection needs only the ID; manifests can contain tens of thousands of files. */
|
|
export function newestReadyLibraryId(game: string): string | null {
|
|
const row = database().prepare("SELECT v.id FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id WHERE l.game=? AND v.state='ready' ORDER BY v.published_at DESC, v.created_at DESC, v.rowid DESC LIMIT 1").get(game) as { id: string } | undefined;
|
|
return row?.id ?? null;
|
|
}
|
|
export function newestReadyLibrary(game: string): GameLibraryVersion | null {
|
|
const id = newestReadyLibraryId(game);
|
|
return id ? getLibraryVersion(id) : null;
|
|
}
|
|
/** Hosts follow the latest successful publication for each game, including legacy libraries. */
|
|
export function latestPublishedLibraries() {
|
|
const latest = database().prepare(`SELECT v.id, l.game, v.version, v.size_bytes AS sizeBytes, v.manifest_sha256 AS manifestSha256, v.published_at AS publishedAt
|
|
FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id
|
|
WHERE v.state='ready' AND l.game=?
|
|
ORDER BY v.published_at DESC,v.created_at DESC,v.rowid DESC LIMIT 1`);
|
|
return GAME_FILE_GAMES.flatMap(game => {
|
|
const version = latest.get(game.id) as { id: string; game: GameLibraryGame; version: string; sizeBytes: number; manifestSha256: string; publishedAt: string | null } | undefined;
|
|
return version ? [{ ...version, name: game.name }] : [];
|
|
});
|
|
}
|
|
export function listLibraries(summary = false): GameLibrary[] {
|
|
reconcileLibraryOperations();
|
|
let summaryCaches: Map<string, HostLibraryCache[]> | undefined;
|
|
if (summary) {
|
|
summaryCaches = new Map();
|
|
for (const row of database().prepare('SELECT * FROM host_library_caches').all() as CacheRow[]) {
|
|
const items = summaryCaches.get(row.version_id) ?? []; items.push(cacheFromRow(row)); summaryCaches.set(row.version_id, items);
|
|
}
|
|
}
|
|
const latest = new Map(latestPublishedLibraries().map(version => [version.game, version.id]));
|
|
return GAME_FILE_GAMES.map(game => {
|
|
const versions = (database().prepare(`SELECT ${summary ? `v.id, v.library_id, v.version, v.description, v.dlc, v.state, v.size_bytes, v.manifest_sha256, v.created_at, v.published_at, v.storage_location, v.source_host_id, v.source_volume, v.error, v.operation_id, CASE WHEN json_array_length(v.manifest)>0 THEN json_array_length(v.manifest) ELSE (SELECT COUNT(*) FROM game_file_upload_entries e WHERE e.version_id=v.id) END AS file_count` : 'v.*'}, l.game FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id
|
|
WHERE l.game=? ORDER BY v.created_at DESC,v.rowid DESC`).all(game.id) as VersionRow[]).map(row => versionFromRow(row, summaryCaches));
|
|
const firstLibrary = database().prepare('SELECT created_at FROM game_file_libraries WHERE game=? ORDER BY created_at,rowid LIMIT 1').get(game.id) as { created_at: string } | undefined;
|
|
return { id: game.id, name: game.name, game: game.id, description: '', createdAt: firstLibrary?.created_at ?? '', latestVersionId: latest.get(game.id) ?? null, versions };
|
|
});
|
|
}
|
|
export function getHostLibraryCache(hostId: string, versionId: string): HostLibraryCache | null {
|
|
const row = database().prepare('SELECT * FROM host_library_caches WHERE host_id=? AND version_id=?').get(hostId, versionId) as CacheRow | undefined;
|
|
return row ? cacheFromRow(row) : null;
|
|
}
|
|
/** Admission and dispatch recheck readiness without loading or parsing the file manifest. */
|
|
export function assertReadyLibrary(hostId: string, versionId: string): void {
|
|
const handle = database();
|
|
const version = handle.prepare('SELECT state, manifest_sha256 FROM game_file_versions WHERE id=?')
|
|
.get(versionId) as Pick<VersionRow, 'state' | 'manifest_sha256'> | undefined;
|
|
if (!version || version.state !== 'ready' || libraryFilesRetired(versionId)) throw new Error('Select an available published game-file version.');
|
|
if (handle.prepare("SELECT 1 FROM ec_host_operations WHERE host_id=? AND kind='library.remove' AND json_extract(payload,'$.libraryVersionId')=? AND state IN ('pending','accepted','running','unknown')").get(hostId, versionId)) throw new Error('This host copy is being removed.');
|
|
const cache = handle.prepare('SELECT status, manifest_sha256 FROM host_library_caches WHERE host_id=? AND version_id=?')
|
|
.get(hostId, versionId) as Pick<CacheRow, 'status' | 'manifest_sha256'> | undefined;
|
|
if (!cache || !['ready', 'outdated'].includes(cache.status) || cache.manifest_sha256 !== version.manifest_sha256) throw new Error('The selected host must finish downloading and verifying this game-file version.');
|
|
}
|
|
export function requireReadyLibrary(hostId: string, versionId: string): GameLibraryVersion {
|
|
assertReadyLibrary(hostId, versionId);
|
|
return getLibraryVersion(versionId)!;
|
|
}
|
|
|
|
function textField(input: unknown, label: string, max: number, required = true): string {
|
|
if (typeof input !== 'string' || input.length > max || (required && !input.trim())) throw new Error(`Provide a valid ${label}.`);
|
|
return input.trim();
|
|
}
|
|
|
|
/** Reuse a game's original storage parent without rewriting immutable historical versions. */
|
|
function libraryForGame(game: GameLibraryGame, now: string): string {
|
|
const handle = database();
|
|
const existing = handle.prepare('SELECT id FROM game_file_libraries WHERE game=? ORDER BY created_at,rowid LIMIT 1').get(game) as { id: string } | undefined;
|
|
if (existing) return existing.id;
|
|
const id = randomUUID();
|
|
// The legacy name column is globally unique, so its private value must not collide with an old user label.
|
|
handle.prepare('INSERT INTO game_file_libraries(id,name,game,description,created_at) VALUES(?,?,?,?,?)').run(id, `game-files:${game}:${id}`, game, '', now);
|
|
return id;
|
|
}
|
|
|
|
/** Register a folder on the controller; no host or Docker volume is needed. */
|
|
export function createLibraryUpload(input: Record<string, unknown>): GameLibraryVersion {
|
|
const game = input.game;
|
|
if (game !== 't5' && game !== 't6') throw new Error('Choose T5 or T6 game files.');
|
|
const files = validateUploadPlan(input.files);
|
|
const handle = database();
|
|
const id = randomUUID(); const now = new Date().toISOString();
|
|
handle.transaction(() => {
|
|
const libraryId = libraryForGame(game, now);
|
|
// Keep legacy non-null columns for existing databases and published versions.
|
|
handle.prepare(`INSERT INTO game_file_versions(id,library_id,version,description,dlc,state,size_bytes,created_at,storage_location,source_host_id,source_volume)
|
|
VALUES(?,?,?,?,'[]','importing',?,?,?,'','')`).run(id, libraryId, id, '', files.reduce((sum, file) => sum + file.size, 0), now, join(gameFilesRoot(), game, id));
|
|
const insert = handle.prepare('INSERT INTO game_file_upload_entries(version_id,path,size,last_modified) VALUES(?,?,?,?)');
|
|
for (const file of files) insert.run(id, file.path, file.size, file.lastModified);
|
|
})();
|
|
return getLibraryVersion(id)!;
|
|
}
|
|
|
|
type UploadEntry = { path: string; size: number; lastModified: number };
|
|
function requireBrowserUpload(id: string): GameLibraryVersion {
|
|
const version = getLibraryVersion(id);
|
|
if (!version || version.sourceHostId !== null || !database().prepare('SELECT 1 FROM game_file_upload_entries WHERE version_id=? LIMIT 1').get(id)) {
|
|
throw Object.assign(new Error('Folder upload not found.'), { status: 404 });
|
|
}
|
|
return version;
|
|
}
|
|
function uploadEntries(id: string): UploadEntry[] {
|
|
return database().prepare('SELECT path,size,last_modified AS lastModified FROM game_file_upload_entries WHERE version_id=? ORDER BY path').all(id) as UploadEntry[];
|
|
}
|
|
async function uploadStorageRoot(version: GameLibraryVersion): Promise<string> {
|
|
// Recover publication interrupted after the atomic rename but before its DB commit.
|
|
try { if ((await stat(version.storageLocation)).isDirectory()) return version.storageLocation; }
|
|
catch (cause) { if ((cause as NodeJS.ErrnoException).code !== 'ENOENT') throw cause; }
|
|
return stagingRoot(version.id);
|
|
}
|
|
export async function browserUploadStatus(id: string) {
|
|
const version = requireBrowserUpload(id);
|
|
const status = version.state === 'ready' ? 'published' as const : publishing.has(id) ? 'verifying' as const : version.state === 'failed' ? 'failed' as const : 'pending' as const;
|
|
const root = await uploadStorageRoot(version);
|
|
const entries = uploadEntries(id);
|
|
const files = [];
|
|
// Avoid opening thousands of files concurrently for a large folder.
|
|
for (const file of entries) files.push({ ...file, ...(status === 'published' ? { receivedBytes: file.size, complete: true } : await getUploadFileProgress(root, file)) });
|
|
const { files: _manifest, ...summary } = version;
|
|
return { status, chunkSize: MAX_UPLOAD_CHUNK_BYTES, files, version: summary, error: version.error };
|
|
}
|
|
export async function uploadBrowserFile(id: string, path: string, chunk: { offset: number; size: number; sha256: string }, body: ReadableStream<Uint8Array> | null) {
|
|
validateGamePath(path);
|
|
const checkWritable = () => {
|
|
const version = requireBrowserUpload(id);
|
|
if (version.state === 'ready' || version.manifestSha256 || publishing.has(id)) throw Object.assign(new Error('This upload is no longer writable.'), { status: 409 });
|
|
};
|
|
checkWritable();
|
|
const file = database().prepare('SELECT path,size,last_modified AS lastModified FROM game_file_upload_entries WHERE version_id=? AND path=?').get(id, path) as UploadEntry | undefined;
|
|
if (!file) throw new Error('File is not listed in this upload.');
|
|
const progress = await receiveUploadChunk(stagingRoot(id), file, chunk, body, checkWritable);
|
|
database().prepare("UPDATE game_file_versions SET state='importing',error=NULL WHERE id=? AND state IN ('importing','failed')").run(id);
|
|
return progress;
|
|
}
|
|
export async function publishBrowserUpload(id: string): Promise<GameLibraryVersion> {
|
|
const version = requireBrowserUpload(id);
|
|
if (version.state === 'ready') return version;
|
|
const existing = publishing.get(id); if (existing) return existing;
|
|
const task = withUploadLock(stagingRoot(id), async () => {
|
|
const root = await uploadStorageRoot(version);
|
|
const files = await hashUploadedFiles(root, uploadEntries(id));
|
|
const checksum = manifestChecksum(files);
|
|
if (version.manifestSha256 && checksum !== version.manifestSha256) throw new Error('Stored files no longer match the verified manifest.');
|
|
database().transaction(() => {
|
|
database().prepare("UPDATE game_file_versions SET state='importing',manifest=?,manifest_sha256=?,error=NULL WHERE id=? AND state IN ('importing','failed')").run(JSON.stringify(files), checksum, id);
|
|
database().prepare('DELETE FROM game_file_entries WHERE version_id=?').run(id);
|
|
const insert = database().prepare('INSERT INTO game_file_entries(version_id,path,size,sha256) VALUES(?,?,?,?)');
|
|
for (const file of files) insert.run(id, file.path, file.size, file.sha256);
|
|
})();
|
|
if (root !== version.storageLocation) {
|
|
await mkdir(dirname(version.storageLocation), { recursive: true });
|
|
await rename(root, version.storageLocation);
|
|
}
|
|
database().prepare("UPDATE game_file_versions SET state='ready',published_at=?,error=NULL WHERE id=? AND state='importing'").run(new Date().toISOString(), id);
|
|
return getLibraryVersion(id)!;
|
|
});
|
|
publishing.set(id, task);
|
|
try { const result = await task; await cleanupControllerLibraries(); return result; } finally { publishing.delete(id); }
|
|
}
|
|
export async function requestBrowserPublication(id: string) {
|
|
const version = requireBrowserUpload(id);
|
|
if (version.state !== 'ready' && !publishing.has(id)) {
|
|
database().prepare("UPDATE game_file_versions SET state='importing',error=NULL WHERE id=? AND state IN ('importing','failed')").run(id);
|
|
void publishBrowserUpload(id).catch(cause => {
|
|
database().prepare("UPDATE game_file_versions SET state='failed',error=? WHERE id=? AND state='importing'")
|
|
.run((cause instanceof Error ? cause.message : 'Folder publication failed.').slice(0, 2000), id);
|
|
});
|
|
}
|
|
return browserUploadStatus(id);
|
|
}
|
|
|
|
export function createLibraryImport(input: Record<string, unknown>): GameLibraryVersion {
|
|
const game = input.game;
|
|
if (game !== 't5' && game !== 't6') throw new Error('Choose T5 or T6 game files.');
|
|
const hostId = textField(input.hostId, 'source host', 128);
|
|
const host = getHost(hostId);
|
|
if (!host || !host.connected || !host.credentialActive || host.mode === 'disabled') throw new Error('Choose an online, enrolled source host.');
|
|
const sourceVolume = game === 't5' ? 't5_game_files' : 't6_game_files';
|
|
const handle = database();
|
|
const id = randomUUID(); const now = new Date().toISOString();
|
|
handle.transaction(() => {
|
|
const libraryId = libraryForGame(game, now);
|
|
handle.prepare(`INSERT INTO game_file_versions(id,library_id,version,description,dlc,state,created_at,storage_location,source_host_id,source_volume)
|
|
VALUES(?,?,?,?,?,'importing',?,?,?,?)`).run(id, libraryId, id, '', '[]', now, join(gameFilesRoot(), game, id), hostId, sourceVolume);
|
|
})();
|
|
try {
|
|
const operation = queueHostOperation(hostId, 'library.import', {
|
|
importId: id, game, sourceVolume, manifestUrl: `/api/agent/imports/${id}`,
|
|
uploadBaseUrl: `/api/agent/imports/${id}/files`, completeUrl: `/api/agent/imports/${id}/complete`
|
|
}, { idempotencyKey: `library-import:${id}` });
|
|
handle.prepare('UPDATE game_file_versions SET operation_id=? WHERE id=?').run(operation.id, id);
|
|
} catch (cause) {
|
|
handle.prepare("UPDATE game_file_versions SET state='failed',error=? WHERE id=?").run((cause as Error).message, id);
|
|
throw cause;
|
|
}
|
|
return getLibraryVersion(id)!;
|
|
}
|
|
|
|
export function requireImportOwner(hostId: string, id: string): GameLibraryVersion {
|
|
const version = getLibraryVersion(id);
|
|
if (!version || version.sourceHostId !== hostId) throw new Error('Import not found for this host.');
|
|
return version;
|
|
}
|
|
export function setImportManifest(hostId: string, id: string, input: unknown): GameLibraryVersion {
|
|
const version = requireImportOwner(hostId, id);
|
|
const files = validateManifest(input); const checksum = manifestChecksum(files);
|
|
if (version.manifestSha256) {
|
|
if (version.manifestSha256 !== checksum) throw new Error('An import manifest cannot change. Start a new version.');
|
|
return version;
|
|
}
|
|
if (version.state !== 'importing') throw new Error('This import is not writable.');
|
|
database().transaction(() => {
|
|
database().prepare('UPDATE game_file_versions SET manifest=?,manifest_sha256=?,size_bytes=? WHERE id=? AND state=\'importing\'').run(JSON.stringify(files), checksum, files.reduce((sum, file) => sum + file.size, 0), id);
|
|
const insert = database().prepare('INSERT INTO game_file_entries(version_id,path,size,sha256) VALUES(?,?,?,?)');
|
|
for (const file of files) insert.run(id, file.path, file.size, file.sha256);
|
|
})();
|
|
return getLibraryVersion(id)!;
|
|
}
|
|
export async function uploadImportFile(hostId: string, id: string, path: string, body: ReadableStream<Uint8Array> | null): Promise<void> {
|
|
const file = getLibraryFile(id, path);
|
|
if (!file || file.sourceHostId !== hostId) throw new Error('File is not listed in a manifest belonging to this host.');
|
|
if (file.state !== 'importing' || publishing.has(id)) throw new Error('This import is not writable.');
|
|
await receiveGameFile(stagingRoot(id), file, body);
|
|
}
|
|
|
|
const publishing = new Map<string, Promise<GameLibraryVersion>>();
|
|
export async function publishImport(hostId: string, id: string): Promise<GameLibraryVersion> {
|
|
const version = requireImportOwner(hostId, id);
|
|
if (version.state === 'ready') return version;
|
|
if (version.state !== 'importing' || !version.files.length || !version.manifestSha256) throw new Error('Import needs a complete manifest before publication.');
|
|
const existing = publishing.get(id); if (existing) return existing;
|
|
const task = (async () => {
|
|
let root = stagingRoot(id);
|
|
// A restart between the atomic directory rename and metadata commit is recoverable.
|
|
try { if ((await stat(version.storageLocation)).isDirectory()) root = version.storageLocation; }
|
|
catch (cause) { if ((cause as NodeJS.ErrnoException).code !== 'ENOENT') throw cause; }
|
|
for (const file of version.files) await verifyGameFile(root, file);
|
|
if (root !== version.storageLocation) {
|
|
await mkdir(dirname(version.storageLocation), { recursive: true });
|
|
await rename(root, version.storageLocation);
|
|
}
|
|
database().prepare("UPDATE game_file_versions SET state='ready',published_at=?,error=NULL WHERE id=? AND state='importing'").run(new Date().toISOString(), id);
|
|
return getLibraryVersion(id)!;
|
|
})();
|
|
publishing.set(id, task);
|
|
try { const result = await task; await cleanupControllerLibraries(); return result; } finally { publishing.delete(id); }
|
|
}
|
|
|
|
/** Large libraries can take minutes to verify. Keep HTTP requests bounded while the source host polls. */
|
|
export function importPublicationStatus(hostId: string, id: string) {
|
|
const version = requireImportOwner(hostId, id);
|
|
return {
|
|
status: version.state === 'ready' ? 'published' as const : version.state === 'failed' ? 'failed' as const : publishing.has(id) ? 'verifying' as const : 'pending' as const,
|
|
version: { id: version.id, game: version.game, sizeBytes: version.sizeBytes, manifestSha256: version.manifestSha256 },
|
|
error: version.error
|
|
};
|
|
}
|
|
export function requestImportPublication(hostId: string, id: string) {
|
|
const version = requireImportOwner(hostId, id);
|
|
if (version.state === 'importing' && !publishing.has(id)) {
|
|
if (!version.files.length || !version.manifestSha256) throw new Error('Import requires a manifest before publication.');
|
|
void publishImport(hostId, id).catch(cause => {
|
|
database().prepare("UPDATE game_file_versions SET state='failed',error=? WHERE id=? AND state='importing'")
|
|
.run((cause instanceof Error ? cause.message : 'Library publication failed.').slice(0, 2000), id);
|
|
});
|
|
}
|
|
return importPublicationStatus(hostId, id);
|
|
}
|
|
|
|
function transferProgress(value: unknown, status: LibraryCacheState): LibraryTransferProgress | null {
|
|
if (value === null || value === undefined) return null;
|
|
if (typeof value !== 'object' || Array.isArray(value)) throw new Error('Invalid transfer progress.');
|
|
const progress = value as Record<string, unknown>;
|
|
const nonnegative = (amount: unknown): amount is number => typeof amount === 'number' && Number.isFinite(amount) && amount >= 0 && amount <= Number.MAX_SAFE_INTEGER;
|
|
if (!['downloading', 'verifying'].includes(progress.phase as string)
|
|
|| typeof progress.startedAt !== 'string' || progress.startedAt.length > 80
|
|
|| !/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|[+-]\d{2}:\d{2})$/.test(progress.startedAt)
|
|
|| !Number.isFinite(Date.parse(progress.startedAt))
|
|
|| !nonnegative(progress.elapsedSeconds) || !nonnegative(progress.phaseElapsedSeconds)
|
|
|| progress.phaseElapsedSeconds > progress.elapsedSeconds
|
|
|| (progress.bytesPerSecond != null && !nonnegative(progress.bytesPerSecond))
|
|
|| (progress.etaSeconds != null && !nonnegative(progress.etaSeconds))
|
|
|| (['downloading', 'verifying'].includes(status) && progress.phase !== status)
|
|
|| (['ready', 'outdated'].includes(status) && progress.phase !== 'verifying')) {
|
|
throw new Error('Invalid transfer progress.');
|
|
}
|
|
return { phase: progress.phase as LibraryTransferProgress['phase'], startedAt: progress.startedAt,
|
|
elapsedSeconds: progress.elapsedSeconds, phaseElapsedSeconds: progress.phaseElapsedSeconds,
|
|
bytesPerSecond: progress.bytesPerSecond as number | null ?? null, etaSeconds: progress.etaSeconds as number | null ?? null };
|
|
}
|
|
|
|
export function updateHostLibraryCaches(hostId: string, input: unknown, options: { snapshot?: boolean } = {}): void {
|
|
if (!Array.isArray(input) || input.length > 2000) throw new Error('Invalid host library report.');
|
|
const handle = database();
|
|
const now = new Date().toISOString();
|
|
handle.transaction(() => {
|
|
const reported = new Set<string>();
|
|
for (const raw of input) {
|
|
if (!raw || typeof raw !== 'object') throw new Error('Invalid library cache report.');
|
|
const id = raw.libraryVersionId ?? raw.id;
|
|
if (typeof id !== 'string') throw new Error('Invalid library version ID.');
|
|
if (reported.has(id)) throw new Error('Duplicate version in host cache report.');
|
|
reported.add(id);
|
|
const version = getLibraryVersion(id); if (!version || version.state !== 'ready') continue;
|
|
let status = raw.status ?? raw.state;
|
|
if (!['missing', 'downloading', 'verifying', 'ready', 'failed', 'outdated'].includes(status)) throw new Error('Invalid library cache state.');
|
|
const checksum = typeof raw.manifestSha256 === 'string' ? raw.manifestSha256 : null;
|
|
let error = typeof raw.error === 'string' ? raw.error.slice(0, 2000) : null;
|
|
if (['ready', 'outdated'].includes(status) && checksum !== version.manifestSha256) { status = 'failed'; error = 'Host manifest checksum does not match the published version.'; }
|
|
const bytes = raw.bytesDownloaded ?? (['ready', 'outdated'].includes(status) ? version.sizeBytes : 0);
|
|
if (!Number.isSafeInteger(bytes) || bytes < 0 || bytes > version.sizeBytes) throw new Error('Invalid library download progress.');
|
|
if (['ready', 'outdated'].includes(status) && bytes !== version.sizeBytes) { status = 'failed'; error = 'Host reported an incomplete library as verified.'; }
|
|
const prior = getHostLibraryCache(hostId, id);
|
|
const terminal = ['ready', 'outdated', 'failed'].includes(status);
|
|
// Older agents do not report timing. Keep a completed measurement on
|
|
// later inventory heartbeats, but never carry an ETA into a new phase.
|
|
let transfer = raw.transfer === undefined && terminal ? prior?.transfer ?? null : transferProgress(raw.transfer, status);
|
|
let verified = raw.bytesVerified ?? (['ready', 'outdated'].includes(status) ? version.sizeBytes : terminal ? prior?.bytesVerified ?? null : null);
|
|
if (verified !== null && (!Number.isSafeInteger(verified) || verified < 0 || verified > version.sizeBytes)) throw new Error('Invalid library verification progress.');
|
|
if (transfer && status === 'verifying' && bytes !== version.sizeBytes) throw new Error('Download must finish before verification starts.');
|
|
if (['ready', 'outdated'].includes(status) && verified !== version.sizeBytes) throw new Error('Host reported incomplete verification as ready.');
|
|
if (status === 'missing') { transfer = null; verified = null; }
|
|
if (terminal && transfer) transfer = { ...transfer, bytesPerSecond: null, etaSeconds: null };
|
|
handle.prepare(`INSERT INTO host_library_caches(host_id,version_id,status,bytes_downloaded,bytes_total,manifest_sha256,error,updated_at,bytes_verified,transfer_progress)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?) ON CONFLICT(host_id,version_id) DO UPDATE SET status=excluded.status,
|
|
bytes_downloaded=excluded.bytes_downloaded,bytes_total=excluded.bytes_total,manifest_sha256=excluded.manifest_sha256,error=excluded.error,updated_at=excluded.updated_at,
|
|
bytes_verified=excluded.bytes_verified,transfer_progress=excluded.transfer_progress`)
|
|
.run(hostId, id, status, bytes, version.sizeBytes, checksum, error, now, verified, transfer ? JSON.stringify(transfer) : null);
|
|
}
|
|
if (options.snapshot) {
|
|
const cached = handle.prepare('SELECT version_id FROM host_library_caches WHERE host_id=?').all(hostId) as { version_id: string }[];
|
|
for (const cache of cached) if (!reported.has(cache.version_id)) {
|
|
handle.prepare("UPDATE host_library_caches SET status='missing',bytes_downloaded=0,manifest_sha256=NULL,error=NULL,bytes_verified=NULL,transfer_progress=NULL,updated_at=? WHERE host_id=? AND version_id=?")
|
|
.run(now, hostId, cache.version_id);
|
|
}
|
|
}
|
|
})();
|
|
}
|
|
|
|
export function queueLibraryDownload(hostId: string, versionId: string, options: { automatic?: boolean } = {}) {
|
|
const version = getLibraryVersion(versionId);
|
|
if (!version || version.state !== 'ready') throw new Error('Choose a published library version.');
|
|
if (libraryFilesRetired(versionId)) throw new Error('These outdated game files have been removed. Upload them again to use them.');
|
|
if (database().prepare("SELECT 1 FROM ec_host_operations WHERE host_id=? AND kind='library.remove' AND json_extract(payload,'$.libraryVersionId')=? AND state IN ('pending','accepted','running','unknown')").get(hostId, versionId)) throw new Error('This host copy is being removed.');
|
|
const host = getHost(hostId);
|
|
if (!host || !host.connected || !host.credentialActive || host.mode === 'disabled') throw new Error('Choose an online, enrolled host.');
|
|
const prior = getHostLibraryCache(hostId, versionId);
|
|
if (prior?.operationId) {
|
|
const operation = getHostOperation(prior.operationId);
|
|
if (operation && ['pending', 'accepted', 'running', 'unknown'].includes(operation.state)) return operation;
|
|
}
|
|
const retainedBytes = prior?.manifestSha256 === version.manifestSha256 ? prior.bytesDownloaded : 0;
|
|
if (Number(host.inventory.storageAvailable ?? 0) < version.sizeBytes - retainedBytes + host.minimumFreeBytes) throw new Error('This host has insufficient storage for this library and its minimum free-space reserve.');
|
|
const operation = queueHostOperation(hostId, 'library.download', {
|
|
libraryVersionId: versionId, game: version.game, manifestUrl: `/api/agent/libraries/${versionId}/manifest`,
|
|
filesBaseUrl: `/api/agent/libraries/${versionId}/files`, statusUrl: `/api/agent/libraries/${versionId}/status`,
|
|
manifestSha256: version.manifestSha256, sizeBytes: version.sizeBytes,
|
|
...(options.automatic ? { automatic: true } : {})
|
|
});
|
|
updateHostLibraryCaches(hostId, [{ libraryVersionId: versionId, status: 'downloading', bytesDownloaded: retainedBytes, manifestSha256: prior?.manifestSha256, bytesVerified: 0, transfer: null }]);
|
|
database().prepare('UPDATE host_library_caches SET operation_id=? WHERE host_id=? AND version_id=?').run(operation.id, hostId, versionId);
|
|
return operation;
|
|
}
|
|
|
|
export function reconcileLibraryOperations(): void {
|
|
const handle = database();
|
|
for (const row of handle.prepare("SELECT id,operation_id FROM game_file_versions WHERE state='importing' AND operation_id IS NOT NULL").all() as { id: string; operation_id: string }[]) {
|
|
const operation = getHostOperation(row.operation_id);
|
|
if (operation?.state === 'failed') handle.prepare("UPDATE game_file_versions SET state='failed',error=? WHERE id=? AND state='importing'").run(operation.error ?? 'Host import failed.', row.id);
|
|
}
|
|
for (const cache of handle.prepare("SELECT * FROM host_library_caches WHERE status IN ('downloading','verifying') AND operation_id IS NOT NULL").all() as CacheRow[]) {
|
|
const operation = getHostOperation(cache.operation_id!);
|
|
if (operation?.state === 'failed') handle.prepare("UPDATE host_library_caches SET status='failed',error=?,updated_at=? WHERE host_id=? AND version_id=?").run(operation.error ?? 'Host download failed.', new Date().toISOString(), cache.host_id, cache.version_id);
|
|
}
|
|
}
|
|
|
|
/** Lightweight list used in launch forms; large manifests stay on the detail endpoint. */
|
|
export function libraryOverview() {
|
|
const hosts = listHosts();
|
|
return { libraries: listLibraries(true).map(library => ({ ...library, versions: library.versions.map(({ files: _files, ...version }) => ({
|
|
...version, caches: hosts.map(host => {
|
|
const cached = version.caches.find(cache => cache.hostId === host.id);
|
|
if (cached) return { ...cached, status: cached.status === 'ready' && library.latestVersionId !== version.id ? 'outdated' as const : cached.status };
|
|
return {
|
|
hostId: host.id, libraryVersionId: version.id, status: 'missing' as const, bytesDownloaded: 0,
|
|
bytesTotal: version.sizeBytes, manifestSha256: null, error: null, updatedAt: version.createdAt, operationId: null
|
|
};
|
|
})
|
|
})) })), hosts };
|
|
}
|
|
|
|
/** Any live reservation (including stopped or provisioning instances) pins its files. */
|
|
function libraryInUse(versionId: string, hostId?: string): boolean {
|
|
const handle = database();
|
|
// Do not infer an empty fleet if its schema is unavailable.
|
|
if (!handle.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='native_instances'").get()) return true;
|
|
if (handle.prepare(`SELECT 1 FROM native_instances WHERE library_id=? AND retired_at IS NULL ${hostId ? 'AND host_id=?' : ''} LIMIT 1`)
|
|
.get(versionId, ...(hostId ? [hostId] : []))) return true;
|
|
return !!handle.prepare(`SELECT 1 FROM ec_host_operations WHERE state IN ('pending','accepted','running','unknown')
|
|
AND (json_extract(payload,'$.libraryVersionId')=? OR json_extract(payload,'$.libraryId')=? OR json_extract(payload,'$.importId')=?)
|
|
${hostId ? 'AND host_id=?' : ''} LIMIT 1`).get(versionId, versionId, versionId, ...(hostId ? [hostId] : []));
|
|
}
|
|
|
|
/** Queue serialized, recoverable host cleanup; an offline host keeps its copy pinned. */
|
|
export function reconcileHostLibraryCleanup(hostId: string): void {
|
|
const handle = database();
|
|
const host = getHost(hostId);
|
|
if (!host?.connected || !host.credentialActive || host.mode === 'disabled' || host.decommission
|
|
|| !host.inventory.capabilities?.includes('library.remove')) return;
|
|
const latest = new Set(latestPublishedLibraries().map(version => version.id));
|
|
const caches = handle.prepare('SELECT * FROM host_library_caches WHERE host_id=?').all(hostId) as CacheRow[];
|
|
for (const cache of caches) {
|
|
if (latest.has(cache.version_id) || (cache.status === 'missing' && cache.bytes_downloaded === 0) || libraryInUse(cache.version_id, hostId)) continue;
|
|
const previous = handle.prepare(`SELECT state,updated_at FROM ec_host_operations WHERE host_id=? AND kind='library.remove'
|
|
AND json_extract(payload,'$.libraryVersionId')=? ORDER BY rowid DESC LIMIT 1`).get(hostId, cache.version_id) as { state: string; updated_at: string } | undefined;
|
|
// Wait for fresh inventory after completion, and back off failures.
|
|
if (previous && (Date.parse(cache.updated_at) <= Date.parse(previous.updated_at)
|
|
|| previous.state === 'failed' && Date.now() - Date.parse(previous.updated_at) < 60_000)) continue;
|
|
const version = getLibraryVersion(cache.version_id)!;
|
|
queueHostOperation(hostId, 'library.remove', { libraryVersionId: version.id, game: version.game });
|
|
}
|
|
}
|
|
|
|
const cleaning = new WeakMap<ReturnType<typeof database>, Promise<void>>();
|
|
/** Reserve deletion durably before yielding, so new downloads cannot race filesystem removal. */
|
|
export function cleanupControllerLibraries(): Promise<void> {
|
|
const handle = database();
|
|
const current = cleaning.get(handle); if (current) return current;
|
|
const task = (async () => {
|
|
const latest = new Set(latestPublishedLibraries().map(version => version.id));
|
|
const versions = handle.prepare(`SELECT v.id,v.storage_location,l.game FROM game_file_versions v
|
|
JOIN game_file_libraries l ON l.id=v.library_id LEFT JOIN game_file_cleanup c ON c.version_id=v.id
|
|
WHERE v.state='ready' AND c.removed_at IS NULL`).all() as { id: string; storage_location: string; game: string }[];
|
|
for (const version of versions) {
|
|
if (latest.has(version.id) || libraryInUse(version.id)) continue;
|
|
// Missing partial downloads still occupy disk. Require a full absence report from every host.
|
|
if (handle.prepare("SELECT 1 FROM host_library_caches WHERE version_id=? AND (status!='missing' OR bytes_downloaded>0) LIMIT 1").get(version.id)) continue;
|
|
// Only delete the controller-owned version directory, never an arbitrary stored path.
|
|
const expected = join(gameFilesRoot(), version.game, version.id);
|
|
if (!/^[a-zA-Z0-9_-]+$/.test(version.id) || resolve(version.storage_location) !== expected) continue;
|
|
handle.prepare('INSERT OR IGNORE INTO game_file_cleanup(version_id,started_at) VALUES(?,?)').run(version.id, new Date().toISOString());
|
|
try {
|
|
for (const path of [gameFilesRoot(), dirname(expected), expected]) {
|
|
const info = await lstat(path).catch(cause => { if (cause.code === 'ENOENT') return null; throw cause; });
|
|
if (info?.isSymbolicLink()) throw new Error('Refusing game-file cleanup through a symlink.');
|
|
}
|
|
await rm(expected, { recursive: true, force: true });
|
|
handle.prepare('UPDATE game_file_cleanup SET removed_at=? WHERE version_id=?').run(new Date().toISOString(), version.id);
|
|
} catch (cause) {
|
|
console.error(`Game-file cleanup will retry for ${version.id}:`, cause);
|
|
}
|
|
}
|
|
})();
|
|
cleaning.set(handle, task);
|
|
void task.finally(() => cleaning.delete(handle)).catch(() => {});
|
|
return task;
|
|
}
|