From 822acebca063e080edc65cd561b35deef5981e5c Mon Sep 17 00:00:00 2001 From: Hide_D Date: Sat, 3 Jan 2026 16:01:29 +0000 Subject: [PATCH] feat: update gateway admin action consumer to handle applied actions and add reset functionality --- .../src/turn/gatewayAdminActions.ts | 40 ++++- app/game-engine/src/turn/turnDaemon.ts | 169 +++++++++++++++--- 2 files changed, 177 insertions(+), 32 deletions(-) diff --git a/app/game-engine/src/turn/gatewayAdminActions.ts b/app/game-engine/src/turn/gatewayAdminActions.ts index a3665dc..5ffb865 100644 --- a/app/game-engine/src/turn/gatewayAdminActions.ts +++ b/app/game-engine/src/turn/gatewayAdminActions.ts @@ -15,7 +15,7 @@ export interface GatewayAdminActionRecord { } export interface GatewayAdminActionResult { - status: Exclude; + status: GatewayAdminActionStatus; detail?: string; } @@ -25,6 +25,10 @@ export interface GatewayAdminActionConsumerOptions { profileName: string; pollIntervalMs?: number; handler: (action: GatewayAdminActionRecord) => Promise; + onActionApplied?: ( + action: GatewayAdminActionRecord, + result: GatewayAdminActionResult + ) => Promise; } export interface GatewayAdminActionConsumer { @@ -117,22 +121,37 @@ export const createGatewayAdminActionConsumer = async ( string, { status: GatewayAdminActionStatus; detail?: string; handledAt: string } >(); + const appliedActions: Array<{ + action: GatewayAdminActionRecord; + result: GatewayAdminActionResult; + }> = []; for (const action of pending) { const key = buildActionKey(action); try { const result = await options.handler(action); - updates.set(key, { - status: result.status, - detail: result.detail, - handledAt: new Date().toISOString(), - }); + if (result.status !== 'REQUESTED') { + updates.set(key, { + status: result.status, + detail: result.detail, + handledAt: new Date().toISOString(), + }); + appliedActions.push({ action, result }); + } } catch (error) { updates.set(key, { status: 'FAILED', detail: error instanceof Error ? error.message : String(error), handledAt: new Date().toISOString(), }); + appliedActions.push({ + action, + result: { + status: 'FAILED', + detail: + error instanceof Error ? error.message : String(error), + }, + }); } } @@ -169,6 +188,12 @@ export const createGatewayAdminActionConsumer = async ( }, }, }); + + if (options.onActionApplied) { + for (const applied of appliedActions) { + await options.onActionApplied(applied.action, applied.result); + } + } } finally { inFlight = false; } @@ -190,6 +215,9 @@ export const createGatewayAdminActionConsumer = async ( clearInterval(timer); timer = null; } + while (inFlight) { + await new Promise((resolve) => setTimeout(resolve, 50)); + } await connector.disconnect(); }; diff --git a/app/game-engine/src/turn/turnDaemon.ts b/app/game-engine/src/turn/turnDaemon.ts index eb983d8..4f90d7b 100644 --- a/app/game-engine/src/turn/turnDaemon.ts +++ b/app/game-engine/src/turn/turnDaemon.ts @@ -26,6 +26,8 @@ import { createReservedTurnHandler } from './reservedTurnHandler.js'; import { createReservedTurnStore } from './reservedTurnStore.js'; import { loadTurnCommandProfile } from './turnCommandProfile.js'; import { loadTurnWorldFromDatabase } from './worldLoader.js'; +import { seedScenarioToDatabase } from '../scenario/scenarioSeeder.js'; +import { createGamePostgresConnector, createGatewayPostgresConnector } from '@sammo-ts/infra'; export interface TurnDaemonRuntimeOptions { profile: string; @@ -151,32 +153,6 @@ export const createTurnDaemonRuntime = async ( if (gatewayGate) { pauseGate = gatewayGate.shouldPause; } - if (options.profileName) { - adminActionConsumer = await createGatewayAdminActionConsumer({ - databaseUrl: options.databaseUrl, - gatewayDatabaseUrl: options.gatewayDatabaseUrl, - profileName: options.profileName, - pollIntervalMs: options.adminActionIntervalMs, - handler: async (action) => { - const reason = action.reason ?? `admin:${action.action ?? 'action'}`; - switch (action.action) { - case 'RESUME': - controlQueue.enqueue({ type: 'resume', reason }); - return { status: 'APPLIED', detail: 'resume queued' }; - case 'PAUSE': - controlQueue.enqueue({ type: 'pause', reason }); - return { status: 'APPLIED', detail: 'pause queued' }; - case 'STOP': - case 'SHUTDOWN': - controlQueue.enqueue({ type: 'shutdown', reason }); - return { status: 'APPLIED', detail: 'shutdown queued' }; - default: - return { status: 'IGNORED', detail: 'not implemented' }; - } - }, - }); - adminActionConsumer.start(); - } if (options.enableDatabaseFlush ?? true) { const dbHooks = await createDatabaseTurnHooks(options.databaseUrl, world, { reservedTurns: reservedTurnStoreHandle?.store, @@ -243,6 +219,147 @@ export const createTurnDaemonRuntime = async ( { profile: options.profile, defaultBudget } ); + const resolveScenarioId = async (): Promise => { + const meta = world.getState().meta as Record; + const raw = meta.scenarioId; + if (typeof raw === 'number' && Number.isFinite(raw)) { + return Math.floor(raw); + } + if (typeof raw === 'string') { + const parsed = Number(raw); + if (Number.isFinite(parsed)) { + return Math.floor(parsed); + } + } + const connector = createGamePostgresConnector({ url: options.databaseUrl }); + await connector.connect(); + try { + const prisma = connector.prisma as unknown as { + worldState: { findFirst: (args: unknown) => Promise<{ scenarioCode: string } | null> }; + }; + const row = await prisma.worldState.findFirst({ + select: { scenarioCode: true }, + }); + if (!row) { + return null; + } + const parsed = Number(row.scenarioCode); + return Number.isFinite(parsed) ? Math.floor(parsed) : null; + } finally { + await connector.disconnect(); + } + }; + + const updateGatewayProfileStatus = async ( + nextStatus: 'RUNNING' | 'STOPPED' + ): Promise<{ previous?: string; updated: boolean }> => { + const connector = createGatewayPostgresConnector({ + url: options.gatewayDatabaseUrl ?? options.databaseUrl, + }); + await connector.connect(); + try { + const prisma = connector.prisma as unknown as { + gatewayProfile: { + findUnique: (args: unknown) => Promise<{ status: string } | null>; + update: (args: unknown) => Promise; + }; + }; + const profile = await prisma.gatewayProfile.findUnique({ + where: { profileName: options.profileName }, + select: { status: true }, + }); + if (!profile) { + return { updated: false }; + } + if (profile.status === 'DISABLED') { + return { previous: profile.status, updated: false }; + } + await prisma.gatewayProfile.update({ + where: { profileName: options.profileName }, + data: { + status: nextStatus, + lastError: null, + }, + }); + return { previous: profile.status, updated: true }; + } finally { + await connector.disconnect(); + } + }; + + let resetInFlight = false; + + if (options.profileName) { + adminActionConsumer = await createGatewayAdminActionConsumer({ + databaseUrl: options.databaseUrl, + gatewayDatabaseUrl: options.gatewayDatabaseUrl, + profileName: options.profileName, + pollIntervalMs: options.adminActionIntervalMs, + handler: async (action) => { + const reason = action.reason ?? `admin:${action.action ?? 'action'}`; + if (action.action === 'RESET_NOW' || action.action === 'RESET_SCHEDULED') { + if (resetInFlight) { + return { status: 'IGNORED', detail: 'reset already in progress' }; + } + if (action.action === 'RESET_SCHEDULED') { + if (!action.scheduledAt) { + return { status: 'FAILED', detail: 'scheduledAt is required' }; + } + const scheduledAt = new Date(action.scheduledAt); + if (Number.isNaN(scheduledAt.getTime())) { + return { status: 'FAILED', detail: 'scheduledAt is invalid' }; + } + if (scheduledAt.getTime() > Date.now()) { + return { status: 'REQUESTED', detail: 'waiting for schedule' }; + } + } + resetInFlight = true; + try { + await lifecycle.stop('admin reset'); + const scenarioId = await resolveScenarioId(); + if (!scenarioId) { + return { status: 'FAILED', detail: 'scenarioId is missing' }; + } + const seedTime = + action.scheduledAt && action.action === 'RESET_SCHEDULED' + ? new Date(action.scheduledAt) + : new Date(); + await seedScenarioToDatabase({ + databaseUrl: options.databaseUrl, + scenarioId, + tickSeconds: world.getState().tickSeconds, + now: seedTime, + }); + const statusUpdate = await updateGatewayProfileStatus('RUNNING'); + return { + status: 'APPLIED', + detail: statusUpdate.updated + ? `seeded scenario ${scenarioId}; status RUNNING` + : `seeded scenario ${scenarioId}; status unchanged`, + }; + } finally { + resetInFlight = false; + } + } + switch (action.action) { + case 'RESUME': + controlQueue.enqueue({ type: 'resume', reason }); + return { status: 'APPLIED', detail: 'resume queued' }; + case 'PAUSE': + controlQueue.enqueue({ type: 'pause', reason }); + return { status: 'APPLIED', detail: 'pause queued' }; + case 'STOP': + case 'SHUTDOWN': + controlQueue.enqueue({ type: 'shutdown', reason }); + return { status: 'APPLIED', detail: 'shutdown queued' }; + default: + return { status: 'IGNORED', detail: 'not implemented' }; + } + }, + }); + adminActionConsumer.start(); + } + return { lifecycle, world,