diff --git a/app/gateway-api/src/adminRouter.ts b/app/gateway-api/src/adminRouter.ts index e76d3a9..b5268a0 100644 --- a/app/gateway-api/src/adminRouter.ts +++ b/app/gateway-api/src/adminRouter.ts @@ -1,10 +1,9 @@ import { randomBytes } from 'node:crypto'; -import path from 'node:path'; import { TRPCError } from '@trpc/server'; import { z } from 'zod'; -import { createGamePostgresConnector, resolvePostgresConfigFromEnv, type GatewayPrisma } from '@sammo-ts/infra'; +import type { GatewayPrisma } from '@sammo-ts/infra'; import { procedure, router } from './trpc.js'; import { listScenarioPreviews, resolveGitBranchCommitSha, resolveGitCommitSha } from './scenario/scenarioCatalog.js'; @@ -13,7 +12,6 @@ import { toPublicUser } from './auth/userRepository.js'; import type { AdminAuthContext } from './adminAuth.js'; import type { GatewayApiContext } from './context.js'; import { GATEWAY_BUILD_STATUSES, GATEWAY_PROFILE_STATUSES } from './orchestrator/profileRepository.js'; -import { seedProfileDatabase } from './orchestrator/seedProfileDatabase.js'; const zProfileStatus = z.enum(GATEWAY_PROFILE_STATUSES); const zBuildStatus = z.enum(GATEWAY_BUILD_STATUSES); @@ -369,14 +367,6 @@ const applySanctionsPatch = (current: UserSanctions, patch: SanctionsPatch): Use const buildAdminPassword = (): string => randomBytes(6).toString('hex'); -const buildServerId = (profileName: string, now: Date): string => { - const year = String(now.getFullYear()).slice(-2); - const month = String(now.getMonth() + 1).padStart(2, '0'); - const day = String(now.getDate()).padStart(2, '0'); - const suffix = randomBytes(2).toString('hex'); - return `${profileName}_${year}${month}${day}_${suffix}`; -}; - // 프로필 메타를 안전하게 읽고, 패치를 병합한다. const readMetaObject = (value: unknown): Record => { if (!value || typeof value !== 'object') { @@ -385,20 +375,6 @@ const readMetaObject = (value: unknown): Record => { return value as Record; }; -const readMetaNumber = (meta: Record, key: string): number | null => { - const raw = meta[key]; - 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); - } - } - return null; -}; - const applyMetaPatch = ( meta: Record, patch: Record @@ -870,12 +846,23 @@ export const adminRouter = router({ profiles: router({ list: adminProcedure.query(async ({ ctx }) => { const profiles = await ctx.profiles.listProfiles(); - const runtimeActions = await ctx.prisma.gatewayRuntimeAction.findMany({ - where: { - profileName: { in: profiles.map((profile) => profile.profileName) }, - }, - orderBy: { createdAt: 'desc' }, - }); + const profileNames = profiles.map((profile) => profile.profileName); + const [runtimeActions, activeOperations] = await Promise.all([ + ctx.prisma.gatewayRuntimeAction.findMany({ + where: { profileName: { in: profileNames } }, + orderBy: { createdAt: 'desc' }, + }), + ctx.prisma.gatewayOperation.findMany({ + where: { + profileName: { in: profileNames }, + status: { in: ['QUEUED', 'RUNNING'] }, + }, + select: { id: true, profileName: true, status: true }, + }), + ]); + const activeOperationByProfile = new Map( + activeOperations.map((operation) => [operation.profileName, operation]) + ); const runtimeActionsByProfile = new Map(); for (const action of runtimeActions) { const bucket = runtimeActionsByProfile.get(action.profileName) ?? []; @@ -891,6 +878,7 @@ export const adminRouter = router({ return profiles.map((profile) => ({ ...profile, runtimeActions: runtimeActionsByProfile.get(profile.profileName) ?? [], + activeOperation: activeOperationByProfile.get(profile.profileName) ?? null, runtime: runtimeMap.get(profile.profileName) ?? { profileName: profile.profileName, apiRunning: false, @@ -1073,40 +1061,24 @@ export const adminRouter = router({ } } - const scenarioValue = String(input.install.scenarioId); - if (profile.scenario !== scenarioValue) { - try { - await ctx.profiles.updateScenario(profile.profileName, scenarioValue); - } catch (error) { - throw new TRPCError({ - code: 'CONFLICT', - message: 'Scenario update failed due to duplication.', - }); - } - } - const gitRef = input.install.gitRef?.trim(); - if (gitRef) { - let resolvedCommitSha: string; - try { - resolvedCommitSha = await resolveGitCommitSha(gitRef); - } catch (error) { - throw new TRPCError({ - code: 'BAD_REQUEST', - message: 'git ref is invalid or unavailable.', - }); + const requestedRef = gitRef || profile.buildCommitSha || 'HEAD'; + let resolvedCommitSha: string; + try { + resolvedCommitSha = await resolveGitCommitSha(requestedRef); + const scenarios = await listScenarioPreviews({ gitRef: resolvedCommitSha }); + if (!scenarios.some((scenario) => scenario.id === input.install.scenarioId)) { + throw new Error('Scenario not found at source.'); } - await ctx.profiles.updateBuildStatus(profile.profileName, profile.buildStatus, { - commitSha: resolvedCommitSha, + } catch { + throw new TRPCError({ + code: 'BAD_REQUEST', + message: 'git ref is invalid or does not contain the scenario.', }); } const scheduledAt = openAt ? (preopenAt ?? openAt).toISOString() : null; const action = scheduledAt ? 'RESET_SCHEDULED' : 'RESET_NOW'; - const meta = readMetaObject(profile.meta); - const actionLog = Array.isArray(meta.adminActions) - ? meta.adminActions.filter((entry) => entry && typeof entry === 'object') - : []; const actionRecord = { action, requestedAt: now.toISOString(), @@ -1117,7 +1089,7 @@ export const adminRouter = router({ ...input.install, openAt: input.install.openAt ?? null, preopenAt: input.install.preopenAt ?? null, - gitRef: gitRef ?? null, + gitRef: resolvedCommitSha, autorunUser: autorunUser ? { limitMinutes: autorunUser.limitMinutes, @@ -1131,31 +1103,27 @@ export const adminRouter = router({ }, }, }; - - const nextMeta = { - ...meta, - adminActions: [...actionLog, actionRecord], - install: actionRecord.install, - installUpdatedAt: now.toISOString(), - }; - - await ctx.profiles.updateMeta(input.profileName, nextMeta); - - if (openAt) { - await ctx.profiles.updateStatus(profile.profileName, profile.status, { - preopenAt: preopenAt ? preopenAt.toISOString() : openAt.toISOString(), - openAt: openAt.toISOString(), - scheduledStartAt: scheduledAt, + try { + const operation = await ctx.profiles.createOperation({ + profileName: input.profileName, + type: 'RESET', + sourceMode: 'COMMIT', + sourceRef: resolvedCommitSha, + payload: { install: actionRecord.install } as GatewayPrisma.JsonObject, + reason: input.reason, + requestedBy: adminAuth.user.id, + scheduledAt: scheduledAt ?? undefined, }); - } else { - await ctx.profiles.updateStatus(profile.profileName, profile.status, { - preopenAt: null, - openAt: null, - scheduledStartAt: null, + return { ok: true, operationId: operation.id, action: actionRecord }; + } catch (error) { + if (!isUniqueConstraintError(error)) { + throw error; + } + throw new TRPCError({ + code: 'CONFLICT', + message: 'This profile already has a queued or running operation.', }); } - - return { ok: true, action: actionRecord }; }), installNow: profileAdminProcedure .input( @@ -1175,82 +1143,69 @@ export const adminRouter = router({ }); } - const scenarioValue = String(input.install.scenarioId); - let updatedProfile = profile; - if (profile.scenario !== scenarioValue) { - const updated = await ctx.profiles.updateScenario(profile.profileName, scenarioValue); - if (updated) { - updatedProfile = updated; - } - } - - const databaseUrl = resolvePostgresConfigFromEnv({ - env: process.env, - schema: updatedProfile.profile, - }).url; - const resourcesRoot = path.resolve(process.cwd(), 'resources'); - const profileMeta = readMetaObject(updatedProfile.meta); - const nextSeasonIdx = readMetaNumber(profileMeta, 'nextSeasonIdx'); - let baseSeason: number | null = null; - const connector = createGamePostgresConnector({ url: databaseUrl }); + const requestedRef = input.install.gitRef?.trim() || profile.buildCommitSha || 'HEAD'; + let resolvedCommitSha: string; try { - await connector.connect(); - const row = await connector.prisma.worldState.findFirst({ - select: { meta: true }, - }); - if (row) { - baseSeason = readMetaNumber(readMetaObject(row.meta), 'season'); + resolvedCommitSha = await resolveGitCommitSha(requestedRef); + const scenarios = await listScenarioPreviews({ gitRef: resolvedCommitSha }); + if (!scenarios.some((scenario) => scenario.id === input.install.scenarioId)) { + throw new Error('Scenario not found at source.'); } - } finally { - await connector.disconnect(); + } catch { + throw new TRPCError({ + code: 'BAD_REQUEST', + message: 'git ref is invalid or does not contain the scenario.', + }); } - const season = nextSeasonIdx ?? baseSeason ?? 1; - const seedNow = new Date(); - const serverId = buildServerId(updatedProfile.profileName, seedNow); - await seedProfileDatabase({ - databaseUrl, - scenarioId: input.install.scenarioId, - tickSeconds: input.install.turnTermMinutes * 60, - now: seedNow, - installOptions: { - turnTermMinutes: input.install.turnTermMinutes, - sync: input.install.sync, - fiction: input.install.fiction, - extend: input.install.extend, - blockGeneralCreate: input.install.blockGeneralCreate, - npcMode: input.install.npcMode, - showImgLevel: input.install.showImgLevel, - tournamentTrig: input.install.tournamentTrig, - joinMode: input.install.joinMode, - season, - serverId, - autorunUser: input.install.autorunUser - ? { - limitMinutes: input.install.autorunUser.limitMinutes, - options: Object.fromEntries( - input.install.autorunUser.options.map((option) => [option, true]) - ), - } - : null, - }, - scenarioOptions: { scenarioRoot: path.join(resourcesRoot, 'scenario') }, - mapOptions: { mapRoot: path.join(resourcesRoot, 'map') }, - unitSetOptions: { unitSetRoot: path.join(resourcesRoot, 'unitset') }, - adminUser: { - id: adminAuth.user.id, - username: adminAuth.user.username, - displayName: adminAuth.user.displayName, - }, - }); - - await ctx.profiles.updateStatus(updatedProfile.profileName, 'RUNNING', { - preopenAt: null, - openAt: null, - scheduledStartAt: null, - }); - - return { ok: true }; + try { + const operation = await ctx.profiles.createOperation({ + profileName: input.profileName, + type: 'RESET', + sourceMode: 'COMMIT', + sourceRef: resolvedCommitSha, + payload: { + install: { + ...input.install, + gitRef: resolvedCommitSha, + adminUser: { + id: adminAuth.user.id, + username: adminAuth.user.username, + displayName: adminAuth.user.displayName, + }, + }, + } as GatewayPrisma.JsonObject, + reason: input.reason, + requestedBy: adminAuth.user.id, + }); + const deadline = Date.now() + 10 * 60_000; + while (Date.now() < deadline) { + await ctx.orchestrator.runOperationsNow(); + const current = await ctx.profiles.getOperation(operation.id); + if (current?.status === 'SUCCEEDED') { + return { ok: true, operationId: operation.id }; + } + if (current?.status === 'FAILED' || current?.status === 'CANCELLED') { + throw new TRPCError({ + code: 'INTERNAL_SERVER_ERROR', + message: current.error ?? 'Profile install operation failed.', + }); + } + await new Promise((resolve) => setTimeout(resolve, 250)); + } + throw new TRPCError({ + code: 'TIMEOUT', + message: 'Profile install operation did not complete in time.', + }); + } catch (error) { + if (!isUniqueConstraintError(error)) { + throw error; + } + throw new TRPCError({ + code: 'CONFLICT', + message: 'This profile already has a queued or running operation.', + }); + } }), requestAction: adminProcedure .input( diff --git a/app/gateway-api/src/index.ts b/app/gateway-api/src/index.ts index ab4388e..4abc427 100644 --- a/app/gateway-api/src/index.ts +++ b/app/gateway-api/src/index.ts @@ -3,6 +3,7 @@ import { fileURLToPath } from 'node:url'; import { runGatewayApiServer } from './server.js'; import { runGatewayOrchestrator } from './orchestrator/orchestratorServer.js'; +import { runProfileSeedCli } from './orchestrator/profileSeedCli.js'; export * from './config.js'; export * from './context.js'; @@ -35,9 +36,15 @@ const isMain = (): boolean => { if (isMain()) { const role = process.env.GATEWAY_ROLE ?? 'api'; - const run = role === 'orchestrator' ? runGatewayOrchestrator : runGatewayApiServer; + const run = + role === 'orchestrator' + ? runGatewayOrchestrator + : role === 'profile-seed' + ? runProfileSeedCli + : runGatewayApiServer; run().catch((error) => { - const prefix = role === 'orchestrator' ? 'gateway-orchestrator' : 'gateway-api'; + const prefix = + role === 'orchestrator' ? 'gateway-orchestrator' : role === 'profile-seed' ? 'profile-seed' : 'gateway-api'; console.error(`[${prefix}] failed to start`, error); process.exitCode = 1; }); diff --git a/app/gateway-api/src/orchestrator/gatewayOrchestrator.ts b/app/gateway-api/src/orchestrator/gatewayOrchestrator.ts index 5f21e9f..c559349 100644 --- a/app/gateway-api/src/orchestrator/gatewayOrchestrator.ts +++ b/app/gateway-api/src/orchestrator/gatewayOrchestrator.ts @@ -1,5 +1,7 @@ +import fs from 'node:fs/promises'; +import os from 'node:os'; import path from 'node:path'; -import { randomBytes } from 'node:crypto'; +import { createHash, randomBytes, randomUUID } from 'node:crypto'; import { type ScenarioInstallOptions } from '@sammo-ts/game-engine'; import { createGamePostgresConnector, resolvePostgresConfigFromEnv } from '@sammo-ts/infra'; @@ -8,13 +10,14 @@ import { isRecord } from '@sammo-ts/common'; import type { BuildCommand, BuildRunner } from './buildRunner.js'; import type { ProcessManager } from './processManager.js'; import type { + GatewayClaimedProfileUpdate, GatewayOperationRecord, GatewayProfileRecord, GatewayProfileRepository, GatewayProfileStatus, } from './profileRepository.js'; import type { GitWorkspaceManager } from './workspaceManager.js'; -import { seedProfileDatabase, type AdminSeedUser } from './seedProfileDatabase.js'; +import type { AdminSeedUser } from './seedProfileDatabase.js'; export interface GatewayProcessConfig { workspaceRoot: string; @@ -102,6 +105,7 @@ interface GatewayAdminActionRecord { handledAt?: string | null; handler?: string | null; detail?: string | null; + installOperationId?: string; install?: { scenarioId?: number; turnTermMinutes?: number; @@ -133,13 +137,20 @@ interface GatewayAdminActionResult { detail?: string; } +const OPERATION_LEASE_DURATION_MS = 10 * 60_000; +const OPERATION_HEARTBEAT_INTERVAL_MS = 60_000; + +class OperationLeaseLostError extends Error {} + const normalizeMeta = (value: unknown): Record => (isRecord(value) ? value : {}); -const buildServerId = (profileName: string, now: Date): string => { +const buildServerId = (profileName: string, now: Date, installOperationId?: string): string => { const year = String(now.getFullYear()).slice(-2); const month = String(now.getMonth() + 1).padStart(2, '0'); const day = String(now.getDate()).padStart(2, '0'); - const suffix = randomBytes(2).toString('hex'); + const suffix = installOperationId + ? createHash('sha256').update(installOperationId).digest('hex').slice(0, 16) + : randomBytes(2).toString('hex'); return `${profileName}_${year}${month}${day}_${suffix}`; }; @@ -276,6 +287,7 @@ const parseInstallOptions = ( joinMode: joinMode === 'full' || joinMode === 'onlyRandom' ? joinMode : undefined, autorunUser: autorunUser ?? null, preopenAt: preopenAt ?? null, + installOperationId: action.installOperationId, }; return { @@ -424,6 +436,7 @@ export const buildWorkspaceCommands = ( ['@sammo-ts/logic', 'build'], ['@sammo-ts/game-api', 'build'], ['@sammo-ts/game-engine', 'build'], + ['@sammo-ts/gateway-api', 'build'], ]; for (const [filter, script] of buildSteps) { commands.push({ @@ -436,6 +449,17 @@ export const buildWorkspaceCommands = ( return commands; }; +export const buildProfileMigrationCommand = ( + workspaceRoot: string, + profileDatabaseUrl: string, + env?: Record +): BuildCommand => ({ + command: 'pnpm', + args: ['--filter', '@sammo-ts/infra', 'prisma:migrate:deploy:game'], + cwd: workspaceRoot, + env: { ...(env ?? {}), DATABASE_URL: profileDatabaseUrl }, +}); + const mapRuntimeStates = (profileNames: string[], processNames: Map): ProfileRuntimeSnapshot[] => profileNames.map((profileName) => { const apiName = buildProcessName(profileName, 'api'); @@ -474,6 +498,10 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { private adminActionInFlight = false; private operationInFlight = false; private readonly resetInFlight = new Set(); + private readonly operationLeaseOwner = randomUUID(); + private readonly inFlightTasks = new Set>(); + private stopping = false; + private stopPromise?: Promise; constructor(options: GatewayOrchestratorOptions) { this.repository = options.repository; @@ -489,19 +517,29 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { } start(): void { - void this.reconcileNow(); - void this.runOperationsNow(); - void this.runAdminActionsNow(); - this.reconcileTimer = setInterval(() => void this.reconcileNow(), this.reconcileIntervalMs); - this.scheduleTimer = setInterval(() => void this.runScheduleNow(), this.scheduleIntervalMs); - this.buildTimer = setInterval(() => void this.runBuildQueueNow(), this.buildIntervalMs); + this.stopping = false; + this.trackTask(this.reconcileNow()); + this.trackTask(this.runOperationsNow()); + this.trackTask(this.runAdminActionsNow()); + this.reconcileTimer = setInterval(() => this.trackTask(this.reconcileNow()), this.reconcileIntervalMs); + this.scheduleTimer = setInterval(() => this.trackTask(this.runScheduleNow()), this.scheduleIntervalMs); + this.buildTimer = setInterval(() => this.trackTask(this.runBuildQueueNow()), this.buildIntervalMs); this.adminActionTimer = setInterval(() => { - void this.runOperationsNow(); - void this.runAdminActionsNow(); + this.trackTask(this.runOperationsNow()); + this.trackTask(this.runAdminActionsNow()); }, this.adminActionIntervalMs); } async stop(): Promise { + if (this.stopPromise) { + return this.stopPromise; + } + this.stopPromise = this.stopAndDrain(); + return this.stopPromise; + } + + private async stopAndDrain(): Promise { + this.stopping = true; if (this.reconcileTimer) { clearInterval(this.reconcileTimer); } @@ -514,6 +552,16 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { if (this.adminActionTimer) { clearInterval(this.adminActionTimer); } + await Promise.allSettled([...this.inFlightTasks]); + } + + private trackTask(task: Promise): void { + this.inFlightTasks.add(task); + void task + .catch((error) => { + console.error('[gateway-orchestrator] scheduled task failed', error); + }) + .finally(() => this.inFlightTasks.delete(task)); } async listRuntimeStates(profileNames: string[]): Promise { @@ -522,7 +570,7 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { } async reconcileNow(): Promise { - if (this.reconcileInFlight) { + if (this.stopping || this.reconcileInFlight) { return; } this.reconcileInFlight = true; @@ -531,8 +579,14 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { if (!profiles.length) { return; } + const activeOperationProfiles = new Set( + (await this.repository.listActiveOperationProfileNames?.(this.now())) ?? [] + ); const processStates = await this.loadProcessStatusMap(); for (const profile of profiles) { + if (this.resetInFlight.has(profile.profileName) || activeOperationProfiles.has(profile.profileName)) { + continue; + } const runtime = mapRuntimeStates([profile.profileName], processStates)[0]; const plan = planProfileReconcile(profile.status, runtime); if (plan.shouldStart) { @@ -547,7 +601,7 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { } async runScheduleNow(): Promise { - if (this.scheduleInFlight) { + if (this.stopping || this.scheduleInFlight) { return; } this.scheduleInFlight = true; @@ -593,7 +647,7 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { } async runBuildQueueNow(): Promise { - if (this.buildInFlight) { + if (this.stopping || this.buildInFlight) { return; } this.buildInFlight = true; @@ -614,9 +668,14 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { startedAt, error: null, }); - const result = await this.runBuildCommands(queued.profileName, queued.buildCommitSha); + const { result, workspace } = await this.runBuildCommands(queued.buildCommitSha); const completedAt = this.now().toISOString(); if (result.ok) { + await this.repository.updateWorkspaceUsage( + queued.profileName, + workspace.root, + this.now().toISOString() + ); await this.repository.updateBuildStatus(queued.profileName, 'SUCCEEDED', { completedAt, error: null, @@ -650,84 +709,222 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { } async runOperationsNow(): Promise { - if (this.operationInFlight || this.buildInFlight) { + if (this.stopping || this.operationInFlight || this.buildInFlight) { return; } this.operationInFlight = true; try { - const operation = await this.repository.claimNextOperation(this.now()); + const operation = await this.repository.claimNextOperation(this.now(), { + ownerId: this.operationLeaseOwner, + durationMs: OPERATION_LEASE_DURATION_MS, + }); if (!operation) { return; } - await this.handleOperation(operation); + const heartbeatTimer = this.repository.renewOperationLease + ? setInterval(() => { + void this.repository + .renewOperationLease?.( + operation.id, + this.operationLeaseOwner, + this.now(), + OPERATION_LEASE_DURATION_MS + ) + .catch((error) => { + console.error('[gateway-orchestrator] operation heartbeat failed', error); + }); + }, OPERATION_HEARTBEAT_INTERVAL_MS) + : undefined; + try { + await this.handleOperation(operation); + } finally { + if (heartbeatTimer) { + clearInterval(heartbeatTimer); + } + } } finally { this.operationInFlight = false; } } - private async handleOperation(operation: GatewayOperationRecord): Promise { - const profile = await this.repository.getProfile(operation.profileName); - if (!profile) { - await this.repository.completeOperation(operation.id, 'FAILED', { - error: 'Profile not found.', - }); + private async assertOperationLease(operationId: string): Promise { + if (!this.repository.renewOperationLease) { return; } + const renewed = await this.repository.renewOperationLease( + operationId, + this.operationLeaseOwner, + this.now(), + OPERATION_LEASE_DURATION_MS + ); + if (!renewed) { + throw new OperationLeaseLostError(`Operation lease lost: ${operationId}`); + } + } + + private async handleOperation(operation: GatewayOperationRecord): Promise { + const assertLease = () => this.assertOperationLease(operation.id); + const profile = await this.repository.getProfile(operation.profileName); + if (!profile) { + await this.repository.completeOperation( + operation.id, + 'FAILED', + { + error: 'Profile not found.', + }, + this.operationLeaseOwner + ); + return; + } + const updateOperationProfile = async ( + patch: GatewayClaimedProfileUpdate, + fallback: () => Promise + ): Promise => { + if (this.repository.updateProfileForOperation) { + const updated = await this.repository.updateProfileForOperation( + operation.id, + this.operationLeaseOwner, + profile.profileName, + patch + ); + if (!updated) { + throw new OperationLeaseLostError(`Operation lease lost while updating profile: ${operation.id}`); + } + return updated; + } + await assertLease(); + return fallback(); + }; + let resolvedCommitSha: string | undefined; try { if (operation.type === 'START') { - const updated = await this.repository.updateStatus(profile.profileName, 'RUNNING', { - preopenAt: null, - openAt: null, - scheduledStartAt: null, - }); - const started = await this.startProfile(updated ?? profile); + const updated = await updateOperationProfile( + { + status: 'RUNNING', + preopenAt: null, + openAt: null, + scheduledStartAt: null, + }, + () => + this.repository.updateStatus(profile.profileName, 'RUNNING', { + preopenAt: null, + openAt: null, + scheduledStartAt: null, + }) + ); + const started = await this.startProfile(updated ?? profile, assertLease); if (!started) { + await updateOperationProfile({ status: 'STOPPED' }, () => + this.repository.updateStatus(profile.profileName, 'STOPPED') + ); throw new Error('Failed to start profile processes.'); } - await this.repository.completeOperation(operation.id, 'SUCCEEDED', { error: null }); + await updateOperationProfile({ lastError: null }, async () => { + await this.repository.updateLastError(profile.profileName, null); + return this.repository.getProfile(profile.profileName); + }); + await this.repository.completeOperation( + operation.id, + 'SUCCEEDED', + { error: null }, + this.operationLeaseOwner + ); return; } if (operation.type === 'STOP') { - await this.repository.updateStatus(profile.profileName, 'STOPPED'); - await this.stopProfile(profile); - await this.repository.completeOperation(operation.id, 'SUCCEEDED', { error: null }); + await updateOperationProfile({ status: 'STOPPED' }, () => + this.repository.updateStatus(profile.profileName, 'STOPPED') + ); + await this.stopProfile(profile, assertLease); + await this.repository.completeOperation( + operation.id, + 'SUCCEEDED', + { error: null }, + this.operationLeaseOwner + ); return; } if (!operation.sourceMode || !operation.sourceRef) { throw new Error('Reset source mode and ref are required.'); } - const commitSha = await this.workspaceManager.resolveCommit(operation.sourceMode, operation.sourceRef); + const commitSha = + operation.resolvedCommitSha ?? + (await this.workspaceManager.resolveCommit(operation.sourceMode, operation.sourceRef)); + resolvedCommitSha = commitSha; + if (!operation.resolvedCommitSha && this.repository.pinOperationResolvedCommit) { + const pinned = await this.repository.pinOperationResolvedCommit( + operation.id, + this.operationLeaseOwner, + commitSha + ); + if (!pinned) { + throw new OperationLeaseLostError(`Operation lease lost while pinning commit: ${operation.id}`); + } + } + await assertLease(); const payload = normalizeMeta(operation.payload); const install = isRecord(payload.install) ? payload.install : {}; + const installOperationId = + typeof payload.installOperationId === 'string' ? payload.installOperationId : operation.id; const resetAction: GatewayAdminActionRecord = { action: operation.scheduledAt ? 'RESET_SCHEDULED' : 'RESET_NOW', requestedAt: operation.createdAt, scheduledAt: operation.scheduledAt ?? null, reason: operation.reason ?? null, + installOperationId, install, }; - const result = await this.handleResetAction(profile, resetAction, commitSha); + const result = await this.handleResetAction(profile, resetAction, commitSha, assertLease, operation.id); if (result.status === 'REQUESTED') { const retryAt = new Date(this.now().getTime() + this.adminActionIntervalMs).toISOString(); - await this.repository.requeueOperation(operation.id, result.detail, retryAt); + await this.repository.requeueOperation(operation.id, result.detail, retryAt, this.operationLeaseOwner); return; } if (result.status !== 'APPLIED') { throw new Error(result.detail ?? 'Reset failed.'); } - await this.repository.completeOperation(operation.id, 'SUCCEEDED', { - resolvedCommitSha: commitSha, - error: null, - }); + await this.repository.completeOperation( + operation.id, + 'SUCCEEDED', + { + resolvedCommitSha: commitSha, + error: null, + }, + this.operationLeaseOwner + ); } catch (error) { + if ( + error instanceof OperationLeaseLostError || + (error instanceof Error && error.message.startsWith('Operation lease lost before')) + ) { + return; + } const detail = error instanceof Error ? error.message : String(error); - await this.repository.completeOperation(operation.id, 'FAILED', { error: detail }); + try { + await this.repository.completeOperation( + operation.id, + 'FAILED', + { + ...(resolvedCommitSha ? { resolvedCommitSha } : {}), + error: detail, + }, + this.operationLeaseOwner + ); + } catch (completionError) { + if ( + completionError instanceof Error && + completionError.message.startsWith('Operation lease lost before') + ) { + return; + } + throw completionError; + } } } private async runAdminActionsNow(): Promise { - if (this.adminActionInFlight) { + if (this.stopping || this.adminActionInFlight) { return; } this.adminActionInFlight = true; @@ -811,7 +1008,9 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { private async handleResetAction( profile: GatewayProfileRecord, action: GatewayAdminActionRecord, - commitShaOverride?: string + commitShaOverride?: string, + assertLease?: () => Promise, + operationId?: string ): Promise { // 리셋 요청을 빌드+재기동 흐름으로 처리한다. if (this.resetInFlight.has(profile.profileName)) { @@ -839,6 +1038,26 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { } this.buildInFlight = true; this.resetInFlight.add(profile.profileName); + let releasePrepared = false; + const updateClaimedProfile = async ( + patch: GatewayClaimedProfileUpdate, + fallback: () => Promise + ): Promise => { + if (operationId && this.repository.updateProfileForOperation) { + const updated = await this.repository.updateProfileForOperation( + operationId, + this.operationLeaseOwner, + profile.profileName, + patch + ); + if (!updated) { + throw new OperationLeaseLostError(`Operation lease lost while updating profile: ${operationId}`); + } + return updated; + } + await assertLease?.(); + return fallback(); + }; try { const { installOptions, @@ -849,86 +1068,178 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { } = parseInstallOptions(action); const tickOverride = installOptions?.turnTermMinutes !== undefined ? installOptions.turnTermMinutes * 60 : undefined; - const seedInfo = await this.resolveResetSeedInfo(profile, { - scenarioId: installScenarioId, - tickSeconds: tickOverride, - }); - if (!seedInfo.scenarioId) { + const scenarioId = installScenarioId ?? parseScenarioId(profile.scenario); + if (!scenarioId) { return { status: 'FAILED', detail: 'scenarioId is missing' }; } + const profileDatabaseUrl = this.resolveProfileDatabaseUrl(profile); + const seedTime = + openAt ?? + (action.scheduledAt && action.action === 'RESET_SCHEDULED' + ? new Date(action.scheduledAt) + : (parseDateTime(action.requestedAt) ?? this.now())); + const startedAt = this.now().toISOString(); + await updateClaimedProfile( + { + buildStatus: 'RUNNING', + buildRequestedAt: startedAt, + buildStartedAt: startedAt, + buildError: null, + buildCommitSha: commitSha, + }, + () => + this.repository.updateBuildStatus(profile.profileName, 'RUNNING', { + requestedAt: startedAt, + startedAt, + error: null, + commitSha, + }) + ); + const { result, workspace } = await this.runBuildCommands(commitSha); + await assertLease?.(); + if (!result.ok) { + const completedAt = this.now().toISOString(); + await updateClaimedProfile( + { + buildStatus: 'FAILED', + buildCompletedAt: completedAt, + buildError: result.output.slice(-4000), + }, + () => + this.repository.updateBuildStatus(profile.profileName, 'FAILED', { + completedAt, + error: result.output.slice(-4000), + }) + ); + return { status: 'FAILED', detail: 'selected workspace build failed' }; + } + await this.assertProfileSeedCli(workspace.root); + const seedInfo = await this.resolveResetSeedInfo( + profile, + { + scenarioId, + tickSeconds: tickOverride, + }, + profileDatabaseUrl + ); const profileMeta = normalizeMeta(profile.meta); const nextSeasonIdx = readMetaNumber(profileMeta, 'nextSeasonIdx'); const baseSeason = readMetaNumber(normalizeMeta(seedInfo.meta), 'season'); const season = nextSeasonIdx ?? baseSeason ?? 1; - const seedTime = - openAt ?? - (action.scheduledAt && action.action === 'RESET_SCHEDULED' ? new Date(action.scheduledAt) : this.now()); - const startedAt = this.now().toISOString(); - let activeProfile = profile; - if (installScenarioId !== null && String(installScenarioId) !== profile.scenario) { - const updated = await this.repository.updateScenario(profile.profileName, String(installScenarioId)); - if (updated) { - activeProfile = updated; - } + await updateClaimedProfile({ status: 'STOPPED' }, () => + this.repository.updateStatus(profile.profileName, 'STOPPED') + ); + await this.stopProfile(profile, assertLease); + await assertLease?.(); + const migrationResult = await this.runProfileMigration(workspace.root, profileDatabaseUrl); + await assertLease?.(); + if (!migrationResult.ok) { + const completedAt = this.now().toISOString(); + await updateClaimedProfile( + { + buildStatus: 'FAILED', + buildCompletedAt: completedAt, + buildError: migrationResult.output.slice(-4000), + }, + () => + this.repository.updateBuildStatus(profile.profileName, 'FAILED', { + completedAt, + error: migrationResult.output.slice(-4000), + }) + ); + return { status: 'FAILED', detail: 'profile database migration failed' }; } - await this.repository.updateStatus(profile.profileName, 'STOPPED'); - await this.stopProfile(activeProfile); - await this.repository.updateBuildStatus(profile.profileName, 'RUNNING', { - requestedAt: startedAt, - startedAt, - error: null, - commitSha, - }); - const result = await this.runBuildCommands(profile.profileName, commitSha); - const completedAt = this.now().toISOString(); - if (!result.ok) { - await this.repository.updateBuildStatus(profile.profileName, 'FAILED', { - completedAt, - error: result.output.slice(-4000), - }); - return { status: 'FAILED', detail: 'build failed' }; - } - const workspace = await this.workspaceManager.prepare(commitSha); - const resourceRoot = path.join(workspace.root, 'resources'); - const serverId = buildServerId(profile.profileName, seedTime); - await seedProfileDatabase({ + const serverId = buildServerId(profile.profileName, seedTime, installOptions?.installOperationId); + const seedResult = await this.runSelectedProfileSeed({ + workspaceRoot: workspace.root, databaseUrl: seedInfo.databaseUrl, - scenarioId: seedInfo.scenarioId, + scenarioId, tickSeconds: seedInfo.tickSeconds, now: seedTime, installOptions: { ...(installOptions ?? {}), season, serverId, + installCommitSha: commitSha, }, - scenarioOptions: { scenarioRoot: path.join(resourceRoot, 'scenario') }, - mapOptions: { mapRoot: path.join(resourceRoot, 'map') }, - unitSetOptions: { unitSetRoot: path.join(resourceRoot, 'unitset') }, adminUser, }); - await this.repository.updateBuildStatus(profile.profileName, 'SUCCEEDED', { - completedAt, - error: null, - }); + await assertLease?.(); + if (!seedResult.ok) { + throw new Error(`Selected profile seed failed: ${seedResult.output.slice(-4000)}`); + } + const completedAt = this.now().toISOString(); const now = this.now(); const shouldPreopen = openAt ? openAt.getTime() > now.getTime() : false; - await this.repository.updateStatus(profile.profileName, shouldPreopen ? 'PREOPEN' : 'RUNNING', { - preopenAt: preopenAt ? preopenAt.toISOString() : openAt ? openAt.toISOString() : null, - openAt: openAt ? openAt.toISOString() : null, - scheduledStartAt: action.scheduledAt ?? null, - }); - const builtProfile = (await this.repository.getProfile(profile.profileName)) ?? activeProfile; - const started = await this.startProfile(builtProfile); + const desiredStatus = shouldPreopen ? 'PREOPEN' : 'RUNNING'; + const publishedProfile = await updateClaimedProfile( + { + scenario: String(scenarioId), + status: desiredStatus, + buildStatus: 'SUCCEEDED', + buildWorkspace: workspace.root, + buildLastUsedAt: completedAt, + buildCompletedAt: completedAt, + buildError: null, + preopenAt: preopenAt ? preopenAt.toISOString() : openAt ? openAt.toISOString() : null, + openAt: openAt ? openAt.toISOString() : null, + scheduledStartAt: action.scheduledAt ?? null, + }, + async () => { + await this.repository.updateWorkspaceUsage(profile.profileName, workspace.root, completedAt); + await this.repository.updateBuildStatus(profile.profileName, 'SUCCEEDED', { + completedAt, + error: null, + }); + if (String(scenarioId) !== profile.scenario) { + await this.repository.updateScenario(profile.profileName, String(scenarioId)); + } + return this.repository.updateStatus(profile.profileName, desiredStatus, { + preopenAt: preopenAt ? preopenAt.toISOString() : openAt ? openAt.toISOString() : null, + openAt: openAt ? openAt.toISOString() : null, + scheduledStartAt: action.scheduledAt ?? null, + }); + } + ); + releasePrepared = true; + const builtProfile = publishedProfile ?? { + ...profile, + scenario: String(scenarioId), + status: desiredStatus, + buildWorkspace: workspace.root, + }; + const started = await this.startProfile(builtProfile, assertLease); if (!started) { + await updateClaimedProfile({ status: 'STOPPED', lastError: 'Failed to start profile processes.' }, () => + this.repository.updateStatus(profile.profileName, 'STOPPED') + ); return { status: 'FAILED', detail: 'reset completed but profile processes failed to start' }; } + await updateClaimedProfile({ lastError: null }, async () => { + await this.repository.updateLastError(profile.profileName, null); + return this.repository.getProfile(profile.profileName); + }); return { status: 'APPLIED', detail: 'reset completed via rebuild' }; } catch (error) { + if (error instanceof OperationLeaseLostError) { + throw error; + } const detail = error instanceof Error ? error.message : String(error); - await this.repository.updateBuildStatus(profile.profileName, 'FAILED', { - completedAt: this.now().toISOString(), - error: detail, - }); + if (!releasePrepared) { + const completedAt = this.now().toISOString(); + await updateClaimedProfile( + { + buildStatus: 'FAILED', + buildCompletedAt: completedAt, + buildError: detail, + }, + () => + this.repository.updateBuildStatus(profile.profileName, 'FAILED', { + completedAt, + error: detail, + }) + ); + } return { status: 'FAILED', detail }; } finally { this.buildInFlight = false; @@ -938,17 +1249,15 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { private async resolveResetSeedInfo( profile: GatewayProfileRecord, - overrides?: { scenarioId?: number | null; tickSeconds?: number } + overrides?: { scenarioId?: number | null; tickSeconds?: number }, + databaseUrlOverride?: string ): Promise<{ databaseUrl: string; scenarioId: number | null; tickSeconds?: number; meta: Record; }> { - const databaseUrl = resolvePostgresConfigFromEnv({ - env: this.processConfig.baseEnv ?? process.env, - schema: profile.profile, - }).url; + const databaseUrl = databaseUrlOverride ?? this.resolveProfileDatabaseUrl(profile); let scenarioId = overrides?.scenarioId ?? parseScenarioId(profile.scenario); let tickSeconds: number | undefined = overrides?.tickSeconds; let meta: Record = {}; @@ -980,15 +1289,84 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { return { databaseUrl, scenarioId, tickSeconds, meta }; } - private async runBuildCommands( - profileName: string, - commitSha: string - ): Promise>> { + private async runBuildCommands(commitSha: string): Promise<{ + result: Awaited>; + workspace: Awaited>; + }> { const workspace = await this.workspaceManager.prepare(commitSha); - const lastUsedAt = this.now().toISOString(); - await this.repository.updateWorkspaceUsage(profileName, workspace.root, lastUsedAt); const commands = buildWorkspaceCommands(workspace.root, workspace.needsInstall, this.processConfig.baseEnv); - return this.buildRunner.run(commands); + return { result: await this.buildRunner.run(commands), workspace }; + } + + private async assertProfileSeedCli(workspaceRoot: string): Promise { + const sourcePath = path.join(workspaceRoot, 'app', 'gateway-api', 'src', 'orchestrator', 'profileSeedCli.ts'); + try { + await fs.access(sourcePath); + } catch { + throw new Error(`Selected commit does not provide the profile seed CLI: ${sourcePath}`); + } + } + + private async runProfileMigration( + workspaceRoot: string, + profileDatabaseUrl: string + ): Promise>> { + return this.buildRunner.run([ + buildProfileMigrationCommand(workspaceRoot, profileDatabaseUrl, this.processConfig.baseEnv), + ]); + } + + private async runSelectedProfileSeed(options: { + workspaceRoot: string; + databaseUrl: string; + scenarioId: number; + tickSeconds?: number; + now: Date; + installOptions?: ScenarioInstallOptions; + adminUser?: AdminSeedUser | null; + }): Promise>> { + const tempDirectory = await fs.mkdtemp(path.join(os.tmpdir(), 'sammo-profile-seed-')); + const requestFile = path.join(tempDirectory, 'request.json'); + try { + await fs.writeFile( + requestFile, + JSON.stringify({ + scenarioId: options.scenarioId, + tickSeconds: options.tickSeconds, + now: options.now.toISOString(), + installOptions: options.installOptions + ? { + ...options.installOptions, + preopenAt: options.installOptions.preopenAt?.toISOString() ?? null, + } + : undefined, + adminUser: options.adminUser, + }), + { encoding: 'utf8', mode: 0o600 } + ); + return await this.buildRunner.run([ + { + command: process.execPath, + args: [path.join(options.workspaceRoot, 'app', 'gateway-api', 'dist', 'index.js')], + cwd: options.workspaceRoot, + env: { + ...(this.processConfig.baseEnv ?? {}), + DATABASE_URL: options.databaseUrl, + GATEWAY_ROLE: 'profile-seed', + PROFILE_SEED_REQUEST_FILE: requestFile, + }, + }, + ]); + } finally { + await fs.rm(tempDirectory, { recursive: true, force: true }); + } + } + + private resolveProfileDatabaseUrl(profile: GatewayProfileRecord): string { + return resolvePostgresConfigFromEnv({ + env: this.processConfig.baseEnv ?? process.env, + schema: profile.profile, + }).url; } async cleanupStaleWorkspaces(): Promise<{ removed: string[]; skipped: string[] }> { @@ -1070,42 +1448,72 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { return cutoff; } - private async startProfile(profile: GatewayProfileRecord): Promise { + private async startProfile(profile: GatewayProfileRecord, assertLease?: () => Promise): Promise { const definitions = buildProcessDefinitions(profile, this.processConfig); + const orderedDefinitions = [ + definitions.api, + definitions.daemon, + definitions.auction, + definitions.battleSim, + definitions.tournament, + ]; + const attemptedNames: string[] = []; try { - await this.processManager.start(definitions.api); - await this.processManager.start(definitions.daemon); - await this.processManager.start(definitions.auction); - await this.processManager.start(definitions.battleSim); - await this.processManager.start(definitions.tournament); - await this.repository.updateLastError(profile.profileName, null); + for (const definition of orderedDefinitions) { + await assertLease?.(); + attemptedNames.push(definition.name); + await this.processManager.start(definition); + await assertLease?.(); + } + if (!assertLease) { + await this.repository.updateLastError(profile.profileName, null); + } return true; } catch (error) { - await this.repository.updateLastError( - profile.profileName, - error instanceof Error ? error.message : 'Failed to start processes.' - ); + if (error instanceof OperationLeaseLostError) { + throw error; + } + await assertLease?.(); + for (const name of attemptedNames.reverse()) { + await assertLease?.(); + try { + await this.processManager.delete(name); + } catch { + // Preserve the original start failure. Reconciliation can retry cleanup. + } + await assertLease?.(); + } + if (!assertLease) { + await this.repository.updateLastError( + profile.profileName, + error instanceof Error ? error.message : 'Failed to start processes.' + ); + } return false; } } - private async stopProfile(profile: GatewayProfileRecord): Promise { + private async stopProfile(profile: GatewayProfileRecord, assertLease?: () => Promise): Promise { const apiName = buildProcessName(profile.profileName, 'api'); const daemonName = buildProcessName(profile.profileName, 'daemon'); const auctionName = buildProcessName(profile.profileName, 'auction'); const battleSimName = buildProcessName(profile.profileName, 'battle-sim'); const tournamentName = buildProcessName(profile.profileName, 'tournament'); + await assertLease?.(); const existingNames = new Set((await this.processManager.list()).map((process) => process.name)); + await assertLease?.(); const failures: string[] = []; for (const name of [apiName, daemonName, auctionName, battleSimName, tournamentName]) { if (!existingNames.has(name)) { continue; } + await assertLease?.(); try { await this.processManager.stop(name); } catch { // Deleting the definition below also terminates a process that raced with stop. } + await assertLease?.(); try { await this.processManager.delete(name); } catch (error) { @@ -1113,6 +1521,7 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle { failures.push(`${name}: ${error instanceof Error ? error.message : String(error)}`); } } + await assertLease?.(); } if (failures.length > 0) { throw new Error(`Failed to stop profile processes: ${failures.join('; ')}`); diff --git a/app/gateway-api/src/orchestrator/profileRepository.ts b/app/gateway-api/src/orchestrator/profileRepository.ts index bf9b10b..c648999 100644 --- a/app/gateway-api/src/orchestrator/profileRepository.ts +++ b/app/gateway-api/src/orchestrator/profileRepository.ts @@ -38,6 +38,10 @@ export interface GatewayOperationRecord { startedAt?: string; completedAt?: string; error?: string; + leaseOwner?: string; + leaseUntil?: string; + heartbeatAt?: string; + attempts?: number; createdAt: string; updatedAt: string; } @@ -88,6 +92,23 @@ export interface GatewayProfileUpsertInput { meta?: GatewayPrisma.JsonObject; } +export interface GatewayClaimedProfileUpdate { + scenario?: string; + status?: GatewayProfileStatus; + buildStatus?: GatewayBuildStatus; + buildCommitSha?: string | null; + buildWorkspace?: string | null; + buildLastUsedAt?: string | null; + preopenAt?: string | null; + openAt?: string | null; + scheduledStartAt?: string | null; + buildRequestedAt?: string | null; + buildStartedAt?: string | null; + buildCompletedAt?: string | null; + buildError?: string | null; + lastError?: string | null; +} + export interface GatewayProfileRepository { listProfiles(): Promise; getProfile(profileName: string): Promise; @@ -122,21 +143,59 @@ export interface GatewayProfileRepository { updateWorkspaceUsage(profileName: string, workspace: string, lastUsedAt: string): Promise; clearWorkspaceUsage(profileNames: string[]): Promise; listOperations(options?: { profileName?: string; limit?: number }): Promise; + listActiveOperationProfileNames?(now: Date): Promise; getOperation(id: string): Promise; createOperation(input: GatewayOperationCreateInput): Promise; - claimNextOperation(now: Date): Promise; + claimNextOperation( + now: Date, + lease?: { ownerId: string; durationMs: number } + ): Promise; + renewOperationLease?(id: string, ownerId: string, now: Date, durationMs: number): Promise; + pinOperationResolvedCommit?(id: string, ownerId: string, resolvedCommitSha: string): Promise; + updateProfileForOperation?( + id: string, + ownerId: string, + profileName: string, + patch: GatewayClaimedProfileUpdate + ): Promise; completeOperation( id: string, status: Extract, - fields?: { resolvedCommitSha?: string | null; error?: string | null } + fields: { resolvedCommitSha?: string | null; error?: string | null } | undefined, + leaseOwner: string + ): Promise; + requeueOperation( + id: string, + detail: string | undefined, + retryAt: string | undefined, + leaseOwner: string ): Promise; - requeueOperation(id: string, detail?: string, retryAt?: string): Promise; cancelOperation(id: string): Promise; retryOperation(id: string, requestedBy: string): Promise; } const toIso = (value: Date | null): string | undefined => (value ? value.toISOString() : undefined); +export const buildRetryOperationPayload = ( + previousPayload: GatewayPrisma.JsonObject, + previousOperationId: string +): GatewayPrisma.JsonObject => ({ + ...previousPayload, + installOperationId: + typeof previousPayload.installOperationId === 'string' + ? previousPayload.installOperationId + : previousOperationId, +}); + +export const buildRetryOperationSource = (previous: { + sourceMode: GatewaySourceMode | null; + sourceRef: string | null; + resolvedCommitSha: string | null; +}): { sourceMode: GatewaySourceMode | null; sourceRef: string | null } => + previous.resolvedCommitSha + ? { sourceMode: 'COMMIT', sourceRef: previous.resolvedCommitSha } + : { sourceMode: previous.sourceMode, sourceRef: previous.sourceRef }; + type GatewayProfileRow = { profileName: string; profile: string; @@ -175,6 +234,10 @@ type GatewayOperationRow = { startedAt: Date | null; completedAt: Date | null; error: string | null; + leaseOwner: string | null; + leaseUntil: Date | null; + heartbeatAt: Date | null; + attempts: number; createdAt: Date; updatedAt: Date; }; @@ -219,6 +282,10 @@ const mapOperation = (row: GatewayOperationRow): GatewayOperationRecord => ({ startedAt: toIso(row.startedAt), completedAt: toIso(row.completedAt), error: row.error ?? undefined, + leaseOwner: row.leaseOwner ?? undefined, + leaseUntil: toIso(row.leaseUntil), + heartbeatAt: toIso(row.heartbeatAt), + attempts: row.attempts, createdAt: row.createdAt.toISOString(), updatedAt: row.updatedAt.toISOString(), }); @@ -424,6 +491,22 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat }); return rows.map(mapOperation); }, + async listActiveOperationProfileNames(now: Date): Promise { + const rows = await prisma.gatewayOperation.findMany({ + where: { + OR: [ + { status: 'RUNNING' }, + { + status: 'QUEUED', + OR: [{ scheduledAt: null }, { scheduledAt: { lte: now } }], + }, + ], + }, + select: { profileName: true }, + distinct: ['profileName'], + }); + return rows.map(({ profileName }) => profileName); + }, async getOperation(id: string): Promise { const row = await prisma.gatewayOperation.findUnique({ where: { id } }); return row ? mapOperation(row) : null; @@ -443,21 +526,55 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat }); return mapOperation(row); }, - async claimNextOperation(now: Date): Promise { + async claimNextOperation( + now: Date, + lease?: { ownerId: string; durationMs: number } + ): Promise { const row = await prisma.$transaction(async (tx) => { - const candidate = await tx.gatewayOperation.findFirst({ - where: { - status: 'QUEUED', - OR: [{ scheduledAt: null }, { scheduledAt: { lte: now } }], - }, + await tx.$queryRaw>` + SELECT pg_advisory_xact_lock(hashtextextended('gateway_operation_claim', 0))::text AS lock_result + `; + const staleBefore = lease ? new Date(now.getTime() - lease.durationMs) : now; + const running = await tx.gatewayOperation.findFirst({ + where: { status: 'RUNNING' }, orderBy: { createdAt: 'asc' }, }); + const runningIsStale = Boolean( + lease && + running && + ((running.leaseUntil && running.leaseUntil < now) || + (!running.leaseUntil && running.startedAt && running.startedAt <= staleBefore)) + ); + if (running && !runningIsStale) { + return null; + } + const candidate = + running ?? + (await tx.gatewayOperation.findFirst({ + where: { + status: 'QUEUED', + OR: [{ scheduledAt: null }, { scheduledAt: { lte: now } }], + }, + orderBy: { createdAt: 'asc' }, + })); if (!candidate) { return null; } const claimed = await tx.gatewayOperation.updateMany({ - where: { id: candidate.id, status: 'QUEUED' }, - data: { status: 'RUNNING', startedAt: now, error: null }, + where: + candidate.status === 'QUEUED' + ? { id: candidate.id, status: 'QUEUED' } + : { id: candidate.id, status: 'RUNNING', leaseUntil: candidate.leaseUntil }, + data: { + status: 'RUNNING', + startedAt: candidate.startedAt ?? now, + completedAt: null, + error: null, + leaseOwner: lease?.ownerId, + leaseUntil: lease ? new Date(now.getTime() + lease.durationMs) : undefined, + heartbeatAt: lease ? now : undefined, + attempts: { increment: 1 }, + }, }); if (claimed.count !== 1) { return null; @@ -466,33 +583,117 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat }); return row ? mapOperation(row) : null; }, + async renewOperationLease(id: string, ownerId: string, now: Date, durationMs: number): Promise { + const renewed = await prisma.gatewayOperation.updateMany({ + where: { id, status: 'RUNNING', leaseOwner: ownerId }, + data: { + leaseUntil: new Date(now.getTime() + durationMs), + heartbeatAt: now, + }, + }); + return renewed.count === 1; + }, + async pinOperationResolvedCommit(id: string, ownerId: string, resolvedCommitSha: string): Promise { + const pinned = await prisma.gatewayOperation.updateMany({ + where: { + id, + status: 'RUNNING', + leaseOwner: ownerId, + OR: [{ resolvedCommitSha: null }, { resolvedCommitSha }], + }, + data: { resolvedCommitSha }, + }); + return pinned.count === 1; + }, + async updateProfileForOperation( + id: string, + ownerId: string, + profileName: string, + patch: GatewayClaimedProfileUpdate + ): Promise { + const row = await prisma.$transaction(async (tx) => { + const owned = await tx.$queryRaw>` + SELECT "id" + FROM "gateway_operation" + WHERE "id" = ${id} + AND "profile_name" = ${profileName} + AND "status" = 'RUNNING' + AND "lease_owner" = ${ownerId} + FOR UPDATE + `; + if (owned.length !== 1) { + return null; + } + const toDate = (value: string | null | undefined): Date | null | undefined => + value === undefined ? undefined : value === null ? null : new Date(value); + return tx.gatewayProfile.update({ + where: { profileName }, + data: { + scenario: patch.scenario, + status: patch.status, + buildStatus: patch.buildStatus, + buildCommitSha: patch.buildCommitSha, + buildWorkspace: patch.buildWorkspace, + buildLastUsedAt: toDate(patch.buildLastUsedAt), + preopenAt: toDate(patch.preopenAt), + openAt: toDate(patch.openAt), + scheduledStartAt: toDate(patch.scheduledStartAt), + buildRequestedAt: toDate(patch.buildRequestedAt), + buildStartedAt: toDate(patch.buildStartedAt), + buildCompletedAt: toDate(patch.buildCompletedAt), + buildError: patch.buildError, + lastError: patch.lastError, + }, + }); + }); + return row ? mapProfile(row) : null; + }, async completeOperation( id: string, status: Extract, - fields?: { resolvedCommitSha?: string | null; error?: string | null } + fields: { resolvedCommitSha?: string | null; error?: string | null } | undefined, + leaseOwner: string ): Promise { - const row = await prisma.gatewayOperation.update({ - where: { id }, + const updated = await prisma.gatewayOperation.updateMany({ + where: { id, status: 'RUNNING', leaseOwner }, data: { status, completedAt: new Date(), - resolvedCommitSha: - fields?.resolvedCommitSha === undefined ? undefined : fields.resolvedCommitSha, + resolvedCommitSha: fields?.resolvedCommitSha === undefined ? undefined : fields.resolvedCommitSha, error: fields?.error === undefined ? undefined : fields.error, + leaseOwner: null, + leaseUntil: null, + heartbeatAt: null, }, }); + if (updated.count !== 1) { + throw new Error(`Operation lease lost before completion: ${id}`); + } + const row = await prisma.gatewayOperation.findUniqueOrThrow({ where: { id } }); return mapOperation(row); }, - async requeueOperation(id: string, detail?: string, retryAt?: string): Promise { - const row = await prisma.gatewayOperation.update({ - where: { id }, + async requeueOperation( + id: string, + detail: string | undefined, + retryAt: string | undefined, + leaseOwner: string + ): Promise { + const updated = await prisma.gatewayOperation.updateMany({ + where: { id, status: 'RUNNING', leaseOwner }, data: { status: 'QUEUED', startedAt: null, error: detail, scheduledAt: retryAt ? new Date(retryAt) : undefined, + leaseOwner: null, + leaseUntil: null, + heartbeatAt: null, }, }); + if (updated.count !== 1) { + throw new Error(`Operation lease lost before requeue: ${id}`); + } + const row = await prisma.gatewayOperation.findUniqueOrThrow({ where: { id } }); return mapOperation(row); }, async cancelOperation(id: string): Promise { @@ -508,13 +709,15 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat if (!previous || (previous.status !== 'FAILED' && previous.status !== 'CANCELLED')) { return null; } + const previousPayload = previous.payload as GatewayPrisma.JsonObject; + const retrySource = buildRetryOperationSource(previous); return tx.gatewayOperation.create({ data: { profileName: previous.profileName, type: previous.type, - sourceMode: previous.sourceMode, - sourceRef: previous.sourceRef, - payload: previous.payload as GatewayPrisma.JsonObject, + sourceMode: retrySource.sourceMode, + sourceRef: retrySource.sourceRef, + payload: buildRetryOperationPayload(previousPayload, previous.id), reason: previous.reason, requestedBy, scheduledAt: null, diff --git a/app/gateway-api/src/orchestrator/profileSeedCli.ts b/app/gateway-api/src/orchestrator/profileSeedCli.ts new file mode 100644 index 0000000..e6db525 --- /dev/null +++ b/app/gateway-api/src/orchestrator/profileSeedCli.ts @@ -0,0 +1,70 @@ +import fs from 'node:fs/promises'; +import path from 'node:path'; + +import type { ScenarioInstallOptions } from '@sammo-ts/game-engine'; + +import { seedProfileDatabase, type AdminSeedUser } from './seedProfileDatabase.js'; + +interface ProfileSeedRequest { + scenarioId: number; + tickSeconds?: number; + now: string; + installOptions?: Omit & { preopenAt?: string | null }; + adminUser?: AdminSeedUser | null; +} + +export const parseProfileSeedRequest = (value: unknown): ProfileSeedRequest => { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error('Profile seed request must be an object.'); + } + const request = value as Record; + if (typeof request.scenarioId !== 'number' || !Number.isInteger(request.scenarioId)) { + throw new Error('Profile seed scenarioId must be an integer.'); + } + if (typeof request.now !== 'string' || Number.isNaN(new Date(request.now).getTime())) { + throw new Error('Profile seed now must be an ISO date-time.'); + } + if ( + request.tickSeconds !== undefined && + (typeof request.tickSeconds !== 'number' || !Number.isFinite(request.tickSeconds)) + ) { + throw new Error('Profile seed tickSeconds must be finite.'); + } + return request as unknown as ProfileSeedRequest; +}; + +export const runProfileSeedCli = async (env: NodeJS.ProcessEnv = process.env): Promise => { + const requestFile = env.PROFILE_SEED_REQUEST_FILE; + const databaseUrl = env.DATABASE_URL; + if (!requestFile) { + throw new Error('PROFILE_SEED_REQUEST_FILE is required.'); + } + if (!databaseUrl) { + throw new Error('DATABASE_URL is required.'); + } + + const request = parseProfileSeedRequest(JSON.parse(await fs.readFile(requestFile, 'utf8'))); + const rawPreopenAt = request.installOptions?.preopenAt; + const preopenAt = typeof rawPreopenAt === 'string' ? new Date(rawPreopenAt) : null; + if (preopenAt && Number.isNaN(preopenAt.getTime())) { + throw new Error('Profile seed preopenAt must be an ISO date-time.'); + } + const resourceRoot = path.join(process.cwd(), 'resources'); + + await seedProfileDatabase({ + databaseUrl, + scenarioId: request.scenarioId, + tickSeconds: request.tickSeconds, + now: new Date(request.now), + installOptions: request.installOptions + ? { + ...request.installOptions, + preopenAt, + } + : undefined, + scenarioOptions: { scenarioRoot: path.join(resourceRoot, 'scenario') }, + mapOptions: { mapRoot: path.join(resourceRoot, 'map') }, + unitSetOptions: { unitSetRoot: path.join(resourceRoot, 'unitset') }, + adminUser: request.adminUser, + }); +}; diff --git a/app/gateway-api/src/orchestrator/seedProfileDatabase.ts b/app/gateway-api/src/orchestrator/seedProfileDatabase.ts index e9c6444..252c51a 100644 --- a/app/gateway-api/src/orchestrator/seedProfileDatabase.ts +++ b/app/gateway-api/src/orchestrator/seedProfileDatabase.ts @@ -1,5 +1,5 @@ import { seedScenarioToDatabase, type ScenarioInstallOptions } from '@sammo-ts/game-engine'; -import { createGamePostgresConnector } from '@sammo-ts/infra'; +import type { GamePrisma } from '@sammo-ts/infra'; import { asRecord } from '@sammo-ts/common'; export interface AdminSeedUser { @@ -64,87 +64,75 @@ const resolveAdminStats = (config: Record) => { }; }; -const resolveAdminName = async ( - prisma: Awaited>['prisma'], - adminUser: AdminSeedUser -): Promise => { +const resolveAdminName = async (prisma: GamePrisma.TransactionClient, adminUser: AdminSeedUser): Promise => { const base = (adminUser.displayName ?? adminUser.username ?? '').trim() || '관리자'; const existing = await prisma.general.findFirst({ where: { name: base } }); if (!existing) { return base; } const suffix = adminUser.id.replace(/[^a-zA-Z0-9]/g, '').slice(0, 4) || 'admin'; - for (let i = 1; i <= 5; i += 1) { + for (let i = 1; ; i += 1) { const candidate = `${base}_${suffix}${i}`; const found = await prisma.general.findFirst({ where: { name: candidate } }); if (!found) { return candidate; } } - return `${base}_${suffix}${Date.now().toString(36)}`; }; // 초기 설치/통합 테스트에서 관리자 장수를 자동 생성한다. -const ensureAdminGeneral = async (databaseUrl: string, adminUser: AdminSeedUser): Promise => { - const connector = createGamePostgresConnector({ url: databaseUrl }); - await connector.connect(); - try { - const prisma = connector.prisma; - const existing = await prisma.general.findFirst({ where: { userId: adminUser.id } }); - if (existing) { - return; - } - - const worldState = await prisma.worldState.findFirst(); - if (!worldState) { - return; - } - - const cityRows = await prisma.city.findMany({ - select: { id: true }, - orderBy: { id: 'asc' }, - }); - const cityId = - cityRows.length > 0 ? cityRows[hashString(adminUser.id) % cityRows.length]!.id : 0; - - const maxId = await prisma.general.aggregate({ _max: { id: true } }); - const nextId = (maxId._max.id ?? 0) + 1; - const stats = resolveAdminStats(asRecord(worldState.config)); - const name = await resolveAdminName(prisma, adminUser); - const meta = asRecord(worldState.meta); - const rawTurnTime = typeof meta.turntime === 'string' ? new Date(meta.turntime) : null; - const turnTime = - rawTurnTime && !Number.isNaN(rawTurnTime.getTime()) ? rawTurnTime : new Date(); - - await prisma.general.create({ - data: { - id: nextId, - userId: adminUser.id, - name, - nationId: 0, - cityId, - troopId: 0, - npcState: 0, - leadership: stats.leadership, - strength: stats.strength, - intel: stats.intelligence, - personalCode: 'None', - specialCode: 'None', - special2Code: 'None', - turnTime, - meta: { - createdBy: 'admin-seed', - killturn: 24, - }, - }, - }); - } finally { - await connector.disconnect(); +const ensureAdminGeneral = async (prisma: GamePrisma.TransactionClient, adminUser: AdminSeedUser): Promise => { + const existing = await prisma.general.findFirst({ where: { userId: adminUser.id } }); + if (existing) { + return; } + + const worldState = await prisma.worldState.findFirst(); + if (!worldState) { + throw new Error('Seeded world state is missing before admin general creation.'); + } + + const cityRows = await prisma.city.findMany({ + select: { id: true }, + orderBy: { id: 'asc' }, + }); + const cityId = cityRows.length > 0 ? cityRows[hashString(adminUser.id) % cityRows.length]!.id : 0; + + const maxId = await prisma.general.aggregate({ _max: { id: true } }); + const nextId = (maxId._max.id ?? 0) + 1; + const stats = resolveAdminStats(asRecord(worldState.config)); + const name = await resolveAdminName(prisma, adminUser); + const meta = asRecord(worldState.meta); + const rawTurnTime = typeof meta.turntime === 'string' ? new Date(meta.turntime) : null; + const turnTime = rawTurnTime && !Number.isNaN(rawTurnTime.getTime()) ? rawTurnTime : new Date(); + + await prisma.general.create({ + data: { + id: nextId, + userId: adminUser.id, + name, + nationId: 0, + cityId, + troopId: 0, + npcState: 0, + leadership: stats.leadership, + strength: stats.strength, + intel: stats.intelligence, + personalCode: 'None', + specialCode: 'None', + special2Code: 'None', + turnTime, + meta: { + createdBy: 'admin-seed', + killturn: 24, + }, + }, + }); }; // 시나리오 시드 후 관리자 장수를 포함한 초기 데이터를 준비한다. export const seedProfileDatabase = async (options: SeedProfileDatabaseOptions) => { + const adminUser = options.adminUser; const result = await seedScenarioToDatabase({ scenarioId: options.scenarioId, databaseUrl: options.databaseUrl, @@ -154,11 +142,8 @@ export const seedProfileDatabase = async (options: SeedProfileDatabaseOptions) = scenarioOptions: options.scenarioOptions, mapOptions: options.mapOptions, unitSetOptions: options.unitSetOptions, + onBeforeCommit: adminUser ? async (transaction) => ensureAdminGeneral(transaction, adminUser) : undefined, }); - if (options.adminUser) { - await ensureAdminGeneral(options.databaseUrl, options.adminUser); - } - return result; }; diff --git a/app/gateway-api/test/adminOperations.test.ts b/app/gateway-api/test/adminOperations.test.ts index 31085d6..ffa889e 100644 --- a/app/gateway-api/test/adminOperations.test.ts +++ b/app/gateway-api/test/adminOperations.test.ts @@ -32,6 +32,7 @@ const buildCaller = async ( }); const session = await sessions.createSession({ ...admin, roles: adminRoles }); const createdInputs: GatewayOperationCreateInput[] = []; + const operationRecords = new Map>>(); const createdRuntimeActions: Array> = []; const flushes: Array<{ userId: string; reason?: string; iconRevision?: string }> = []; const profile = { @@ -59,10 +60,12 @@ const buildCaller = async ( updateWorkspaceUsage: async () => {}, clearWorkspaceUsage: async () => {}, listOperations: async () => [], - getOperation: async () => null, + getOperation: async (id) => operationRecords.get(id) ?? null, createOperation: async (input) => { createdInputs.push(input); - return createOperation(input); + const operation = await createOperation(input); + operationRecords.set(operation.id, operation); + return operation; }, claimNextOperation: async () => null, completeOperation: async () => { @@ -103,7 +106,11 @@ const buildCaller = async ( reconcileNow: async () => {}, runScheduleNow: async () => {}, runBuildQueueNow: async () => {}, - runOperationsNow: async () => {}, + runOperationsNow: async () => { + for (const [id, operation] of operationRecords) { + operationRecords.set(id, { ...operation, status: 'SUCCEEDED' }); + } + }, cleanupStaleWorkspaces: async () => ({ removed: [], skipped: [] }), listRuntimeStates: async () => [], }, @@ -180,6 +187,83 @@ describe('admin operation API', () => { }); }); +describe('legacy profile install API', () => { + const install = { + scenarioId: 1010, + turnTermMinutes: 60, + sync: false, + fiction: 1, + extend: false, + blockGeneralCreate: 0, + npcMode: 0, + showImgLevel: 0, + tournamentTrig: false, + joinMode: 'full' as const, + gitRef: 'HEAD', + }; + + const buildResetOperation = (input: GatewayOperationCreateInput) => ({ + id: '22222222-2222-4222-8222-222222222222', + profileName: input.profileName, + type: 'RESET' as const, + status: 'QUEUED' as const, + sourceMode: input.sourceMode, + sourceRef: input.sourceRef, + payload: input.payload ?? {}, + requestedBy: input.requestedBy, + createdAt: '2026-07-31T00:00:00.000Z', + updatedAt: '2026-07-31T00:00:00.000Z', + }); + + it('queues profiles.install instead of seeding the live database directly', async () => { + const harness = await buildCaller(async (input) => buildResetOperation(input)); + + await expect( + harness.caller.admin.profiles.install({ + profileName: 'che:2', + install, + reason: 'durable install', + }) + ).resolves.toMatchObject({ ok: true, operationId: '22222222-2222-4222-8222-222222222222' }); + expect(harness.createdInputs).toHaveLength(1); + expect(harness.createdInputs[0]).toMatchObject({ + profileName: 'che:2', + type: 'RESET', + sourceMode: 'COMMIT', + reason: 'durable install', + }); + expect(harness.createdInputs[0]?.payload).toMatchObject({ + install: { + scenarioId: 1010, + adminUser: { id: harness.admin.id }, + }, + }); + }); + + it('queues installNow through the same reset operation boundary', async () => { + const harness = await buildCaller(async (input) => buildResetOperation(input)); + + await expect( + harness.caller.admin.profiles.installNow({ + profileName: 'che:2', + install, + reason: 'no direct seed', + }) + ).resolves.toEqual({ ok: true, operationId: '22222222-2222-4222-8222-222222222222' }); + expect(harness.createdInputs[0]).toMatchObject({ + type: 'RESET', + sourceMode: 'COMMIT', + reason: 'no direct seed', + payload: { + install: { + scenarioId: 1010, + adminUser: { id: harness.admin.id }, + }, + }, + }); + }); +}); + describe('admin runtime clock action API', () => { const unusedCreateOperation: GatewayProfileRepository['createOperation'] = async () => { throw new Error('not used'); diff --git a/app/gateway-api/test/orchestratorOperations.test.ts b/app/gateway-api/test/orchestratorOperations.test.ts index 8d42eae..bdec9a9 100644 --- a/app/gateway-api/test/orchestratorOperations.test.ts +++ b/app/gateway-api/test/orchestratorOperations.test.ts @@ -41,11 +41,14 @@ const createHarness = ( failStart = false, failStop = false, processesPresent = true, - missingOnDelete = false + missingOnDelete = false, + workspaceManager?: GitWorkspaceManager, + startGate?: Promise ) => { let nextOperation: GatewayOperationRecord | null = operation; const statuses: string[] = []; const completions: GatewayOperationStatus[] = []; + const completionFields: Array<{ resolvedCommitSha?: string | null; error?: string | null } | undefined> = []; const started: ProcessDefinition[] = []; const stopped: string[] = []; const deleted: string[] = []; @@ -67,6 +70,7 @@ const createHarness = ( updateWorkspaceUsage: async () => {}, clearWorkspaceUsage: async () => {}, listOperations: async () => [], + listActiveOperationProfileNames: async () => [profile.profileName], getOperation: async () => operation, createOperation: async () => operation, claimNextOperation: async () => { @@ -74,8 +78,9 @@ const createHarness = ( nextOperation = null; return result; }, - completeOperation: async (_id, status) => { + completeOperation: async (_id, status, fields) => { completions.push(status); + completionFields.push(fields); return { ...operation, status }; }, requeueOperation: async () => ({ ...operation, status: 'QUEUED' }), @@ -94,7 +99,8 @@ const createHarness = ( ] : [], start: async (definition) => { - if (failStart) { + await startGate; + if (failStart && started.length === 2) { throw new Error('pm2 unavailable'); } started.push(definition); @@ -121,10 +127,12 @@ const createHarness = ( buildRunner: { run: async () => ({ ok: true, exitCode: 0, output: '' }), }, - workspaceManager: new GitWorkspaceManager({ - repoRoot: '/tmp/not-used', - worktreeRoot: '/tmp/not-used-worktrees', - }), + workspaceManager: + workspaceManager ?? + new GitWorkspaceManager({ + repoRoot: '/tmp/not-used', + worktreeRoot: '/tmp/not-used-worktrees', + }), processConfig: { workspaceRoot: '/srv/sammo', redisKeyPrefix: 'sammo:test', @@ -137,10 +145,20 @@ const createHarness = ( adminActionIntervalMs: 60_000, }); - return { orchestrator, statuses, completions, started, stopped, deleted }; + return { orchestrator, statuses, completions, completionFields, started, stopped, deleted }; }; describe('GatewayOrchestrator first-class operations', () => { + it('does not reconcile a profile while a durable operation is active', async () => { + const harness = createHarness(buildOperation('START')); + + await harness.orchestrator.reconcileNow(); + + expect(harness.started).toEqual([]); + expect(harness.stopped).toEqual([]); + expect(harness.deleted).toEqual([]); + }); + it('starts every profile process and records success', async () => { const harness = createHarness(buildOperation('START')); @@ -212,6 +230,11 @@ describe('GatewayOrchestrator first-class operations', () => { await harness.orchestrator.runOperationsNow(); expect(harness.completions).toEqual(['FAILED']); + expect(harness.deleted).toEqual([ + 'sammo:che:2:auction-worker', + 'sammo:che:2:turn-daemon', + 'sammo:che:2:game-api', + ]); }); it('attempts to stop every role before reporting a partial PM2 failure', async () => { @@ -235,4 +258,60 @@ describe('GatewayOrchestrator first-class operations', () => { ]); expect(harness.completions).toEqual(['FAILED']); }); + + it('records the resolved commit even when reset workspace preparation fails', async () => { + const resolvedCommitSha = 'abcdef0123456789abcdef0123456789abcdef01'; + const resetOperation: GatewayOperationRecord = { + id: '33333333-3333-4333-8333-333333333333', + profileName: profile.profileName, + type: 'RESET', + status: 'RUNNING', + sourceMode: 'COMMIT', + sourceRef: 'requested-ref', + payload: { install: { scenarioId: 1010, turnTermMinutes: 60 } }, + requestedBy: 'admin', + createdAt: '2026-07-31T00:00:00.000Z', + startedAt: '2026-07-31T00:00:00.000Z', + updatedAt: '2026-07-31T00:00:00.000Z', + }; + const workspaceManager = { + resolveCommit: async () => resolvedCommitSha, + prepare: async () => { + throw new Error('injected workspace preparation failure'); + }, + remove: async () => {}, + } as unknown as GitWorkspaceManager; + const harness = createHarness(resetOperation, false, false, true, false, workspaceManager); + + await harness.orchestrator.runOperationsNow(); + + expect(harness.completions).toEqual(['FAILED']); + expect(harness.completionFields).toEqual([ + { + resolvedCommitSha, + error: 'injected workspace preparation failure', + }, + ]); + }); + + it('drains an in-flight operation before shutdown completes', async () => { + let releaseStart: (() => void) | undefined; + const startGate = new Promise((resolve) => { + releaseStart = resolve; + }); + const harness = createHarness(buildOperation('START'), false, false, true, false, undefined, startGate); + harness.orchestrator.start(); + await new Promise((resolve) => setImmediate(resolve)); + + let stopped = false; + const stopPromise = harness.orchestrator.stop().then(() => { + stopped = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(stopped).toBe(false); + + releaseStart?.(); + await stopPromise; + expect(harness.completions).toEqual(['SUCCEEDED']); + }); }); diff --git a/app/gateway-api/test/orchestratorPlan.test.ts b/app/gateway-api/test/orchestratorPlan.test.ts index 1b59d2a..d96a84a 100644 --- a/app/gateway-api/test/orchestratorPlan.test.ts +++ b/app/gateway-api/test/orchestratorPlan.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from 'vitest'; import path from 'node:path'; import { + buildProfileMigrationCommand, buildProcessDefinitions, buildWorkspaceCommands, planProfileReconcile, @@ -161,7 +162,24 @@ describe('buildWorkspaceCommands', () => { ['--filter', '@sammo-ts/logic', 'build'], ['--filter', '@sammo-ts/game-api', 'build'], ['--filter', '@sammo-ts/game-engine', 'build'], + ['--filter', '@sammo-ts/gateway-api', 'build'], ]); expect(commands.every(({ cwd }) => cwd === workspaceRoot)).toBe(true); }); + + it('deploys the game schema migration after building the selected workspace', () => { + const workspaceRoot = '/srv/sammo/worktrees/0123456789abcdef'; + const databaseUrl = 'postgresql://integration.invalid/sammo?schema=che'; + const command = buildProfileMigrationCommand(workspaceRoot, databaseUrl, { NODE_ENV: 'production' }); + + expect(command).toEqual({ + command: 'pnpm', + args: ['--filter', '@sammo-ts/infra', 'prisma:migrate:deploy:game'], + cwd: workspaceRoot, + env: { + NODE_ENV: 'production', + DATABASE_URL: databaseUrl, + }, + }); + }); }); diff --git a/app/gateway-api/test/profileOperationLease.integration.test.ts b/app/gateway-api/test/profileOperationLease.integration.test.ts new file mode 100644 index 0000000..cc922f6 --- /dev/null +++ b/app/gateway-api/test/profileOperationLease.integration.test.ts @@ -0,0 +1,221 @@ +import { createGatewayPostgresConnector } from '@sammo-ts/infra'; +import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest'; + +import { createGatewayProfileRepository } from '../src/orchestrator/profileRepository.js'; + +const databaseUrl = process.env.GATEWAY_OPERATION_DATABASE_URL; +const describeDatabase = describe.runIf(Boolean(databaseUrl)); +const profileName = 'lease-test:1010'; +const secondProfileName = 'lease-test-2:1010'; + +describeDatabase('gateway operation lease and profile serialization', () => { + const connector = createGatewayPostgresConnector({ url: databaseUrl ?? '' }); + const repository = createGatewayProfileRepository(connector.prisma); + + beforeAll(async () => { + await connector.connect(); + await repository.upsertProfile({ + profile: 'lease-test', + scenario: '1010', + apiPort: 15999, + status: 'STOPPED', + }); + await repository.upsertProfile({ + profile: 'lease-test-2', + scenario: '1010', + apiPort: 15998, + status: 'STOPPED', + }); + }); + + afterEach(async () => { + await connector.prisma.gatewayOperation.deleteMany({ + where: { profileName: { in: [profileName, secondProfileName] } }, + }); + await connector.prisma.gatewayProfile.updateMany({ + where: { profileName: { in: [profileName, secondProfileName] } }, + data: { buildStatus: 'IDLE', buildError: null }, + }); + }); + + afterAll(async () => { + await connector.prisma.gatewayOperation.deleteMany({ + where: { profileName: { in: [profileName, secondProfileName] } }, + }); + await connector.prisma.gatewayProfile.deleteMany({ + where: { profileName: { in: [profileName, secondProfileName] } }, + }); + await connector.disconnect(); + }); + + it('allows only one queued or running operation for a profile', async () => { + const results = await Promise.allSettled([ + repository.createOperation({ + profileName, + type: 'RESET', + sourceMode: 'BRANCH', + sourceRef: 'main', + requestedBy: 'admin-a', + }), + repository.createOperation({ + profileName, + type: 'STOP', + requestedBy: 'admin-b', + }), + ]); + + expect(results.filter(({ status }) => status === 'fulfilled')).toHaveLength(1); + expect(results.filter(({ status }) => status === 'rejected')).toHaveLength(1); + await expect(repository.listOperations({ profileName })).resolves.toHaveLength(1); + }); + + it('serializes running operations globally across profiles', async () => { + const first = await repository.createOperation({ + profileName, + type: 'STOP', + requestedBy: 'admin-a', + }); + const second = await repository.createOperation({ + profileName: secondProfileName, + type: 'START', + requestedBy: 'admin-b', + }); + const now = new Date('2030-01-01T00:00:00.000Z'); + await expect( + repository.claimNextOperation(now, { ownerId: 'worker-a', durationMs: 1_000 }) + ).resolves.toMatchObject({ id: first.id, leaseOwner: 'worker-a' }); + await expect( + repository.claimNextOperation(now, { ownerId: 'worker-b', durationMs: 1_000 }) + ).resolves.toBeNull(); + await repository.completeOperation(first.id, 'SUCCEEDED', { error: null }, 'worker-a'); + await expect( + repository.claimNextOperation(now, { ownerId: 'worker-b', durationMs: 1_000 }) + ).resolves.toMatchObject({ id: second.id, leaseOwner: 'worker-b' }); + }); + + it('does not let a future queued operation suppress runtime reconciliation early', async () => { + const now = new Date('2030-01-01T00:00:00.000Z'); + await repository.createOperation({ + profileName, + type: 'RESET', + sourceMode: 'COMMIT', + sourceRef: 'abcdef', + scheduledAt: new Date(now.getTime() + 60_000).toISOString(), + requestedBy: 'admin', + }); + + await expect(repository.listActiveOperationProfileNames?.(now)).resolves.not.toContain(profileName); + await expect(repository.listActiveOperationProfileNames?.(new Date(now.getTime() + 60_000))).resolves.toContain( + profileName + ); + }); + + it('reclaims the same expired RUNNING operation without creating a second generation', async () => { + const operation = await repository.createOperation({ + profileName, + type: 'RESET', + sourceMode: 'COMMIT', + sourceRef: 'abcdef', + payload: { install: { scenarioId: 1010 } }, + requestedBy: 'admin', + }); + const startedAt = new Date('2030-01-01T00:00:00.000Z'); + const firstClaim = await repository.claimNextOperation(startedAt, { + ownerId: 'worker-a', + durationMs: 1_000, + }); + expect(firstClaim).toMatchObject({ id: operation.id, attempts: 1, leaseOwner: 'worker-a' }); + await expect( + repository.pinOperationResolvedCommit?.(operation.id, 'worker-a', 'pinned-commit-a') + ).resolves.toBe(true); + + await expect( + repository.claimNextOperation(new Date(startedAt.getTime() + 999), { + ownerId: 'worker-b', + durationMs: 1_000, + }) + ).resolves.toBeNull(); + const reclaimed = await repository.claimNextOperation(new Date(startedAt.getTime() + 1_001), { + ownerId: 'worker-b', + durationMs: 1_000, + }); + expect(reclaimed).toMatchObject({ + id: operation.id, + attempts: 2, + leaseOwner: 'worker-b', + resolvedCommitSha: 'pinned-commit-a', + }); + await expect( + repository.pinOperationResolvedCommit?.(operation.id, 'worker-a', 'pinned-commit-a') + ).resolves.toBe(false); + await expect( + repository.pinOperationResolvedCommit?.(operation.id, 'worker-b', 'pinned-commit-b') + ).resolves.toBe(false); + await expect( + repository.updateProfileForOperation?.(operation.id, 'worker-b', profileName, { + buildStatus: 'RUNNING', + buildError: 'worker-b-marker', + }) + ).resolves.toMatchObject({ buildStatus: 'RUNNING', buildError: 'worker-b-marker' }); + await expect( + repository.updateProfileForOperation?.(operation.id, 'worker-a', profileName, { + buildStatus: 'FAILED', + buildError: 'stale-worker-a', + }) + ).resolves.toBeNull(); + await expect(repository.getProfile(profileName)).resolves.toMatchObject({ + buildStatus: 'RUNNING', + buildError: 'worker-b-marker', + }); + await expect(repository.renewOperationLease?.(operation.id, 'worker-a', startedAt, 1_000)).resolves.toBe(false); + await expect(repository.renewOperationLease?.(operation.id, 'worker-b', startedAt, 1_000)).resolves.toBe(true); + await expect( + repository.completeOperation(operation.id, 'FAILED', { error: 'stale worker' }, 'worker-a') + ).rejects.toThrow('Operation lease lost before completion'); + await expect(repository.getOperation(operation.id)).resolves.toMatchObject({ + status: 'RUNNING', + leaseOwner: 'worker-b', + }); + await expect(repository.requeueOperation(operation.id, 'stale worker', undefined, 'worker-a')).rejects.toThrow( + 'Operation lease lost before requeue' + ); + await expect( + repository.completeOperation(operation.id, 'SUCCEEDED', { error: null }, 'worker-b') + ).resolves.toMatchObject({ status: 'SUCCEEDED' }); + }); + + it('pins retry to the first resolved commit and preserves its install generation', async () => { + const operation = await repository.createOperation({ + profileName, + type: 'RESET', + sourceMode: 'BRANCH', + sourceRef: 'main', + payload: { install: { scenarioId: 1010 } }, + requestedBy: 'admin-a', + }); + const claimed = await repository.claimNextOperation(new Date('2030-01-01T00:00:00.000Z'), { + ownerId: 'worker-a', + durationMs: 1_000, + }); + expect(claimed?.id).toBe(operation.id); + await repository.completeOperation( + operation.id, + 'FAILED', + { + resolvedCommitSha: 'abcdef0123456789abcdef0123456789abcdef01', + error: 'injected start failure', + }, + 'worker-a' + ); + + const retry = await repository.retryOperation(operation.id, 'admin-b'); + expect(retry).toMatchObject({ + sourceMode: 'COMMIT', + sourceRef: 'abcdef0123456789abcdef0123456789abcdef01', + payload: { + installOperationId: operation.id, + install: { scenarioId: 1010 }, + }, + }); + }); +}); diff --git a/app/gateway-api/test/profileRepository.test.ts b/app/gateway-api/test/profileRepository.test.ts new file mode 100644 index 0000000..39155e2 --- /dev/null +++ b/app/gateway-api/test/profileRepository.test.ts @@ -0,0 +1,54 @@ +import { describe, expect, it } from 'vitest'; + +import { buildRetryOperationPayload, buildRetryOperationSource } from '../src/orchestrator/profileRepository.js'; + +describe('buildRetryOperationPayload', () => { + it('pins the first failed operation as the install generation', () => { + expect( + buildRetryOperationPayload({ install: { scenarioId: 1010 } }, '11111111-1111-4111-8111-111111111111') + ).toEqual({ + install: { scenarioId: 1010 }, + installOperationId: '11111111-1111-4111-8111-111111111111', + }); + }); + + it('preserves the original install generation across chained retries', () => { + expect( + buildRetryOperationPayload( + { + install: { scenarioId: 903 }, + installOperationId: 'original-install-generation', + }, + 'newer-failed-operation' + ) + ).toEqual({ + install: { scenarioId: 903 }, + installOperationId: 'original-install-generation', + }); + }); +}); + +describe('buildRetryOperationSource', () => { + it('pins a resolved branch operation to the original commit', () => { + expect( + buildRetryOperationSource({ + sourceMode: 'BRANCH', + sourceRef: 'main', + resolvedCommitSha: 'abcdef0123456789abcdef0123456789abcdef01', + }) + ).toEqual({ + sourceMode: 'COMMIT', + sourceRef: 'abcdef0123456789abcdef0123456789abcdef01', + }); + }); + + it('keeps the requested source when resolution never completed', () => { + expect( + buildRetryOperationSource({ + sourceMode: 'BRANCH', + sourceRef: 'main', + resolvedCommitSha: null, + }) + ).toEqual({ sourceMode: 'BRANCH', sourceRef: 'main' }); + }); +}); diff --git a/app/gateway-api/test/profileSeedAtomicity.integration.test.ts b/app/gateway-api/test/profileSeedAtomicity.integration.test.ts new file mode 100644 index 0000000..12cb2a2 --- /dev/null +++ b/app/gateway-api/test/profileSeedAtomicity.integration.test.ts @@ -0,0 +1,99 @@ +import path from 'node:path'; + +import { createGamePostgresConnector } from '@sammo-ts/infra'; +import { describe, expect, it } from 'vitest'; + +import { seedProfileDatabase } from '../src/orchestrator/seedProfileDatabase.js'; + +const databaseUrl = process.env.PROFILE_SEED_DATABASE_URL; +const schema = process.env.PROFILE_SEED_DATABASE_SCHEMA ?? 'profile_seed_atomicity'; +const describeDatabase = describe.runIf(Boolean(databaseUrl)); +const resourceRoot = path.resolve(process.cwd(), '../../resources'); + +describeDatabase('profile seed atomicity', () => { + it('rolls back the new season when administrator general creation fails', async () => { + if (!databaseUrl) { + throw new Error('PROFILE_SEED_DATABASE_URL is required.'); + } + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(schema)) { + throw new Error('PROFILE_SEED_DATABASE_SCHEMA must be a safe PostgreSQL identifier.'); + } + + const connector = createGamePostgresConnector({ url: databaseUrl }); + await seedProfileDatabase({ + databaseUrl, + scenarioId: 1010, + now: new Date('2034-01-01T00:00:00.000Z'), + installOptions: { + serverId: 'profile-seed-baseline', + installOperationId: 'profile-seed-baseline-operation', + }, + scenarioOptions: { scenarioRoot: path.join(resourceRoot, 'scenario') }, + mapOptions: { mapRoot: path.join(resourceRoot, 'map') }, + unitSetOptions: { unitSetRoot: path.join(resourceRoot, 'unitset') }, + }); + + await connector.connect(); + const prisma = connector.prisma; + const readSnapshot = async () => ({ + world: await prisma.worldState.findMany({ orderBy: { id: 'asc' } }), + nations: await prisma.nation.findMany({ orderBy: { id: 'asc' } }), + cities: await prisma.city.findMany({ orderBy: { id: 'asc' } }), + generals: await prisma.general.findMany({ orderBy: { id: 'asc' } }), + history: await prisma.gameHistory.findMany({ orderBy: { id: 'asc' } }), + }); + const before = await readSnapshot(); + + try { + await prisma.$executeRawUnsafe(` + CREATE OR REPLACE FUNCTION "${schema}".reject_admin_seed() + RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.meta ->> 'createdBy' = 'admin-seed' THEN + RAISE EXCEPTION 'injected administrator seed failure'; + END IF; + RETURN NEW; + END; + $$ + `); + await prisma.$executeRawUnsafe(` + CREATE TRIGGER reject_admin_seed + BEFORE INSERT ON "${schema}"."general" + FOR EACH ROW EXECUTE FUNCTION "${schema}".reject_admin_seed() + `); + + await expect( + seedProfileDatabase({ + databaseUrl, + scenarioId: 903, + now: new Date('2035-02-02T00:00:00.000Z'), + installOptions: { + serverId: 'profile-seed-failed', + installOperationId: 'profile-seed-failed-operation', + }, + scenarioOptions: { scenarioRoot: path.join(resourceRoot, 'scenario') }, + mapOptions: { mapRoot: path.join(resourceRoot, 'map') }, + unitSetOptions: { unitSetRoot: path.join(resourceRoot, 'unitset') }, + adminUser: { + id: 'profile-seed-admin', + username: 'profile-seed-admin', + displayName: '프로필 관리자', + }, + }) + ).rejects.toThrow('injected administrator seed failure'); + + expect(await readSnapshot()).toEqual(before); + await expect(prisma.general.findFirst({ where: { userId: 'profile-seed-admin' } })).resolves.toBeNull(); + await expect( + prisma.gameHistory.findUnique({ where: { serverId: 'profile-seed-failed' } }) + ).resolves.toBeNull(); + } finally { + await prisma.$executeRawUnsafe(`DROP TRIGGER IF EXISTS reject_admin_seed ON "${schema}"."general"`); + await prisma.$executeRawUnsafe(`DROP FUNCTION IF EXISTS "${schema}".reject_admin_seed()`); + await prisma.gameHistory.deleteMany({ + where: { serverId: { in: ['profile-seed-baseline', 'profile-seed-failed'] } }, + }); + await connector.disconnect(); + } + }); +}); diff --git a/app/gateway-api/test/profileSeedCli.integration.test.ts b/app/gateway-api/test/profileSeedCli.integration.test.ts new file mode 100644 index 0000000..9e79c00 --- /dev/null +++ b/app/gateway-api/test/profileSeedCli.integration.test.ts @@ -0,0 +1,101 @@ +import { spawn } from 'node:child_process'; +import fs from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; + +import { createGamePostgresConnector } from '@sammo-ts/infra'; +import { describe, expect, it } from 'vitest'; + +const databaseUrl = process.env.PROFILE_SEED_CLI_DATABASE_URL; +const describeDatabase = describe.runIf(Boolean(databaseUrl)); +const workspaceRoot = path.resolve(process.cwd(), '../..'); + +const runSeedCli = async (requestFile: string): Promise<{ code: number | null; output: string }> => + new Promise((resolve) => { + const child = spawn(process.execPath, [path.join(workspaceRoot, 'app/gateway-api/dist/index.js')], { + cwd: workspaceRoot, + env: { + ...process.env, + DATABASE_URL: databaseUrl, + GATEWAY_ROLE: 'profile-seed', + PROFILE_SEED_REQUEST_FILE: requestFile, + }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + let output = ''; + child.stdout.on('data', (chunk) => { + output += chunk.toString(); + }); + child.stderr.on('data', (chunk) => { + output += chunk.toString(); + }); + child.on('close', (code) => resolve({ code, output })); + }); + +describeDatabase('selected workspace profile seed CLI', () => { + it('seeds through the built gateway artifact with the serialized operation identity', async () => { + if (!databaseUrl) { + throw new Error('PROFILE_SEED_CLI_DATABASE_URL is required.'); + } + const tempDirectory = await fs.mkdtemp(path.join(os.tmpdir(), 'profile-seed-cli-test-')); + const requestFile = path.join(tempDirectory, 'request.json'); + const connector = createGamePostgresConnector({ url: databaseUrl }); + try { + await fs.writeFile( + requestFile, + JSON.stringify({ + scenarioId: 1010, + tickSeconds: 60, + now: '2036-03-03T00:00:00.000Z', + installOptions: { + serverId: 'selected-cli-seed', + installOperationId: 'selected-cli-operation', + installCommitSha: 'selected-cli-commit', + }, + adminUser: { + id: 'selected-cli-admin', + username: 'selected-cli-admin', + displayName: '선택 CLI 관리자', + }, + }), + { encoding: 'utf8', mode: 0o600 } + ); + + const result = await runSeedCli(requestFile); + expect(result, result.output).toMatchObject({ code: 0 }); + await connector.connect(); + const world = await connector.prisma.worldState.findFirstOrThrow(); + expect(world).toMatchObject({ + scenarioCode: '1010', + meta: { + installOperationId: 'selected-cli-operation', + installCommitSha: 'selected-cli-commit', + }, + }); + const adminGeneral = await connector.prisma.general.findFirstOrThrow({ + where: { userId: 'selected-cli-admin' }, + }); + expect(adminGeneral).toMatchObject({ meta: { createdBy: 'admin-seed' } }); + const history = await connector.prisma.gameHistory.findUniqueOrThrow({ + where: { serverId: 'selected-cli-seed' }, + }); + + const retry = await runSeedCli(requestFile); + expect(retry, retry.output).toMatchObject({ code: 0 }); + await expect(connector.prisma.worldState.findFirstOrThrow()).resolves.toEqual(world); + await expect( + connector.prisma.general.findFirstOrThrow({ where: { userId: 'selected-cli-admin' } }) + ).resolves.toEqual(adminGeneral); + await expect( + connector.prisma.gameHistory.findUniqueOrThrow({ where: { serverId: 'selected-cli-seed' } }) + ).resolves.toEqual(history); + await expect( + connector.prisma.gameHistory.count({ where: { serverId: 'selected-cli-seed' } }) + ).resolves.toBe(1); + } finally { + await connector.prisma.gameHistory.deleteMany({ where: { serverId: 'selected-cli-seed' } }); + await connector.disconnect(); + await fs.rm(tempDirectory, { recursive: true, force: true }); + } + }); +}); diff --git a/app/gateway-api/test/profileSeedCli.test.ts b/app/gateway-api/test/profileSeedCli.test.ts new file mode 100644 index 0000000..b0d5dfc --- /dev/null +++ b/app/gateway-api/test/profileSeedCli.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from 'vitest'; + +import { parseProfileSeedRequest } from '../src/orchestrator/profileSeedCli.js'; + +describe('parseProfileSeedRequest', () => { + it('accepts the serializable selected-workspace seed contract', () => { + expect( + parseProfileSeedRequest({ + scenarioId: 1010, + tickSeconds: 60, + now: '2030-01-01T00:00:00.000Z', + installOptions: { + installOperationId: 'operation-id', + installCommitSha: 'abcdef', + preopenAt: null, + }, + adminUser: { id: 'admin', username: 'admin' }, + }) + ).toMatchObject({ + scenarioId: 1010, + tickSeconds: 60, + installOptions: { installOperationId: 'operation-id', installCommitSha: 'abcdef' }, + }); + }); + + it('rejects an invalid timestamp before touching the database', () => { + expect(() => parseProfileSeedRequest({ scenarioId: 1010, now: 'not-a-date' })).toThrow( + 'Profile seed now must be an ISO date-time.' + ); + }); +}); diff --git a/packages/infra/prisma/gateway-migrations/20260731002000_add_active_operation_profile_unique/migration.sql b/packages/infra/prisma/gateway-migrations/20260731002000_add_active_operation_profile_unique/migration.sql new file mode 100644 index 0000000..6c402df --- /dev/null +++ b/packages/infra/prisma/gateway-migrations/20260731002000_add_active_operation_profile_unique/migration.sql @@ -0,0 +1,6 @@ +-- A profile database can have only one destructive/runtime operation in flight. +-- Existing duplicates intentionally make this migration fail so an operator can +-- inspect and resolve them instead of silently discarding an operation. +CREATE UNIQUE INDEX "gateway_operation_one_active_per_profile_idx" + ON "gateway_operation" ("profile_name") + WHERE "status" IN ('QUEUED', 'RUNNING'); diff --git a/packages/infra/prisma/gateway-migrations/20260731003000_add_operation_lease/migration.sql b/packages/infra/prisma/gateway-migrations/20260731003000_add_operation_lease/migration.sql new file mode 100644 index 0000000..93fec06 --- /dev/null +++ b/packages/infra/prisma/gateway-migrations/20260731003000_add_operation_lease/migration.sql @@ -0,0 +1,8 @@ +ALTER TABLE "gateway_operation" + ADD COLUMN "lease_owner" TEXT, + ADD COLUMN "lease_until" TIMESTAMPTZ, + ADD COLUMN "heartbeat_at" TIMESTAMPTZ, + ADD COLUMN "attempts" INTEGER NOT NULL DEFAULT 0; + +CREATE INDEX "gateway_operation_status_lease_until_created_at_idx" + ON "gateway_operation" ("status", "lease_until", "created_at"); diff --git a/packages/infra/prisma/gateway.prisma b/packages/infra/prisma/gateway.prisma index 26f2ad0..7d65690 100644 --- a/packages/infra/prisma/gateway.prisma +++ b/packages/infra/prisma/gateway.prisma @@ -174,6 +174,8 @@ model GatewayRuntimeAction { } model GatewayOperation { + /// A partial unique index in the gateway migration chain permits only one + /// QUEUED or RUNNING operation per profile. id String @id @default(uuid()) profileName String @map("profile_name") profile GatewayProfile @relation(fields: [profileName], references: [profileName], onDelete: Cascade) @@ -189,10 +191,15 @@ model GatewayOperation { startedAt DateTime? @map("started_at") completedAt DateTime? @map("completed_at") error String? + leaseOwner String? @map("lease_owner") + leaseUntil DateTime? @map("lease_until") + heartbeatAt DateTime? @map("heartbeat_at") + attempts Int @default(0) createdAt DateTime @default(now()) @map("created_at") updatedAt DateTime @updatedAt @map("updated_at") @@index([status, scheduledAt, createdAt]) + @@index([status, leaseUntil, createdAt]) @@index([profileName, createdAt]) @@map("gateway_operation") }