import { recordAgentReceipt } from '../provisioning/agent-events.js'; import { EventEmitter } from 'node:events'; import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypto'; import { isIP } from 'node:net'; import { db } from '../db/index.js'; import { ensureHostSchema } from './schema.js'; import type { Host, HostDecommission, HostDeletion, HostInventory, HostMode, HostOperation, HostOperationState } from '../../hosts.js'; export const AGENT_PROTOCOL_VERSION = 1; export const HEARTBEAT_MAX_AGE_MS = 90_000; const now = () => new Date().toISOString(); const workEvents = new EventEmitter(); workEvents.setMaxListeners(0); export function waitForHostWork(hostId: string, signal: AbortSignal, timeoutMs = 20_000): Promise { return new Promise(resolve => { const done = () => { clearTimeout(timer); workEvents.off(hostId, done); signal.removeEventListener('abort', done); resolve(); }; const timer = setTimeout(done, timeoutMs); workEvents.on(hostId, done); signal.addEventListener('abort', done, { once: true }); if (signal.aborted) done(); }); } const digest = (token: string) => createHash('sha256').update(token).digest('hex'); function database() { const handle = db(); ensureHostSchema(handle); return handle; } type HostRow = { id: string; name: string; hostname: string; mode: HostMode; port_min: number; port_max: number; bind_address: string; public_address: string; minimum_free_bytes: number; created_at: string; enrolled_at: string | null; last_heartbeat: string | null; inventory: string; credential_active: number }; const hostSql = `SELECT h.*, EXISTS(SELECT 1 FROM ec_host_credentials c WHERE c.host_id = h.id AND c.revoked_at IS NULL) AS credential_active FROM ec_hosts h`; type DecommissionRow = { host_id: string; operation_id: string; token_hash: string; stage: HostDecommission['stage']; requested_at: string; updated_at: string; cleaned_at: string | null; error: string | null }; function decommissionRow(hostId: string): DecommissionRow | undefined { return database().prepare('SELECT * FROM ec_host_decommissions WHERE host_id=?').get(hostId) as DecommissionRow | undefined; } function presentDecommission(row: DecommissionRow | undefined): HostDecommission | null { return row ? { operationId: row.operation_id, stage: row.stage, requestedAt: row.requested_at, updatedAt: row.updated_at, error: row.error, retryable: row.stage === 'failed' && getHostOperation(row.operation_id)?.state === 'failed' } : null; } function hostFromRow(row: HostRow): Host { const inventory: HostInventory = JSON.parse(row.inventory); const connected = !!row.credential_active && !!row.last_heartbeat && Date.now() - Date.parse(row.last_heartbeat) <= HEARTBEAT_MAX_AGE_MS; const reasons: string[] = []; if (!row.enrolled_at) reasons.push('Agent has not enrolled.'); else if (!row.credential_active) reasons.push('Agent credential has been revoked.'); if (!connected) reasons.push('Agent heartbeat is unavailable or stale.'); if (inventory.protocolVersion !== AGENT_PROTOCOL_VERSION) reasons.push('A compatible agent protocol is required.'); if (!inventory.dockerRunning) reasons.push('Docker is not ready.'); if (!inventory.directoriesWritable) reasons.push('Managed directories are not writable.'); if (typeof inventory.storageAvailable !== 'number' || inventory.storageAvailable < row.minimum_free_bytes) reasons.push('Insufficient available game storage.'); const infrastructureReady = reasons.length === 0; const runtimeReady = !!inventory.runtimeImages?.length; const libraryReady = !!inventory.libraries?.some(library => library.state === 'ready'); if (!runtimeReady) reasons.push('The Plutainer runtime image is not ready yet.'); if (!libraryReady) reasons.push('No verified game-file library is available yet.'); if (row.mode === 'draining') reasons.push('Host is draining; new deployments are blocked.'); if (row.mode === 'disabled') reasons.push('Host is disabled.'); const decommission = presentDecommission(decommissionRow(row.id)); if (decommission) reasons.push('Host deletion is in progress; waiting for confirmed cleanup and agent removal.'); return { id: row.id, name: row.name, hostname: row.hostname, mode: row.mode, portMin: row.port_min, portMax: row.port_max, bindAddress: row.bind_address, publicAddress: row.public_address, minimumFreeBytes: row.minimum_free_bytes, createdAt: row.created_at, lastHeartbeat: row.last_heartbeat, connected, connectionState: connected ? 'online' : 'offline', readiness: decommission ? 'deleting' : row.mode === 'draining' ? 'draining' : row.mode === 'disabled' ? 'disabled' : !row.enrolled_at ? 'preparing' : !infrastructureReady ? 'degraded' : !runtimeReady ? 'needs-runtime' : !libraryReady ? 'needs-game-files' : 'ready', readinessReasons: reasons, inventory, enrolled: !!row.enrolled_at, credentialActive: !!row.credential_active, decommission }; } export function listHosts(): Host[] { return (database().prepare(hostSql + ' WHERE h.deleted_at IS NULL ORDER BY h.name, h.id').all() as HostRow[]).map(hostFromRow); } export function getHost(id: string): Host | null { const row = database().prepare(hostSql + ' WHERE h.id = ? AND h.deleted_at IS NULL').get(id) as HostRow | undefined; return row ? hostFromRow(row) : null; } function requireHost(id: string): Host { const host = getHost(id); if (!host) throw new Error('Host was not found.'); return host; } export class HostDeletionError extends Error { constructor(message: string, readonly status: 400 | 401 | 404 | 409, readonly blockers: string[] = []) { super(message); } } /** Controller records can be removed only after all remote instance ownership is resolved. */ export function hostDeletionStatus(id: string): HostDeletion { const host = getHost(id); if (!host) throw new HostDeletionError('Host not found.', 404); if (host.decommission && !host.decommission.retryable) return { allowed: true, blockers: [] }; const handle = database(); const blockers: string[] = []; if (host.enrolled && !host.inventory.capabilities?.includes('host.decommission')) blockers.push('Update this host agent before deleting it: generate an installer from Agent access, rerun it on this host, and wait for a fresh heartbeat.'); if (host.enrolled && !host.credentialActive) blockers.push('Re-enroll this host so its agent can receive and complete the cleanup request.'); const instances = handle.prepare('SELECT server FROM native_instances WHERE host_id=? AND retired_at IS NULL ORDER BY server').all(id) as { server: string }[]; if (instances.length) { const names = instances.slice(0, 5).map(instance => instance.server).join(', ') + (instances.length > 5 ? ', …' : ''); blockers.push(`Remove the ${instances.length} server instance${instances.length === 1 ? '' : 's'} assigned to this host first (${names}). Queued and failed deployments also require removal.`); } const ports = handle.prepare('SELECT COUNT(*) AS count FROM ec_host_ports WHERE host_id=?').get(id) as { count: number }; if (ports.count && !instances.length) blockers.push(`This host still has ${ports.count} reserved game port${ports.count === 1 ? '' : 's'}. Finish or reconcile the corresponding server removals first.`); // A pending command may have reached the agent before its acknowledgement. // Only commands that have never been delivered are safe to cancel locally. const operations = handle.prepare(`SELECT kind,state FROM ec_host_operations WHERE host_id=? AND (state IN ('accepted','running','unknown') OR (state='pending' AND last_delivered_at IS NOT NULL)) ORDER BY created_at`).all(id) as { kind: string; state: string }[]; if (operations.length) { const work = operations.slice(0, 5).map(operation => `${operation.kind}: ${operation.state}`).join(', ') + (operations.length > 5 ? ', …' : ''); blockers.push(`${operations.length} delivered operation${operations.length === 1 ? ' is' : 's are'} unresolved (${work}). Let the agent reconnect and finish or reconcile this work before deleting the host.`); } return { allowed: blockers.length === 0, blockers }; } /** Keep the host visible and authenticated until its independent cleanup worker proves removal. */ export function deleteHost(id: string): { deleted: boolean; operationId?: string } { const handle = database(); return handle.transaction(() => { const host = getHost(id); if (!host) throw new HostDeletionError('Host not found.', 404); if (host.decommission && !host.decommission.retryable) return { deleted: false, operationId: host.decommission.operationId }; const status = hostDeletionStatus(id); if (!status.allowed) throw new HostDeletionError(status.blockers.join(' '), 409, status.blockers); const timestamp = now(); const reason = 'Host deletion requested before this operation was delivered to the agent.'; handle.prepare(`INSERT INTO ec_host_operation_events(operation_id,state,created_at,error) SELECT id,'failed',?,? FROM ec_host_operations WHERE host_id=? AND state='pending' AND last_delivered_at IS NULL`).run(timestamp, reason, id); handle.prepare("UPDATE ec_host_operations SET state='failed',error=?,updated_at=? WHERE host_id=? AND state='pending' AND last_delivered_at IS NULL").run(reason, timestamp, id); handle.prepare('UPDATE ec_host_enrollments SET consumed_at=? WHERE host_id=? AND consumed_at IS NULL').run(timestamp, id); if (!host.enrolled) { handle.prepare("UPDATE ec_hosts SET deleted_at=?,mode='disabled' WHERE id=? AND deleted_at IS NULL").run(timestamp, id); return { deleted: true }; } const operationId = randomUUID(); const token = 'ec_cleanup_' + randomBytes(32).toString('base64url'); const payload = { hostId: id, operationId, receiptToken: token, receiptUrl: `/api/agent/decommission/${operationId}` }; handle.prepare("INSERT INTO ec_host_operations(id,host_id,kind,payload,state,created_at,updated_at) VALUES(?,?,'host.decommission',?,'pending',?,?)").run(operationId, id, JSON.stringify(payload), timestamp, timestamp); handle.prepare("INSERT INTO ec_host_operation_events(operation_id,state,created_at) VALUES(?,'pending',?)").run(operationId, timestamp); handle.prepare(`INSERT INTO ec_host_decommissions(host_id,operation_id,token_hash,requested_at,updated_at) VALUES(?,?,?,?,?) ON CONFLICT(host_id) DO UPDATE SET operation_id=excluded.operation_id,token_hash=excluded.token_hash,stage='queued',requested_at=excluded.requested_at,updated_at=excluded.updated_at,cleaned_at=NULL,error=NULL`) .run(id, operationId, digest(token), timestamp, timestamp); handle.prepare("UPDATE ec_hosts SET mode='draining' WHERE id=?").run(id); queueMicrotask(() => workEvents.emit(id)); return { deleted: false, operationId }; }).immediate(); } export function receiveDecommissionReceipt(operationId: string, token: string, input: Record): { acknowledged: true } { if (!token || token.length > 200) throw new HostDeletionError('A valid cleanup receipt credential is required.', 401); return database().transaction(() => { const row = database().prepare('SELECT * FROM ec_host_decommissions WHERE operation_id=?').get(operationId) as DecommissionRow | undefined; if (!row || !timingSafeEqual(Buffer.from(row.token_hash, 'hex'), Buffer.from(digest(token), 'hex'))) throw new HostDeletionError('A valid cleanup receipt credential is required.', 401); if (!['cleaned', 'completed', 'failed'].includes(input.stage as string)) throw new HostDeletionError('Invalid cleanup receipt stage.', 400); if (row.stage === 'completed') return { acknowledged: true as const }; const operation = getHostOperation(operationId)!; if (!['accepted','running','unknown'].includes(operation.state)) throw new HostDeletionError('Cleanup has not been accepted by the host.', 409); const timestamp = now(); if (input.stage === 'failed') { database().prepare("UPDATE ec_host_decommissions SET stage='failed',error=?,updated_at=? WHERE host_id=?") .run(typeof input.error === 'string' ? input.error.slice(0, 2000) : 'Host cleanup needs attention. The cleanup worker will retry.', timestamp, row.host_id); return { acknowledged: true as const }; } if (input.stage === 'cleaned') { database().prepare("UPDATE ec_host_decommissions SET stage='cleaned',cleaned_at=COALESCE(cleaned_at,?),error=NULL,updated_at=? WHERE host_id=?").run(timestamp, timestamp, row.host_id); return { acknowledged: true as const }; } if (!row.cleaned_at) throw new HostDeletionError('The host must confirm file cleanup before confirming agent removal.', 409); database().prepare("UPDATE ec_host_decommissions SET stage='completed',error=NULL,updated_at=? WHERE host_id=?").run(timestamp, row.host_id); database().prepare("UPDATE ec_host_operations SET state='completed',result=?,error=NULL,updated_at=? WHERE id=?").run(JSON.stringify({ filesRemoved: true, agentRemoved: true }), timestamp, operationId); database().prepare("INSERT INTO ec_host_operation_events(operation_id,state,created_at) VALUES(?,'completed',?)").run(operationId, timestamp); database().prepare('UPDATE ec_host_credentials SET revoked_at=? WHERE host_id=? AND revoked_at IS NULL').run(timestamp, row.host_id); database().prepare('UPDATE ec_host_enrollments SET consumed_at=? WHERE host_id=? AND consumed_at IS NULL').run(timestamp, row.host_id); if (database().prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='host_library_caches'").get()) { database().prepare("UPDATE host_library_caches SET status='missing',bytes_downloaded=0,manifest_sha256=NULL,error=NULL,updated_at=? WHERE host_id=?").run(timestamp, row.host_id); } database().prepare("UPDATE ec_hosts SET deleted_at=?,mode='disabled' WHERE id=?").run(timestamp, row.host_id); return { acknowledged: true as const }; }).immediate(); } function textField(value: unknown, label: string, max = 200): string { if (typeof value !== 'string' || !value.trim() || value.length > max || /[\x00-\x1f]/.test(value)) throw new Error(`${label} must be a nonempty string up to ${max} characters.`); return value.trim(); } function port(value: unknown): number { if (typeof value !== 'number' || !Number.isInteger(value) || value < 1 || value > 65535) throw new Error('Ports must be whole numbers between 1 and 65535.'); return value; } export function createHost(input: Record): Host { const id = randomUUID(); const name = textField(input.name, 'Host name'); const min = port(input.portMin ?? 28960); const max = port(input.portMax ?? 28980); if (max < min) throw new Error('The last game port must not precede the first.'); const bind = input.bindAddress === undefined ? '0.0.0.0' : textField(input.bindAddress, 'Bind address'); if (!isIP(bind)) throw new Error('Bind address must be an IPv4 or IPv6 address.'); database().prepare('INSERT INTO ec_hosts(id,name,port_min,port_max,bind_address,created_at) VALUES(?,?,?,?,?,?)').run(id, name, min, max, bind, now()); return requireHost(id); } export function updateHost(id: string, input: Record): Host { const host = requireHost(id); if (host.decommission) throw new Error('Host cleanup is in progress; its settings cannot change.'); const name = input.name === undefined ? host.name : textField(input.name, 'Host name'); const mode = input.mode ?? host.mode; if (!['active', 'draining', 'disabled'].includes(mode as string)) throw new Error('Host mode must be active, draining, or disabled.'); const min = port(input.portMin ?? host.portMin); const max = port(input.portMax ?? host.portMax); if (max < min) throw new Error('The last game port must not precede the first.'); const bind = input.bindAddress === undefined ? host.bindAddress : textField(input.bindAddress, 'Bind address'); if (!isIP(bind)) throw new Error('Bind address must be an IPv4 or IPv6 address.'); const publicAddress = input.publicAddress === undefined ? host.publicAddress : input.publicAddress === '' ? '' : textField(input.publicAddress, 'Public address'); const minimum = input.minimumFreeBytes ?? host.minimumFreeBytes; if (typeof minimum !== 'number' || !Number.isSafeInteger(minimum) || minimum < 0) throw new Error('Minimum free storage must be a nonnegative byte count.'); database().prepare('UPDATE ec_hosts SET name=?,mode=?,port_min=?,port_max=?,bind_address=?,public_address=?,minimum_free_bytes=? WHERE id=?').run(name, mode, min, max, bind, publicAddress, minimum, id); return requireHost(id); } export function createEnrollment(hostId: string, ttlMs = 15 * 60_000): { id: string; hostId: string; token: string; expiresAt: string } { const host = requireHost(hostId); if (host.decommission && !host.decommission.retryable) throw new Error('Host cleanup is in progress; enrollment is unavailable.'); if (!Number.isInteger(ttlMs) || ttlMs <= 0 || ttlMs > 60 * 60_000) throw new Error('Enrollment lifetime must be at most one hour.'); const token = 'ec_enroll_' + randomBytes(32).toString('base64url'); const id = randomUUID(); const expiresAt = new Date(Date.now() + ttlMs).toISOString(); database().transaction(() => { database().prepare('UPDATE ec_host_enrollments SET consumed_at=? WHERE host_id=? AND consumed_at IS NULL').run(now(), hostId); database().prepare('INSERT INTO ec_host_enrollments(id,host_id,token_hash,created_at,expires_at) VALUES(?,?,?,?,?)').run(id, hostId, digest(token), now(), expiresAt); })(); return { id, hostId, token, expiresAt }; } function issueCredential(hostId: string, credential = 'ec_host_' + randomBytes(32).toString('base64url')): string { database().prepare('INSERT INTO ec_host_credentials(id,host_id,token_hash,created_at) VALUES(?,?,?,?)').run(randomUUID(), hostId, digest(credential), now()); return credential; } export function validateInventory(value: unknown): HostInventory { if (!value || typeof value !== 'object' || Array.isArray(value)) throw new Error('Host inventory must be an object.'); const inventory = value as HostInventory; if (JSON.stringify(inventory).length > 256_000) throw new Error('Host inventory is too large.'); for (const key of ['hostname','agentVersion','os','architecture','cpu','dockerVersion']) { if (inventory[key] !== undefined && (typeof inventory[key] !== 'string' || String(inventory[key]).length > 1000)) throw new Error(`Invalid inventory ${key}.`); } for (const key of ['cpuCount','memoryTotal','memoryAvailable','storageTotal','storageAvailable','protocolVersion']) { if (inventory[key] !== undefined && (typeof inventory[key] !== 'number' || !Number.isSafeInteger(inventory[key]) || Number(inventory[key]) < 0)) throw new Error(`Invalid inventory ${key}.`); } if (inventory.cpuUsagePercent !== undefined && (typeof inventory.cpuUsagePercent !== 'number' || !Number.isFinite(inventory.cpuUsagePercent) || inventory.cpuUsagePercent < 0 || inventory.cpuUsagePercent > 100)) throw new Error('Invalid inventory cpuUsagePercent.'); for (const key of ['dockerRunning','directoriesWritable']) if (inventory[key] !== undefined && typeof inventory[key] !== 'boolean') throw new Error(`Invalid inventory ${key}.`); if (inventory.runtimeImages !== undefined && (!Array.isArray(inventory.runtimeImages) || inventory.runtimeImages.some(image => typeof image !== 'string'))) throw new Error('Invalid runtime image inventory.'); if (inventory.capabilities !== undefined && (!Array.isArray(inventory.capabilities) || inventory.capabilities.length > 32 || inventory.capabilities.some(value => typeof value !== 'string' || value.length > 80))) throw new Error('Invalid agent capability inventory.'); if (inventory.agentUpdate !== undefined) { const update = inventory.agentUpdate; const version = (value: unknown) => typeof value === 'string' && value.length <= 80 && /^\d+\.\d+\.\d+$/.test(value); if (!update || typeof update !== 'object' || Array.isArray(update) || typeof update.enabled !== 'boolean' || !['current','checking','deferred','updating','updated','failed'].includes(update.state) || !version(update.currentVersion) || (update.targetVersion !== undefined && !version(update.targetVersion)) || (update.error !== undefined && (typeof update.error !== 'string' || update.error.length > 2000)) || (update.checkedAt !== undefined && (typeof update.checkedAt !== 'string' || update.checkedAt.length > 80 || !Number.isFinite(Date.parse(update.checkedAt))))) { throw new Error('Invalid agent update inventory.'); } // Only expose display fields, never the local recovery journal or secrets. inventory.agentUpdate = { enabled: update.enabled, state: update.state, currentVersion: update.currentVersion, ...(update.targetVersion === undefined ? {} : { targetVersion: update.targetVersion }), ...(update.error === undefined ? {} : { error: update.error }), ...(update.checkedAt === undefined ? {} : { checkedAt: update.checkedAt }) }; } if (inventory.libraries !== undefined && (!Array.isArray(inventory.libraries) || inventory.libraries.length > 2000 || inventory.libraries.some(library => !library || typeof library !== 'object' || typeof library.id !== 'string' || !library.id || library.id.length > 200 || !['missing','downloading','verifying','ready','failed','outdated'].includes(library.state)))) throw new Error('Invalid game library inventory.'); return inventory; } export function enrollHost(token: unknown, rawInventory: unknown): { hostId: string; credential: string; heartbeatSeconds: number } { if (typeof token !== 'string' || token.length > 200) throw new Error('Enrollment token is invalid or expired.'); const inventory = validateInventory(rawInventory); return database().transaction(() => { const enrollment = database().prepare('SELECT id,host_id FROM ec_host_enrollments WHERE token_hash=? AND consumed_at IS NULL AND expires_at>?').get(digest(token), now()) as { id: string; host_id: string } | undefined; if (!enrollment) throw new Error('Enrollment token is invalid or expired.'); const changed = database().prepare('UPDATE ec_host_enrollments SET consumed_at=? WHERE id=? AND consumed_at IS NULL').run(now(), enrollment.id); if (changed.changes !== 1) throw new Error('Enrollment token has already been consumed.'); revokeCredentials(enrollment.host_id); const credential = issueCredential(enrollment.host_id); database().prepare('UPDATE ec_hosts SET enrolled_at=?,last_heartbeat=?,hostname=?,inventory=? WHERE id=?').run(now(), now(), inventory.hostname ?? '', JSON.stringify(inventory), enrollment.host_id); return { hostId: enrollment.host_id, credential, heartbeatSeconds: 15 }; })(); } export function authenticateCredential(token: string): Host | null { if (!token || token.length > 200) return null; const row = database().prepare('SELECT id,host_id FROM ec_host_credentials WHERE token_hash=? AND revoked_at IS NULL').get(digest(token)) as { id: string; host_id: string } | undefined; if (!row) return null; database().prepare('UPDATE ec_host_credentials SET last_used_at=? WHERE id=?').run(now(), row.id); return getHost(row.host_id); } export function revokeCredentials(hostId: string): void { const host = requireHost(hostId); if (host.decommission && !host.decommission.retryable) throw new Error('Host cleanup needs its credentials until removal is confirmed.'); database().prepare('UPDATE ec_host_credentials SET revoked_at=? WHERE host_id=? AND revoked_at IS NULL').run(now(), hostId); } export function rotateCredential(hostId: string, nextCredential?: string): string { requireHost(hostId); if (nextCredential !== undefined && !/^ec_host_[A-Za-z0-9_-]{43}$/.test(nextCredential)) throw new Error('New credentials must contain 32 random bytes encoded as base64url.'); return database().transaction(() => { if (nextCredential) { const existing = database().prepare('SELECT host_id,revoked_at FROM ec_host_credentials WHERE token_hash=?').get(digest(nextCredential)) as { host_id: string; revoked_at: string | null } | undefined; if (existing) { if (existing.host_id !== hostId || existing.revoked_at) throw new Error('This credential cannot be reused.'); return nextCredential; } } revokeCredentials(hostId); return issueCredential(hostId, nextCredential); })(); } export function heartbeat(hostId: string, value: unknown): Host { requireHost(hostId); const inventory = validateInventory(value); database().prepare('UPDATE ec_hosts SET last_heartbeat=?,hostname=?,inventory=? WHERE id=?').run(now(), inventory.hostname ?? '', JSON.stringify(inventory), hostId); return requireHost(hostId); } export function requireReadyHost(hostId: string, options: { libraryId?: string; requiredBytes?: number; image?: string; port?: number; bindAddress?: string; protocol?: 'udp' | 'tcp' } = {}): Host { const host = requireHost(hostId); const reasons = [...host.readinessReasons]; if (options.requiredBytes !== undefined && (host.inventory.storageAvailable ?? 0) < options.requiredBytes + host.minimumFreeBytes) reasons.push('Insufficient storage for this deployment.'); if (options.libraryId && !host.inventory.libraries?.some(library => library.id === options.libraryId && library.state === 'ready')) reasons.push('The selected game library is not ready on this host.'); if (options.image && !host.inventory.runtimeImages?.includes(options.image)) reasons.push('The selected runtime image is not cached on this host.'); if (options.port !== undefined && !hostPortAvailable(hostId, options.port, options.bindAddress, options.protocol)) reasons.push('The selected host port is not available.'); if (reasons.length) throw new Error(reasons.join(' ')); return host; } type OperationRow = { id: string; host_id: string; kind: string; instance_id: string | null; generation: number | null; payload: string; state: HostOperationState; result: string | null; error: string | null; created_at: string; updated_at: string }; function operationFromRow(row: OperationRow): HostOperation { return { id: row.id, hostId: row.host_id, kind: row.kind, instanceId: row.instance_id, generation: row.generation, payload: JSON.parse(row.payload), state: row.state, result: row.result === null ? null : JSON.parse(row.result), error: row.error, createdAt: row.created_at, updatedAt: row.updated_at }; } export function getHostOperation(id: string): HostOperation | null { const row = database().prepare('SELECT * FROM ec_host_operations WHERE id=?').get(id) as OperationRow | undefined; return row ? operationFromRow(row) : null; } export function listHostOperations(hostId: string, limit = 100, summary = false): HostOperation[] { return (database().prepare(`SELECT ${summary ? `id,host_id,kind,instance_id,generation,state,error,created_at,updated_at,'{}' AS payload,NULL AS result` : '*'} FROM ec_host_operations WHERE host_id=? ORDER BY created_at DESC,rowid DESC LIMIT ?`).all(hostId, Math.min(500, Math.max(1, limit))) as OperationRow[]).map(operationFromRow); } /** Heartbeats can tell an executing create to stop at its next safe command boundary. */ export function pendingCreationCancellations(hostId: string): string[] { return (database().prepare(`SELECT json_extract(payload,'$.createOperationId') AS id FROM ec_host_operations WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown')`).all(hostId) as { id: string }[]).map(row => row.id); } export function queueHostOperation(hostId: string, kind: string, payload: Record, options: { instanceId?: string; generation?: number; idempotencyKey?: string; id?: string } = {}): HostOperation { const host = requireHost(hostId); if (host.decommission) throw new Error('Host cleanup is in progress; new operations are blocked.'); if (kind === 'host.decommission') throw new Error('Use the host deletion workflow to request cleanup.'); if (!/^[a-z][a-z0-9_.-]{0,79}$/.test(kind)) throw new Error('Invalid operation kind.'); if (!payload || typeof payload !== 'object' || Array.isArray(payload)) throw new Error('Operation payload must be an object.'); if (options.generation !== undefined && (!Number.isSafeInteger(options.generation) || options.generation < 1)) throw new Error('Instance generation must be a positive integer.'); const operation = database().transaction(() => { if (options.idempotencyKey) { const existing = database().prepare('SELECT * FROM ec_host_operations WHERE host_id=? AND idempotency_key=?').get(hostId, options.idempotencyKey) as OperationRow | undefined; if (existing) { if (existing.kind !== kind || existing.instance_id !== (options.instanceId ?? null) || existing.generation !== (options.generation ?? null) || existing.payload !== JSON.stringify(payload)) throw new Error('Operation idempotency key is already associated with a different request.'); return operationFromRow(existing); } } if (['instance.create','create'].includes(kind)) requireReadyHost(hostId); if (host.mode === 'disabled' && kind !== 'instance.cancel-create') throw new Error('Host is disabled. Enable it before requesting operations.'); const id = options.id ?? randomUUID(); const timestamp = now(); database().prepare('INSERT INTO ec_host_operations(id,host_id,kind,instance_id,generation,idempotency_key,payload,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?)').run(id, hostId, kind, options.instanceId ?? null, options.generation ?? null, options.idempotencyKey ?? null, JSON.stringify(payload), timestamp, timestamp); database().prepare('INSERT INTO ec_host_operation_events(operation_id,state,created_at) VALUES(?,?,?)').run(id, 'pending', timestamp); return getHostOperation(id)!; })(); queueMicrotask(() => workEvents.emit(hostId)); return operation; } export function pendingHostOperations(hostId: string, lane: 'all' | 'read' | 'write' = 'all'): HostOperation[] { const host = requireHost(hostId); if (host.mode === 'disabled' || !host.connected) return []; // Older agents hold activity.lock across the entire delivered batch, including // observations. Filtering only the unsupported cleanup can keep the updater // busy indefinitely. Drain the old batch by withholding ALL further work; // heartbeats and receipts still flow, and the new capability releases this hold. // This must live in the controller so the fix also works before an agent update. if (!host.inventory.capabilities?.includes('instance.cancel-create') && database().prepare(`SELECT 1 FROM ec_host_operations WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown') LIMIT 1`).get(hostId)) return []; // Accepted/running/unknown commands keep the same immutable ID. The agent // consults its durable journal before acknowledging or executing a replay. const readOnly = `(kind IN ('host.inspect','instance.inspect','files.read','files.list','files.stat','logs.read') OR (kind='rcon.execute' AND json_extract(payload,'$.command') IN ('status','getstatus','version')))`; const laneFilter = lane === 'all' ? '' : ` AND ${lane === 'write' ? 'NOT ' : ''}${readOnly}`; return database().transaction(() => { const rows = database().prepare(`SELECT * FROM ec_host_operations o WHERE host_id=? AND state IN ('pending','accepted','running','unknown') ${laneFilter} AND (? OR state != 'pending' OR kind NOT IN ('instance.create','create')) AND (kind!='instance.cancel-create' OR ?) AND NOT EXISTS (SELECT 1 FROM ec_host_operations cancellation WHERE cancellation.host_id=o.host_id AND cancellation.kind='instance.cancel-create' AND json_extract(cancellation.payload,'$.createOperationId')=o.id) ORDER BY CASE WHEN kind='instance.cancel-create' THEN 0 ELSE 1 END,COALESCE(last_delivered_at,''),created_at,rowid LIMIT 25`) .all(hostId, Number(host.readiness === 'ready'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false)) as OperationRow[]; const deliveredAt = now(); for (const operation of rows) database().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(deliveredAt, operation.id); return rows.map(operationFromRow); })(); } export function reportHostOperation(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperation { const operation = database().transaction(() => { requireHost(hostId); const operation = getHostOperation(id); if (!operation || operation.hostId !== hostId) throw new Error('Operation was not found for this host.'); if (operation.kind === 'host.decommission' && input.state === 'completed') throw new Error('Host removal requires the cleanup worker completion receipt.'); if (!['accepted','running','completed','failed','unknown'].includes(input.state as string)) throw new Error('Invalid operation state.'); if (['completed','failed'].includes(operation.state)) return operation; const state = input.state as HostOperationState; if (operation.state === 'running' && state === 'accepted') return operation; const error = input.error === undefined || input.error === null ? null : String(input.error).slice(0, 4000); const result = input.result === undefined ? null : JSON.stringify(input.result); if (result !== null && Buffer.byteLength(result) > 128 * 1024 * 1024) throw new Error('Operation result exceeds the limit.'); database().prepare('UPDATE ec_host_operations SET state=?,result=?,error=?,updated_at=? WHERE id=?').run(state, result, error, now(), id); if (operation.kind === 'host.decommission') { if (state === 'failed') database().prepare("UPDATE ec_host_decommissions SET stage='failed',error=?,updated_at=? WHERE operation_id=?").run(error ?? 'The agent could not start cleanup. Repair the installation, then retry deletion.', now(), id); else database().prepare("UPDATE ec_host_decommissions SET stage='running',updated_at=? WHERE operation_id=? AND stage IN ('queued','running')").run(now(), id); } if (operation.state !== state) { database().prepare('INSERT INTO ec_host_operation_events(operation_id,state,created_at,error) VALUES(?,?,?,?)').run(id, state, now(), error); if (operation.kind.startsWith('instance.') && operation.kind !== 'instance.inspect' || operation.kind === 'files.stage') { recordAgentReceipt(id, state, now()); } } return getHostOperation(id)!; })(); queueMicrotask(() => workEvents.emit(`operation:${id}`)); return operation; } export async function waitHostOperation(id: string, timeoutMs = 30_000): Promise { const deadline = Date.now() + timeoutMs; while (true) { const operation = getHostOperation(id); if (!operation) throw new Error('Host operation was not found.'); if (['completed','failed','unknown'].includes(operation.state)) return operation; if (Date.now() >= deadline) throw new Error(`Host operation ${id} is awaiting acknowledgement; reconcile this operation ID before retrying.`); await waitForHostWork(`operation:${id}`, new AbortController().signal, Math.min(1000, Math.max(1, deadline - Date.now()))); } } export function hostPortAvailable(hostId: string, candidate: number, bindAddress?: string, protocol: 'udp' | 'tcp' = 'udp', instanceId?: string): boolean { const host = requireHost(hostId); port(candidate); const bind = bindAddress ?? host.bindAddress; if (host.decommission) return false; if (candidate < host.portMin || candidate > host.portMax) return false; const rows = database().prepare('SELECT instance_id,bind_address FROM ec_host_ports WHERE host_id=? AND protocol=? AND port=?').all(hostId, protocol, candidate) as { instance_id: string; bind_address: string }[]; return !rows.some(row => row.instance_id !== instanceId && (row.bind_address === bind || ['0.0.0.0','::'].includes(bind) || ['0.0.0.0','::'].includes(row.bind_address))); } export function nextHostPort(hostId: string, additionalPorts: number[] = [], bindAddress?: string, protocol: 'udp' | 'tcp' = 'udp'): number | null { const host = requireHost(hostId); for (let candidate = host.portMin; candidate <= host.portMax; candidate++) if (!additionalPorts.includes(candidate) && hostPortAvailable(hostId, candidate, bindAddress, protocol)) return candidate; return null; } export function reserveHostPort(hostId: string, instanceId: string, options: { port?: number; bindAddress?: string; protocol?: 'udp' | 'tcp' } = {}): number { const host = requireHost(hostId); const bind = options.bindAddress ?? host.bindAddress; const protocol = options.protocol ?? 'udp'; if (host.decommission) throw new Error('Host cleanup is in progress; new port reservations are blocked.'); if (!isIP(bind)) throw new Error('Bind address must be an IP address.'); if (!['udp','tcp'].includes(protocol)) throw new Error('Port protocol must be UDP or TCP.'); if (!instanceId) throw new Error('An immutable instance ID is required.'); return database().transaction(() => { const existing = database().prepare('SELECT port FROM ec_host_ports WHERE host_id=? AND instance_id=? AND bind_address=? AND protocol=?').get(hostId, instanceId, bind, protocol) as { port: number } | undefined; if (existing && (options.port === undefined || options.port === existing.port)) return existing.port; const allocated = options.port ?? nextHostPort(hostId, [], bind, protocol); if (allocated === null || !hostPortAvailable(hostId, allocated, bind, protocol, instanceId)) throw new Error('No available game port on the selected host.'); database().prepare('INSERT INTO ec_host_ports(host_id,instance_id,bind_address,protocol,port,created_at) VALUES(?,?,?,?,?,?)').run(hostId, instanceId, bind, protocol, allocated, now()); return allocated; })(); } export function releaseHostPorts(instanceId: string): void { database().prepare('DELETE FROM ec_host_ports WHERE instance_id=?').run(instanceId); }