import { randomUUID } from 'node:crypto'; import { GamePrisma, type GamePrismaClient } from '@sammo-ts/infra'; import { normalizeTurnDaemonCommand } from '../turn/commandRegistry.js'; import type { TurnDaemonCommand, TurnDaemonCommandResponder, TurnDaemonCommandResult, TurnDaemonControlQueue, TurnDaemonStatus, } from './types.js'; const asJson = (value: unknown): GamePrisma.InputJsonValue => value as GamePrisma.InputJsonValue; const delay = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); export class DatabaseTurnDaemonCommandQueue implements TurnDaemonControlQueue, TurnDaemonCommandResponder { private readonly localQueue: TurnDaemonCommand[] = []; private readonly workerId = randomUUID(); private readonly leaseDurationMs = 60_000; private readonly maxAttempts = 3; constructor(private readonly db: GamePrismaClient) {} async initialize(): Promise { await this.recoverExpiredLeases(); } enqueue(command: TurnDaemonCommand): void { this.localQueue.push(command); } async drain(): Promise { const local = this.localQueue.splice(0, this.localQueue.length); const remote = await this.claimPending(); return local.concat(remote); } async waitUntil(deadlineMs: number | null): Promise { while (deadlineMs === null || Date.now() < deadlineMs) { const local = this.localQueue.shift(); if (local) { return local; } const remote = await this.claimPending(1); if (remote[0]) { return remote[0]; } const remaining = deadlineMs === null ? 100 : Math.max(1, Math.min(100, deadlineMs - Date.now())); await delay(remaining); } return null; } getDepth(): number { return this.localQueue.length; } async publishStatus(requestId: string, status: TurnDaemonStatus): Promise { await this.complete(requestId, { status }); } async publishCommandResult(requestId: string, result: TurnDaemonCommandResult): Promise { await this.complete(requestId, result); } async publishCommandError(requestId: string, error: unknown): Promise { const message = error instanceof Error ? error.message : 'Unknown command error.'; await this.db.$transaction(async (transaction) => { const event = await transaction.inputEvent.findUnique({ where: { requestId }, select: { status: true, target: true, lockedBy: true, attempts: true, }, }); if ( !event || event.target !== 'ENGINE' || event.status !== 'PROCESSING' || event.lockedBy !== this.workerId ) { return; } const terminal = event.attempts >= this.maxAttempts; await transaction.inputEvent.updateMany({ where: { requestId, target: 'ENGINE', status: 'PROCESSING', lockedBy: this.workerId, attempts: event.attempts, }, data: { status: terminal ? 'FAILED' : 'PENDING', processingAt: null, lockedBy: null, leaseUntil: null, completedAt: terminal ? new Date() : null, result: GamePrisma.DbNull, error: message, }, }); }); } private async claimPending(limit = 100): Promise { await this.recoverExpiredLeases(); return this.db.$transaction(async (transaction) => { const rows = await transaction.$queryRaw< Array<{ sequence: bigint; requestId: string; eventType: string; payload: unknown; createdAt: Date; }> >(GamePrisma.sql` SELECT "sequence", "request_id" AS "requestId", "event_type" AS "eventType", "payload", "created_at" AS "createdAt" FROM "input_event" WHERE "target" = 'ENGINE'::"InputEventTarget" AND "status" = 'PENDING'::"InputEventStatus" ORDER BY "sequence" ASC FOR UPDATE SKIP LOCKED LIMIT ${limit} `); if (rows.length === 0) { return []; } await transaction.inputEvent.updateMany({ where: { sequence: { in: rows.map((row) => row.sequence) }, target: 'ENGINE', status: 'PENDING', }, data: { status: 'PROCESSING', processingAt: new Date(), lockedBy: this.workerId, leaseUntil: new Date(Date.now() + this.leaseDurationMs), attempts: { increment: 1 }, }, }); const commands: TurnDaemonCommand[] = []; for (const row of rows) { const command = normalizeTurnDaemonCommand({ requestId: row.requestId, sentAt: row.createdAt.toISOString(), command: row.payload as TurnDaemonCommand, }); if (!command) { await transaction.inputEvent.update({ where: { sequence: row.sequence }, data: { status: 'FAILED', error: `Invalid command payload for ${row.eventType}`, completedAt: new Date(), lockedBy: null, leaseUntil: null, }, }); continue; } commands.push(command); } return commands; }); } private async complete(requestId: string, result: unknown): Promise { const completed = await this.db.inputEvent.updateMany({ where: { requestId, target: 'ENGINE', status: 'PROCESSING', lockedBy: this.workerId, }, data: { status: 'SUCCEEDED', result: asJson(result), completedAt: new Date(), error: null, lockedBy: null, leaseUntil: null, }, }); if (completed.count > 0) { return; } // Database hooks commit mutation results atomically with game state and // may already have set SUCCEEDED. Only the worker that still owns the // lease may clear that committed row's claim metadata. await this.db.inputEvent.updateMany({ where: { requestId, target: 'ENGINE', status: 'SUCCEEDED', lockedBy: this.workerId, }, data: { lockedBy: null, leaseUntil: null, }, }); } private async recoverExpiredLeases(): Promise { const now = new Date(); await this.db.inputEvent.updateMany({ where: { target: 'ENGINE', status: 'PROCESSING', leaseUntil: { lt: now }, }, data: { status: 'PENDING', processingAt: null, lockedBy: null, leaseUntil: null, }, }); } }