diff --git a/app/game-engine/src/turn/cli.ts b/app/game-engine/src/turn/cli.ts index 89b32af..5672aab 100644 --- a/app/game-engine/src/turn/cli.ts +++ b/app/game-engine/src/turn/cli.ts @@ -14,6 +14,7 @@ export interface TurnDaemonCliOptions { schedule?: TurnSchedule; budget?: Partial; enableDatabaseFlush?: boolean; + adminActionIntervalMs?: number; env?: NodeJS.ProcessEnv; } @@ -97,6 +98,8 @@ export const runTurnDaemonCli = async ( parseBoolean(env.TURN_FLUSH_DB) ?? true; const pauseGateIntervalMs = parseNumber(env.TURN_PAUSE_GATE_MS); + const adminActionIntervalMs = + options.adminActionIntervalMs ?? parseNumber(env.TURN_ADMIN_ACTION_MS); const runtime = await createTurnDaemonRuntime({ profile, @@ -108,6 +111,7 @@ export const runTurnDaemonCli = async ( schedule: options.schedule, enableDatabaseFlush, pauseGateIntervalMs, + adminActionIntervalMs, }); let closed = false; diff --git a/app/game-engine/src/turn/gatewayAdminActions.ts b/app/game-engine/src/turn/gatewayAdminActions.ts new file mode 100644 index 0000000..a3665dc --- /dev/null +++ b/app/game-engine/src/turn/gatewayAdminActions.ts @@ -0,0 +1,200 @@ +import { createGatewayPostgresConnector } from '@sammo-ts/infra'; + +export type GatewayAdminActionStatus = 'REQUESTED' | 'APPLIED' | 'FAILED' | 'IGNORED'; + +export interface GatewayAdminActionRecord { + action?: string; + requestedAt?: string; + durationMinutes?: number | null; + scheduledAt?: string | null; + reason?: string | null; + status?: GatewayAdminActionStatus | string | null; + handledAt?: string | null; + handler?: string | null; + detail?: string | null; +} + +export interface GatewayAdminActionResult { + status: Exclude; + detail?: string; +} + +export interface GatewayAdminActionConsumerOptions { + databaseUrl: string; + gatewayDatabaseUrl?: string; + profileName: string; + pollIntervalMs?: number; + handler: (action: GatewayAdminActionRecord) => Promise; +} + +export interface GatewayAdminActionConsumer { + start(): void; + stop(): Promise; +} + +type GatewayProfileRow = { + meta: unknown; +}; + +type GatewayProfileClient = { + findUnique(args: unknown): Promise; + update(args: unknown): Promise; +}; + +const DEFAULT_POLL_MS = 5000; + +const isRecord = (value: unknown): value is Record => + Boolean(value) && typeof value === 'object' && !Array.isArray(value); + +const normalizeMeta = (value: unknown): Record => + isRecord(value) ? value : {}; + +const normalizeStatus = (value: unknown): GatewayAdminActionStatus | null => { + if (typeof value === 'string') { + return value as GatewayAdminActionStatus; + } + return null; +}; + +const buildActionKey = (action: GatewayAdminActionRecord): string => + [ + action.action ?? '', + action.requestedAt ?? '', + action.scheduledAt ?? '', + action.reason ?? '', + ].join('|'); + +export const createGatewayAdminActionConsumer = async ( + options: GatewayAdminActionConsumerOptions +): Promise => { + const connector = createGatewayPostgresConnector({ + url: options.gatewayDatabaseUrl ?? options.databaseUrl, + }); + await connector.connect(); + const prisma = connector.prisma as unknown as { + gatewayProfile: GatewayProfileClient; + }; + + let timer: NodeJS.Timeout | null = null; + let inFlight = false; + + const pollOnce = async (): Promise => { + if (inFlight) { + return; + } + inFlight = true; + try { + const profile = await prisma.gatewayProfile.findUnique({ + where: { profileName: options.profileName }, + }); + if (!profile) { + return; + } + const meta = normalizeMeta(profile.meta); + const rawActions = Array.isArray(meta.adminActions) + ? meta.adminActions + : []; + if (!rawActions.length) { + return; + } + + const pending = rawActions.filter((entry): entry is GatewayAdminActionRecord => { + if (!isRecord(entry)) { + return false; + } + if (!entry.action || typeof entry.action !== 'string') { + return false; + } + const status = normalizeStatus(entry.status) ?? 'REQUESTED'; + return status === 'REQUESTED'; + }); + + if (!pending.length) { + return; + } + + const updates = new Map< + string, + { status: GatewayAdminActionStatus; detail?: string; handledAt: string } + >(); + + 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(), + }); + } catch (error) { + updates.set(key, { + status: 'FAILED', + detail: error instanceof Error ? error.message : String(error), + handledAt: new Date().toISOString(), + }); + } + } + + if (!updates.size) { + return; + } + + const nextActions = rawActions.map((entry) => { + if (!isRecord(entry)) { + return entry; + } + const action = entry as GatewayAdminActionRecord; + const key = buildActionKey(action); + const update = updates.get(key); + if (!update) { + return entry; + } + return { + ...action, + status: update.status, + handledAt: update.handledAt, + handler: action.handler ?? 'turn-daemon', + detail: update.detail ?? action.detail ?? null, + }; + }); + + await prisma.gatewayProfile.update({ + where: { profileName: options.profileName }, + data: { + meta: { + ...meta, + adminActions: nextActions, + adminActionsUpdatedAt: new Date().toISOString(), + }, + }, + }); + } finally { + inFlight = false; + } + }; + + const start = (): void => { + if (timer) { + return; + } + timer = setInterval( + () => void pollOnce(), + options.pollIntervalMs ?? DEFAULT_POLL_MS + ); + void pollOnce(); + }; + + const stop = async (): Promise => { + if (timer) { + clearInterval(timer); + timer = null; + } + await connector.disconnect(); + }; + + return { + start, + stop, + }; +}; diff --git a/app/game-engine/src/turn/turnDaemon.ts b/app/game-engine/src/turn/turnDaemon.ts index f36a7bf..eb983d8 100644 --- a/app/game-engine/src/turn/turnDaemon.ts +++ b/app/game-engine/src/turn/turnDaemon.ts @@ -20,6 +20,7 @@ import type { import { InMemoryTurnWorld } from './inMemoryWorld.js'; import { InMemoryTurnProcessor } from './inMemoryTurnProcessor.js'; import { InMemoryTurnStateStore } from './inMemoryStateStore.js'; +import { createGatewayAdminActionConsumer } from './gatewayAdminActions.js'; import { createGatewayProfileGate } from './gatewayProfileGate.js'; import { createReservedTurnHandler } from './reservedTurnHandler.js'; import { createReservedTurnStore } from './reservedTurnStore.js'; @@ -43,6 +44,7 @@ export interface TurnDaemonRuntimeOptions { pauseGateIntervalMs?: number; commandProfile?: TurnCommandProfile; commandProfilePath?: string; + adminActionIntervalMs?: number; } export interface TurnDaemonRuntime { @@ -134,6 +136,9 @@ export const createTurnDaemonRuntime = async ( let hooks: TurnDaemonHooks | undefined; let close = async () => {}; let pauseGate: (() => Promise) | undefined; + let adminActionConsumer: Awaited< + ReturnType + > | null = null; const gatewayGate = options.profileName ? await createGatewayProfileGate({ @@ -146,6 +151,32 @@ 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, @@ -163,6 +194,7 @@ export const createTurnDaemonRuntime = async ( await reservedTurnStoreHandle.close(); } await gatewayGate?.close(); + await adminActionConsumer?.stop(); }; } else if (reservedTurnStoreHandle) { hooks = { @@ -173,6 +205,7 @@ export const createTurnDaemonRuntime = async ( close = async () => { await reservedTurnStoreHandle.close(); await gatewayGate?.close(); + await adminActionConsumer?.stop(); }; } else if (gatewayGate) { hooks = { @@ -182,6 +215,11 @@ export const createTurnDaemonRuntime = async ( }; close = async () => { await gatewayGate.close(); + await adminActionConsumer?.stop(); + }; + } else if (adminActionConsumer) { + close = async () => { + await adminActionConsumer?.stop(); }; }