feat: Implement admin router and orchestrator for managing profiles and builds
- Added `adminRouter` for handling profile management, including listing, upserting, and updating statuses. - Introduced `BuildRunner` interface and `PnpmBuildRunner` class for executing build commands. - Created `GatewayOrchestrator` to manage profile states, reconcile processes, and handle build queues. - Implemented `Pm2ProcessManager` for managing processes using PM2. - Developed `GatewayProfileRepository` for interacting with the database to manage profiles. - Added utility functions for resolving workspace roots and managing process definitions. - Included tests for profile reconciliation logic.
This commit is contained in:
@@ -23,6 +23,7 @@
|
||||
"@trpc/server": "^11.4.4",
|
||||
"date-fns": "^4.1.0",
|
||||
"fastify": "^5.3.3",
|
||||
"pm2": "^5.4.3",
|
||||
"redis": "^5.10.0",
|
||||
"zod": "^4.2.1"
|
||||
}
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
import { TRPCError } from '@trpc/server';
|
||||
import { z } from 'zod';
|
||||
|
||||
import { procedure, router } from './trpc.js';
|
||||
import {
|
||||
GATEWAY_BUILD_STATUSES,
|
||||
GATEWAY_PROFILE_STATUSES,
|
||||
} from './orchestrator/profileRepository.js';
|
||||
|
||||
const zProfileStatus = z.enum(GATEWAY_PROFILE_STATUSES);
|
||||
const zBuildStatus = z.enum(GATEWAY_BUILD_STATUSES);
|
||||
|
||||
const adminProcedure = procedure.use(({ ctx, next }) => {
|
||||
if (!ctx.adminToken) {
|
||||
throw new TRPCError({
|
||||
code: 'FORBIDDEN',
|
||||
message: 'Admin token is not configured.',
|
||||
});
|
||||
}
|
||||
const provided =
|
||||
ctx.requestHeaders['x-admin-token'] ??
|
||||
ctx.requestHeaders['authorization'] ??
|
||||
'';
|
||||
const token =
|
||||
Array.isArray(provided) ? provided[0] ?? '' : (provided as string);
|
||||
if (!token || token !== ctx.adminToken) {
|
||||
throw new TRPCError({
|
||||
code: 'UNAUTHORIZED',
|
||||
message: 'Invalid admin token.',
|
||||
});
|
||||
}
|
||||
return next();
|
||||
});
|
||||
|
||||
export const adminRouter = router({
|
||||
profiles: router({
|
||||
list: adminProcedure.query(async ({ ctx }) => {
|
||||
const profiles = await ctx.profiles.listProfiles();
|
||||
const runtimeStates = await ctx.orchestrator.listRuntimeStates(
|
||||
profiles.map((profile) => profile.profileName)
|
||||
);
|
||||
const runtimeMap = new Map(
|
||||
runtimeStates.map((state) => [state.profileName, state])
|
||||
);
|
||||
return profiles.map((profile) => ({
|
||||
...profile,
|
||||
runtime: runtimeMap.get(profile.profileName) ?? {
|
||||
profileName: profile.profileName,
|
||||
apiRunning: false,
|
||||
daemonRunning: false,
|
||||
},
|
||||
}));
|
||||
}),
|
||||
upsert: adminProcedure
|
||||
.input(
|
||||
z.object({
|
||||
profile: z.string().min(1).max(32),
|
||||
scenario: z.string().min(1).max(64),
|
||||
apiPort: z.number().int().min(1).max(65535),
|
||||
status: zProfileStatus.optional(),
|
||||
scheduledStartAt: z.string().datetime().optional(),
|
||||
})
|
||||
)
|
||||
.mutation(async ({ ctx, input }) => {
|
||||
const status = input.status ?? 'STOPPED';
|
||||
return ctx.profiles.upsertProfile({
|
||||
profile: input.profile,
|
||||
scenario: input.scenario,
|
||||
apiPort: input.apiPort,
|
||||
status,
|
||||
scheduledStartAt: input.scheduledStartAt,
|
||||
});
|
||||
}),
|
||||
setStatus: adminProcedure
|
||||
.input(
|
||||
z.object({
|
||||
profileName: z.string().min(1),
|
||||
status: zProfileStatus,
|
||||
scheduledStartAt: z.string().datetime().optional(),
|
||||
})
|
||||
)
|
||||
.mutation(async ({ ctx, input }) => {
|
||||
if (input.status === 'RESERVED' && !input.scheduledStartAt) {
|
||||
throw new TRPCError({
|
||||
code: 'BAD_REQUEST',
|
||||
message: 'scheduledStartAt is required for RESERVED status.',
|
||||
});
|
||||
}
|
||||
const result = await ctx.profiles.updateStatus(
|
||||
input.profileName,
|
||||
input.status,
|
||||
input.status === 'RESERVED' ? input.scheduledStartAt : null
|
||||
);
|
||||
await ctx.orchestrator.reconcileNow();
|
||||
return result;
|
||||
}),
|
||||
requestBuild: adminProcedure
|
||||
.input(
|
||||
z.object({
|
||||
profileName: z.string().min(1),
|
||||
})
|
||||
)
|
||||
.mutation(async ({ ctx, input }) => {
|
||||
const requestedAt = new Date().toISOString();
|
||||
const result = await ctx.profiles.updateBuildStatus(
|
||||
input.profileName,
|
||||
'QUEUED',
|
||||
{
|
||||
requestedAt,
|
||||
error: null,
|
||||
}
|
||||
);
|
||||
return result;
|
||||
}),
|
||||
setBuildStatus: adminProcedure
|
||||
.input(
|
||||
z.object({
|
||||
profileName: z.string().min(1),
|
||||
status: zBuildStatus,
|
||||
})
|
||||
)
|
||||
.mutation(async ({ ctx, input }) =>
|
||||
ctx.profiles.updateBuildStatus(input.profileName, input.status)
|
||||
),
|
||||
reconcileNow: adminProcedure.mutation(async ({ ctx }) => {
|
||||
await ctx.orchestrator.reconcileNow();
|
||||
return { ok: true };
|
||||
}),
|
||||
}),
|
||||
});
|
||||
@@ -12,6 +12,12 @@ export interface GatewayApiConfig {
|
||||
kakaoAdminKey?: string;
|
||||
kakaoRedirectUri: string;
|
||||
publicBaseUrl: string;
|
||||
adminToken?: string;
|
||||
orchestratorEnabled: boolean;
|
||||
orchestratorReconcileIntervalMs: number;
|
||||
orchestratorScheduleIntervalMs: number;
|
||||
orchestratorBuildIntervalMs: number;
|
||||
workspaceRootHint: string;
|
||||
}
|
||||
|
||||
const parseNumber = (value: string | undefined, fallback: number, label: string): number => {
|
||||
@@ -25,6 +31,20 @@ const parseNumber = (value: string | undefined, fallback: number, label: string)
|
||||
return parsed;
|
||||
};
|
||||
|
||||
const parseBoolean = (value: string | undefined, fallback: boolean): boolean => {
|
||||
if (!value) {
|
||||
return fallback;
|
||||
}
|
||||
const normalized = value.trim().toLowerCase();
|
||||
if (['1', 'true', 'yes', 'y', 'on'].includes(normalized)) {
|
||||
return true;
|
||||
}
|
||||
if (['0', 'false', 'no', 'n', 'off'].includes(normalized)) {
|
||||
return false;
|
||||
}
|
||||
return fallback;
|
||||
};
|
||||
|
||||
export const resolveGatewayApiConfigFromEnv = (
|
||||
env: NodeJS.ProcessEnv = process.env
|
||||
): GatewayApiConfig => {
|
||||
@@ -61,5 +81,23 @@ export const resolveGatewayApiConfigFromEnv = (
|
||||
kakaoAdminKey: env.KAKAO_ADMIN_KEY,
|
||||
kakaoRedirectUri,
|
||||
publicBaseUrl,
|
||||
adminToken: env.GATEWAY_ADMIN_TOKEN,
|
||||
orchestratorEnabled: parseBoolean(env.GATEWAY_ORCHESTRATOR_ENABLED, true),
|
||||
orchestratorReconcileIntervalMs: parseNumber(
|
||||
env.GATEWAY_ORCHESTRATOR_RECONCILE_MS,
|
||||
15000,
|
||||
'GATEWAY_ORCHESTRATOR_RECONCILE_MS'
|
||||
),
|
||||
orchestratorScheduleIntervalMs: parseNumber(
|
||||
env.GATEWAY_ORCHESTRATOR_SCHEDULE_MS,
|
||||
5000,
|
||||
'GATEWAY_ORCHESTRATOR_SCHEDULE_MS'
|
||||
),
|
||||
orchestratorBuildIntervalMs: parseNumber(
|
||||
env.GATEWAY_ORCHESTRATOR_BUILD_MS,
|
||||
10000,
|
||||
'GATEWAY_ORCHESTRATOR_BUILD_MS'
|
||||
),
|
||||
workspaceRootHint: env.GATEWAY_WORKSPACE_ROOT ?? process.cwd(),
|
||||
};
|
||||
};
|
||||
|
||||
@@ -3,6 +3,8 @@ import type { GatewaySessionService } from './auth/sessionService.js';
|
||||
import type { UserRepository } from './auth/userRepository.js';
|
||||
import type { KakaoOAuthClient } from './auth/kakaoClient.js';
|
||||
import type { OAuthSessionStore } from './auth/oauthSessionStore.js';
|
||||
import type { GatewayProfileRepository } from './orchestrator/profileRepository.js';
|
||||
import type { GatewayOrchestratorHandle } from './orchestrator/gatewayOrchestrator.js';
|
||||
|
||||
export interface GatewayApiContext {
|
||||
users: UserRepository;
|
||||
@@ -13,6 +15,10 @@ export interface GatewayApiContext {
|
||||
kakaoClient: KakaoOAuthClient;
|
||||
oauthSessions: OAuthSessionStore;
|
||||
publicBaseUrl: string;
|
||||
profiles: GatewayProfileRepository;
|
||||
orchestrator: GatewayOrchestratorHandle;
|
||||
adminToken?: string;
|
||||
requestHeaders: Record<string, string | string[] | undefined>;
|
||||
}
|
||||
|
||||
export const createGatewayApiContext = (options: {
|
||||
@@ -24,6 +30,10 @@ export const createGatewayApiContext = (options: {
|
||||
kakaoClient: KakaoOAuthClient;
|
||||
oauthSessions: OAuthSessionStore;
|
||||
publicBaseUrl: string;
|
||||
profiles: GatewayProfileRepository;
|
||||
orchestrator: GatewayOrchestratorHandle;
|
||||
adminToken?: string;
|
||||
requestHeaders?: Record<string, string | string[] | undefined>;
|
||||
}): GatewayApiContext => ({
|
||||
users: options.users,
|
||||
sessions: options.sessions,
|
||||
@@ -33,4 +43,8 @@ export const createGatewayApiContext = (options: {
|
||||
kakaoClient: options.kakaoClient,
|
||||
oauthSessions: options.oauthSessions,
|
||||
publicBaseUrl: options.publicBaseUrl,
|
||||
profiles: options.profiles,
|
||||
orchestrator: options.orchestrator,
|
||||
adminToken: options.adminToken,
|
||||
requestHeaders: options.requestHeaders ?? {},
|
||||
});
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
import { spawn } from 'node:child_process';
|
||||
|
||||
export interface BuildCommand {
|
||||
command: string;
|
||||
args: string[];
|
||||
cwd: string;
|
||||
env?: Record<string, string>;
|
||||
}
|
||||
|
||||
export interface BuildResult {
|
||||
ok: boolean;
|
||||
exitCode: number | null;
|
||||
output: string;
|
||||
}
|
||||
|
||||
export interface BuildRunner {
|
||||
run(commands: BuildCommand[]): Promise<BuildResult>;
|
||||
}
|
||||
|
||||
const runCommand = (command: BuildCommand): Promise<BuildResult> =>
|
||||
new Promise((resolve) => {
|
||||
const child = spawn(command.command, command.args, {
|
||||
cwd: command.cwd,
|
||||
env: command.env,
|
||||
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({
|
||||
ok: code === 0,
|
||||
exitCode: code,
|
||||
output,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
export class PnpmBuildRunner implements BuildRunner {
|
||||
async run(commands: BuildCommand[]): Promise<BuildResult> {
|
||||
let mergedOutput = '';
|
||||
for (const command of commands) {
|
||||
const result = await runCommand(command);
|
||||
mergedOutput += result.output;
|
||||
if (!result.ok) {
|
||||
return {
|
||||
ok: false,
|
||||
exitCode: result.exitCode,
|
||||
output: mergedOutput,
|
||||
};
|
||||
}
|
||||
}
|
||||
return {
|
||||
ok: true,
|
||||
exitCode: 0,
|
||||
output: mergedOutput,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,328 @@
|
||||
import path from 'node:path';
|
||||
|
||||
import type { BuildRunner } from './buildRunner.js';
|
||||
import type { ProcessManager } from './processManager.js';
|
||||
import type {
|
||||
GatewayProfileRecord,
|
||||
GatewayProfileRepository,
|
||||
GatewayProfileStatus,
|
||||
} from './profileRepository.js';
|
||||
|
||||
export interface GatewayProcessConfig {
|
||||
workspaceRoot: string;
|
||||
redisKeyPrefix: string;
|
||||
gameTokenSecret: string;
|
||||
baseEnv?: Record<string, string>;
|
||||
}
|
||||
|
||||
export interface GatewayOrchestratorOptions {
|
||||
repository: GatewayProfileRepository;
|
||||
processManager: ProcessManager;
|
||||
buildRunner: BuildRunner;
|
||||
processConfig: GatewayProcessConfig;
|
||||
reconcileIntervalMs: number;
|
||||
scheduleIntervalMs: number;
|
||||
buildIntervalMs: number;
|
||||
now?: () => Date;
|
||||
}
|
||||
|
||||
export interface ProfileRuntimeState {
|
||||
apiRunning: boolean;
|
||||
daemonRunning: boolean;
|
||||
}
|
||||
|
||||
export interface ProfileRuntimeSnapshot extends ProfileRuntimeState {
|
||||
profileName: string;
|
||||
}
|
||||
|
||||
export interface GatewayOrchestratorHandle {
|
||||
start(): void;
|
||||
stop(): Promise<void>;
|
||||
reconcileNow(): Promise<void>;
|
||||
runScheduleNow(): Promise<void>;
|
||||
runBuildQueueNow(): Promise<void>;
|
||||
listRuntimeStates(profileNames: string[]): Promise<ProfileRuntimeSnapshot[]>;
|
||||
}
|
||||
|
||||
export const planProfileReconcile = (
|
||||
status: GatewayProfileStatus,
|
||||
runtime: ProfileRuntimeState
|
||||
): { shouldStart: boolean; shouldStop: boolean } => {
|
||||
if (status === 'RUNNING') {
|
||||
return {
|
||||
shouldStart: !(runtime.apiRunning && runtime.daemonRunning),
|
||||
shouldStop: false,
|
||||
};
|
||||
}
|
||||
return {
|
||||
shouldStart: false,
|
||||
shouldStop: runtime.apiRunning || runtime.daemonRunning,
|
||||
};
|
||||
};
|
||||
|
||||
const buildProcessName = (profileName: string, role: 'api' | 'daemon'): string =>
|
||||
`sammo:${profileName}:${role === 'api' ? 'game-api' : 'turn-daemon'}`;
|
||||
|
||||
const buildProcessDefinitions = (
|
||||
profile: GatewayProfileRecord,
|
||||
config: GatewayProcessConfig
|
||||
): { api: { name: string; script: string; cwd: string; env: Record<string, string> };
|
||||
daemon: { name: string; script: string; cwd: string; env: Record<string, string> } } => {
|
||||
const baseEnv = { ...(config.baseEnv ?? {}) };
|
||||
const apiName = buildProcessName(profile.profileName, 'api');
|
||||
const daemonName = buildProcessName(profile.profileName, 'daemon');
|
||||
const apiCwd = path.join(config.workspaceRoot, 'app', 'game-api');
|
||||
const daemonCwd = path.join(config.workspaceRoot, 'app', 'game-engine');
|
||||
const apiScript = path.join(apiCwd, 'dist', 'index.js');
|
||||
const daemonScript = path.join(daemonCwd, 'dist', 'index.js');
|
||||
const apiEnv = {
|
||||
...baseEnv,
|
||||
PROFILE: profile.profile,
|
||||
SCENARIO: profile.scenario,
|
||||
GAME_API_PORT: String(profile.apiPort),
|
||||
GATEWAY_REDIS_PREFIX: config.redisKeyPrefix,
|
||||
GAME_TOKEN_SECRET: config.gameTokenSecret,
|
||||
};
|
||||
const daemonEnv = {
|
||||
...baseEnv,
|
||||
TURN_PROFILE: profile.profile,
|
||||
PROFILE: profile.profile,
|
||||
SCENARIO: profile.scenario,
|
||||
};
|
||||
return {
|
||||
api: {
|
||||
name: apiName,
|
||||
script: apiScript,
|
||||
cwd: apiCwd,
|
||||
env: apiEnv,
|
||||
},
|
||||
daemon: {
|
||||
name: daemonName,
|
||||
script: daemonScript,
|
||||
cwd: daemonCwd,
|
||||
env: daemonEnv,
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
const mapRuntimeStates = (
|
||||
profileNames: string[],
|
||||
processNames: Map<string, boolean>
|
||||
): ProfileRuntimeSnapshot[] =>
|
||||
profileNames.map((profileName) => {
|
||||
const apiName = buildProcessName(profileName, 'api');
|
||||
const daemonName = buildProcessName(profileName, 'daemon');
|
||||
return {
|
||||
profileName,
|
||||
apiRunning: processNames.get(apiName) ?? false,
|
||||
daemonRunning: processNames.get(daemonName) ?? false,
|
||||
};
|
||||
});
|
||||
|
||||
export class GatewayOrchestrator implements GatewayOrchestratorHandle {
|
||||
private readonly repository: GatewayProfileRepository;
|
||||
private readonly processManager: ProcessManager;
|
||||
private readonly buildRunner: BuildRunner;
|
||||
private readonly processConfig: GatewayProcessConfig;
|
||||
private readonly reconcileIntervalMs: number;
|
||||
private readonly scheduleIntervalMs: number;
|
||||
private readonly buildIntervalMs: number;
|
||||
private readonly now: () => Date;
|
||||
private reconcileTimer?: NodeJS.Timeout;
|
||||
private scheduleTimer?: NodeJS.Timeout;
|
||||
private buildTimer?: NodeJS.Timeout;
|
||||
private reconcileInFlight = false;
|
||||
private scheduleInFlight = false;
|
||||
private buildInFlight = false;
|
||||
|
||||
constructor(options: GatewayOrchestratorOptions) {
|
||||
this.repository = options.repository;
|
||||
this.processManager = options.processManager;
|
||||
this.buildRunner = options.buildRunner;
|
||||
this.processConfig = options.processConfig;
|
||||
this.reconcileIntervalMs = options.reconcileIntervalMs;
|
||||
this.scheduleIntervalMs = options.scheduleIntervalMs;
|
||||
this.buildIntervalMs = options.buildIntervalMs;
|
||||
this.now = options.now ?? (() => new Date());
|
||||
}
|
||||
|
||||
start(): void {
|
||||
void this.reconcileNow();
|
||||
this.reconcileTimer = setInterval(
|
||||
() => void this.reconcileNow(),
|
||||
this.reconcileIntervalMs
|
||||
);
|
||||
this.scheduleTimer = setInterval(
|
||||
() => void this.runScheduleNow(),
|
||||
this.scheduleIntervalMs
|
||||
);
|
||||
this.buildTimer = setInterval(
|
||||
() => void this.runBuildQueueNow(),
|
||||
this.buildIntervalMs
|
||||
);
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
if (this.reconcileTimer) {
|
||||
clearInterval(this.reconcileTimer);
|
||||
}
|
||||
if (this.scheduleTimer) {
|
||||
clearInterval(this.scheduleTimer);
|
||||
}
|
||||
if (this.buildTimer) {
|
||||
clearInterval(this.buildTimer);
|
||||
}
|
||||
}
|
||||
|
||||
async listRuntimeStates(profileNames: string[]): Promise<ProfileRuntimeSnapshot[]> {
|
||||
const processStates = await this.loadProcessStatusMap();
|
||||
return mapRuntimeStates(profileNames, processStates);
|
||||
}
|
||||
|
||||
async reconcileNow(): Promise<void> {
|
||||
if (this.reconcileInFlight) {
|
||||
return;
|
||||
}
|
||||
this.reconcileInFlight = true;
|
||||
try {
|
||||
const profiles = await this.repository.listProfiles();
|
||||
if (!profiles.length) {
|
||||
return;
|
||||
}
|
||||
const processStates = await this.loadProcessStatusMap();
|
||||
for (const profile of profiles) {
|
||||
const runtime = mapRuntimeStates([profile.profileName], processStates)[0];
|
||||
const plan = planProfileReconcile(profile.status, runtime);
|
||||
if (plan.shouldStart) {
|
||||
await this.startProfile(profile);
|
||||
} else if (plan.shouldStop) {
|
||||
await this.stopProfile(profile);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
this.reconcileInFlight = false;
|
||||
}
|
||||
}
|
||||
|
||||
async runScheduleNow(): Promise<void> {
|
||||
if (this.scheduleInFlight) {
|
||||
return;
|
||||
}
|
||||
this.scheduleInFlight = true;
|
||||
try {
|
||||
const now = this.now();
|
||||
const due = await this.repository.listReservedToStart(now);
|
||||
for (const profile of due) {
|
||||
try {
|
||||
await this.repository.updateStatus(
|
||||
profile.profileName,
|
||||
'RUNNING',
|
||||
null
|
||||
);
|
||||
await this.startProfile(profile);
|
||||
} catch (error) {
|
||||
await this.repository.updateLastError(
|
||||
profile.profileName,
|
||||
error instanceof Error ? error.message : 'Failed to start scheduled profile.'
|
||||
);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
this.scheduleInFlight = false;
|
||||
}
|
||||
}
|
||||
|
||||
async runBuildQueueNow(): Promise<void> {
|
||||
if (this.buildInFlight) {
|
||||
return;
|
||||
}
|
||||
this.buildInFlight = true;
|
||||
try {
|
||||
const queued = await this.repository.findQueuedBuild();
|
||||
if (!queued) {
|
||||
return;
|
||||
}
|
||||
const startedAt = this.now().toISOString();
|
||||
await this.repository.updateBuildStatus(queued.profileName, 'RUNNING', {
|
||||
startedAt,
|
||||
error: null,
|
||||
});
|
||||
const result = await this.buildRunner.run([
|
||||
{
|
||||
command: 'pnpm',
|
||||
args: ['--filter', '@sammo-ts/game-api', 'build'],
|
||||
cwd: this.processConfig.workspaceRoot,
|
||||
env: this.processConfig.baseEnv,
|
||||
},
|
||||
{
|
||||
command: 'pnpm',
|
||||
args: ['--filter', '@sammo-ts/game-engine', 'build'],
|
||||
cwd: this.processConfig.workspaceRoot,
|
||||
env: this.processConfig.baseEnv,
|
||||
},
|
||||
]);
|
||||
const completedAt = this.now().toISOString();
|
||||
if (result.ok) {
|
||||
await this.repository.updateBuildStatus(queued.profileName, 'SUCCEEDED', {
|
||||
completedAt,
|
||||
error: null,
|
||||
});
|
||||
if (queued.status !== 'RUNNING' && queued.status !== 'DISABLED') {
|
||||
await this.repository.updateStatus(
|
||||
queued.profileName,
|
||||
'COMPLETED',
|
||||
queued.scheduledStartAt ? queued.scheduledStartAt : null
|
||||
);
|
||||
}
|
||||
} else {
|
||||
await this.repository.updateBuildStatus(queued.profileName, 'FAILED', {
|
||||
completedAt,
|
||||
error: result.output.slice(-4000),
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
this.buildInFlight = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async startProfile(profile: GatewayProfileRecord): Promise<void> {
|
||||
const definitions = buildProcessDefinitions(profile, this.processConfig);
|
||||
try {
|
||||
await this.processManager.start(definitions.api);
|
||||
await this.processManager.start(definitions.daemon);
|
||||
await this.repository.updateLastError(profile.profileName, null);
|
||||
} catch (error) {
|
||||
await this.repository.updateLastError(
|
||||
profile.profileName,
|
||||
error instanceof Error ? error.message : 'Failed to start processes.'
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async stopProfile(profile: GatewayProfileRecord): Promise<void> {
|
||||
const apiName = buildProcessName(profile.profileName, 'api');
|
||||
const daemonName = buildProcessName(profile.profileName, 'daemon');
|
||||
try {
|
||||
await this.processManager.stop(apiName);
|
||||
} catch {
|
||||
await this.processManager.delete(apiName);
|
||||
}
|
||||
try {
|
||||
await this.processManager.stop(daemonName);
|
||||
} catch {
|
||||
await this.processManager.delete(daemonName);
|
||||
}
|
||||
}
|
||||
|
||||
private async loadProcessStatusMap(): Promise<Map<string, boolean>> {
|
||||
const processes = await this.processManager.list();
|
||||
const statusMap = new Map<string, boolean>();
|
||||
for (const process of processes) {
|
||||
const status = process.status.toLowerCase();
|
||||
const running =
|
||||
status === 'online' || status === 'launching' || status === 'stopping';
|
||||
statusMap.set(process.name, running);
|
||||
}
|
||||
return statusMap;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
import { createRequire } from 'node:module';
|
||||
|
||||
import type { ProcessManager, ManagedProcessInfo, ProcessDefinition } from './processManager.js';
|
||||
|
||||
type Pm2Module = typeof import('pm2');
|
||||
|
||||
const require = createRequire(import.meta.url);
|
||||
|
||||
const loadPm2 = (): Pm2Module => require('pm2') as Pm2Module;
|
||||
|
||||
const withPm2 = async <T>(handler: (pm2: Pm2Module) => Promise<T>): Promise<T> => {
|
||||
const pm2 = loadPm2();
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
pm2.connect((error) => {
|
||||
if (error) {
|
||||
reject(error);
|
||||
return;
|
||||
}
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
try {
|
||||
return await handler(pm2);
|
||||
} finally {
|
||||
pm2.disconnect();
|
||||
}
|
||||
};
|
||||
|
||||
export class Pm2ProcessManager implements ProcessManager {
|
||||
async list(): Promise<ManagedProcessInfo[]> {
|
||||
return withPm2(
|
||||
(pm2) =>
|
||||
new Promise<ManagedProcessInfo[]>((resolve, reject) => {
|
||||
pm2.list((error, list) => {
|
||||
if (error) {
|
||||
reject(error);
|
||||
return;
|
||||
}
|
||||
const normalized =
|
||||
list?.map((item) => ({
|
||||
name: item.name ?? 'unknown',
|
||||
status: item.pm2_env?.status ?? 'unknown',
|
||||
pid: item.pid ?? undefined,
|
||||
})) ?? [];
|
||||
resolve(normalized);
|
||||
});
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
async start(definition: ProcessDefinition): Promise<void> {
|
||||
await withPm2(
|
||||
(pm2) =>
|
||||
new Promise<void>((resolve, reject) => {
|
||||
pm2.start(
|
||||
{
|
||||
name: definition.name,
|
||||
script: definition.script,
|
||||
cwd: definition.cwd,
|
||||
args: definition.args,
|
||||
env: definition.env,
|
||||
autorestart: true,
|
||||
time: true,
|
||||
},
|
||||
(error) => {
|
||||
if (error) {
|
||||
reject(error);
|
||||
return;
|
||||
}
|
||||
resolve();
|
||||
}
|
||||
);
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
async stop(name: string): Promise<void> {
|
||||
await withPm2(
|
||||
(pm2) =>
|
||||
new Promise<void>((resolve, reject) => {
|
||||
pm2.stop(name, (error) => {
|
||||
if (error) {
|
||||
reject(error);
|
||||
return;
|
||||
}
|
||||
resolve();
|
||||
});
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
async delete(name: string): Promise<void> {
|
||||
await withPm2(
|
||||
(pm2) =>
|
||||
new Promise<void>((resolve, reject) => {
|
||||
pm2.delete(name, (error) => {
|
||||
if (error) {
|
||||
reject(error);
|
||||
return;
|
||||
}
|
||||
resolve();
|
||||
});
|
||||
})
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
export interface ManagedProcessInfo {
|
||||
name: string;
|
||||
status: string;
|
||||
pid?: number;
|
||||
}
|
||||
|
||||
export interface ProcessDefinition {
|
||||
name: string;
|
||||
script: string;
|
||||
cwd: string;
|
||||
args?: string[];
|
||||
env?: Record<string, string>;
|
||||
}
|
||||
|
||||
export interface ProcessManager {
|
||||
list(): Promise<ManagedProcessInfo[]>;
|
||||
start(definition: ProcessDefinition): Promise<void>;
|
||||
stop(name: string): Promise<void>;
|
||||
delete(name: string): Promise<void>;
|
||||
}
|
||||
@@ -0,0 +1,235 @@
|
||||
import { Prisma, type PrismaClient } from '@prisma/client';
|
||||
|
||||
export const GATEWAY_PROFILE_STATUSES = [
|
||||
'COMPLETED',
|
||||
'RESERVED',
|
||||
'RUNNING',
|
||||
'STOPPED',
|
||||
'DISABLED',
|
||||
] as const;
|
||||
export type GatewayProfileStatus = (typeof GATEWAY_PROFILE_STATUSES)[number];
|
||||
|
||||
export const GATEWAY_BUILD_STATUSES = [
|
||||
'IDLE',
|
||||
'QUEUED',
|
||||
'RUNNING',
|
||||
'FAILED',
|
||||
'SUCCEEDED',
|
||||
] as const;
|
||||
export type GatewayBuildStatus = (typeof GATEWAY_BUILD_STATUSES)[number];
|
||||
|
||||
export interface GatewayProfileRecord {
|
||||
profileName: string;
|
||||
profile: string;
|
||||
scenario: string;
|
||||
apiPort: number;
|
||||
status: GatewayProfileStatus;
|
||||
buildStatus: GatewayBuildStatus;
|
||||
scheduledStartAt?: string;
|
||||
buildRequestedAt?: string;
|
||||
buildStartedAt?: string;
|
||||
buildCompletedAt?: string;
|
||||
buildError?: string;
|
||||
lastError?: string;
|
||||
meta: Prisma.JsonObject;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
export interface GatewayProfileUpsertInput {
|
||||
profile: string;
|
||||
scenario: string;
|
||||
apiPort: number;
|
||||
status?: GatewayProfileStatus;
|
||||
scheduledStartAt?: string;
|
||||
meta?: Prisma.JsonObject;
|
||||
}
|
||||
|
||||
export interface GatewayProfileRepository {
|
||||
listProfiles(): Promise<GatewayProfileRecord[]>;
|
||||
getProfile(profileName: string): Promise<GatewayProfileRecord | null>;
|
||||
upsertProfile(input: GatewayProfileUpsertInput): Promise<GatewayProfileRecord>;
|
||||
updateStatus(
|
||||
profileName: string,
|
||||
status: GatewayProfileStatus,
|
||||
scheduledStartAt?: string | null
|
||||
): Promise<GatewayProfileRecord | null>;
|
||||
updateBuildStatus(
|
||||
profileName: string,
|
||||
status: GatewayBuildStatus,
|
||||
fields?: {
|
||||
requestedAt?: string | null;
|
||||
startedAt?: string | null;
|
||||
completedAt?: string | null;
|
||||
error?: string | null;
|
||||
}
|
||||
): Promise<GatewayProfileRecord | null>;
|
||||
listReservedToStart(now: Date): Promise<GatewayProfileRecord[]>;
|
||||
findQueuedBuild(): Promise<GatewayProfileRecord | null>;
|
||||
updateLastError(profileName: string, lastError: string | null): Promise<void>;
|
||||
}
|
||||
|
||||
const toIso = (value: Date | null): string | undefined =>
|
||||
value ? value.toISOString() : undefined;
|
||||
|
||||
const mapProfile = (row: {
|
||||
profileName: string;
|
||||
profile: string;
|
||||
scenario: string;
|
||||
apiPort: number;
|
||||
status: GatewayProfileStatus;
|
||||
buildStatus: GatewayBuildStatus;
|
||||
scheduledStartAt: Date | null;
|
||||
buildRequestedAt: Date | null;
|
||||
buildStartedAt: Date | null;
|
||||
buildCompletedAt: Date | null;
|
||||
buildError: string | null;
|
||||
lastError: string | null;
|
||||
meta: Prisma.JsonValue;
|
||||
createdAt: Date;
|
||||
updatedAt: Date;
|
||||
}): GatewayProfileRecord => ({
|
||||
profileName: row.profileName,
|
||||
profile: row.profile,
|
||||
scenario: row.scenario,
|
||||
apiPort: row.apiPort,
|
||||
status: row.status,
|
||||
buildStatus: row.buildStatus,
|
||||
scheduledStartAt: toIso(row.scheduledStartAt),
|
||||
buildRequestedAt: toIso(row.buildRequestedAt),
|
||||
buildStartedAt: toIso(row.buildStartedAt),
|
||||
buildCompletedAt: toIso(row.buildCompletedAt),
|
||||
buildError: row.buildError ?? undefined,
|
||||
lastError: row.lastError ?? undefined,
|
||||
meta: (row.meta ?? {}) as Prisma.JsonObject,
|
||||
createdAt: row.createdAt.toISOString(),
|
||||
updatedAt: row.updatedAt.toISOString(),
|
||||
});
|
||||
|
||||
const buildProfileName = (profile: string, scenario: string): string =>
|
||||
`${profile}:${scenario}`;
|
||||
|
||||
export const createGatewayProfileRepository = (
|
||||
prisma: PrismaClient
|
||||
): GatewayProfileRepository => ({
|
||||
async listProfiles(): Promise<GatewayProfileRecord[]> {
|
||||
const rows = await prisma.gatewayProfile.findMany({
|
||||
orderBy: [{ profile: 'asc' }, { scenario: 'asc' }],
|
||||
});
|
||||
return rows.map(mapProfile);
|
||||
},
|
||||
async getProfile(profileName: string): Promise<GatewayProfileRecord | null> {
|
||||
const row = await prisma.gatewayProfile.findUnique({
|
||||
where: { profileName },
|
||||
});
|
||||
return row ? mapProfile(row) : null;
|
||||
},
|
||||
async upsertProfile(input: GatewayProfileUpsertInput): Promise<GatewayProfileRecord> {
|
||||
const profileName = buildProfileName(input.profile, input.scenario);
|
||||
const row = await prisma.gatewayProfile.upsert({
|
||||
where: { profileName },
|
||||
create: {
|
||||
profileName,
|
||||
profile: input.profile,
|
||||
scenario: input.scenario,
|
||||
apiPort: input.apiPort,
|
||||
status: input.status ?? 'STOPPED',
|
||||
scheduledStartAt: input.scheduledStartAt
|
||||
? new Date(input.scheduledStartAt)
|
||||
: null,
|
||||
meta: (input.meta ?? {}) as Prisma.JsonObject,
|
||||
},
|
||||
update: {
|
||||
apiPort: input.apiPort,
|
||||
status: input.status,
|
||||
scheduledStartAt: input.scheduledStartAt
|
||||
? new Date(input.scheduledStartAt)
|
||||
: input.scheduledStartAt === null
|
||||
? null
|
||||
: undefined,
|
||||
meta: input.meta ? (input.meta as Prisma.JsonObject) : undefined,
|
||||
},
|
||||
});
|
||||
return mapProfile(row);
|
||||
},
|
||||
async updateStatus(
|
||||
profileName: string,
|
||||
status: GatewayProfileStatus,
|
||||
scheduledStartAt?: string | null
|
||||
): Promise<GatewayProfileRecord | null> {
|
||||
const row = await prisma.gatewayProfile.update({
|
||||
where: { profileName },
|
||||
data: {
|
||||
status,
|
||||
scheduledStartAt:
|
||||
scheduledStartAt === undefined
|
||||
? undefined
|
||||
: scheduledStartAt
|
||||
? new Date(scheduledStartAt)
|
||||
: null,
|
||||
},
|
||||
});
|
||||
return row ? mapProfile(row) : null;
|
||||
},
|
||||
async updateBuildStatus(
|
||||
profileName: string,
|
||||
status: GatewayBuildStatus,
|
||||
fields?: {
|
||||
requestedAt?: string | null;
|
||||
startedAt?: string | null;
|
||||
completedAt?: string | null;
|
||||
error?: string | null;
|
||||
}
|
||||
): Promise<GatewayProfileRecord | null> {
|
||||
const row = await prisma.gatewayProfile.update({
|
||||
where: { profileName },
|
||||
data: {
|
||||
buildStatus: status,
|
||||
buildRequestedAt:
|
||||
fields?.requestedAt === undefined
|
||||
? undefined
|
||||
: fields?.requestedAt
|
||||
? new Date(fields.requestedAt)
|
||||
: null,
|
||||
buildStartedAt:
|
||||
fields?.startedAt === undefined
|
||||
? undefined
|
||||
: fields?.startedAt
|
||||
? new Date(fields.startedAt)
|
||||
: null,
|
||||
buildCompletedAt:
|
||||
fields?.completedAt === undefined
|
||||
? undefined
|
||||
: fields?.completedAt
|
||||
? new Date(fields.completedAt)
|
||||
: null,
|
||||
buildError: fields?.error === undefined ? undefined : fields.error,
|
||||
},
|
||||
});
|
||||
return row ? mapProfile(row) : null;
|
||||
},
|
||||
async listReservedToStart(now: Date): Promise<GatewayProfileRecord[]> {
|
||||
const rows = await prisma.gatewayProfile.findMany({
|
||||
where: {
|
||||
status: 'RESERVED',
|
||||
scheduledStartAt: {
|
||||
lte: now,
|
||||
},
|
||||
},
|
||||
});
|
||||
return rows.map(mapProfile);
|
||||
},
|
||||
async findQueuedBuild(): Promise<GatewayProfileRecord | null> {
|
||||
const row = await prisma.gatewayProfile.findFirst({
|
||||
where: { buildStatus: 'QUEUED' },
|
||||
orderBy: { buildRequestedAt: 'asc' },
|
||||
});
|
||||
return row ? mapProfile(row) : null;
|
||||
},
|
||||
async updateLastError(profileName: string, lastError: string | null): Promise<void> {
|
||||
await prisma.gatewayProfile.update({
|
||||
where: { profileName },
|
||||
data: { lastError },
|
||||
});
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,25 @@
|
||||
import fs from 'node:fs';
|
||||
import path from 'node:path';
|
||||
|
||||
const WORKSPACE_MARKERS = ['pnpm-workspace.yaml', 'package.json'];
|
||||
|
||||
const hasWorkspaceMarker = (dir: string): boolean =>
|
||||
WORKSPACE_MARKERS.some((marker) => fs.existsSync(path.join(dir, marker)));
|
||||
|
||||
export const resolveWorkspaceRoot = (
|
||||
startDir: string,
|
||||
maxDepth = 5
|
||||
): string => {
|
||||
let current = path.resolve(startDir);
|
||||
for (let depth = 0; depth <= maxDepth; depth += 1) {
|
||||
if (hasWorkspaceMarker(current)) {
|
||||
return current;
|
||||
}
|
||||
const parent = path.dirname(current);
|
||||
if (parent === current) {
|
||||
break;
|
||||
}
|
||||
current = parent;
|
||||
}
|
||||
return path.resolve(startDir);
|
||||
};
|
||||
@@ -9,6 +9,7 @@ import { decryptGameSessionToken, encryptGameSessionToken } from '@sammo-ts/comm
|
||||
import { procedure, router } from './trpc.js';
|
||||
import { toPublicUser } from './auth/userRepository.js';
|
||||
import type { UserOAuthInfo } from './auth/userRepository.js';
|
||||
import { adminRouter } from './adminRouter.js';
|
||||
|
||||
const zUsername = z.string().min(2).max(32);
|
||||
const zPassword = z.string().min(6).max(128);
|
||||
@@ -27,6 +28,7 @@ export const appRouter = router({
|
||||
now: new Date().toISOString(),
|
||||
})),
|
||||
}),
|
||||
admin: adminRouter,
|
||||
auth: router({
|
||||
kakaoStart: procedure
|
||||
.input(
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import fastify from 'fastify';
|
||||
import fastify, { type FastifyRequest } from 'fastify';
|
||||
import cors from '@fastify/cors';
|
||||
import { fastifyTRPCPlugin } from '@trpc/server/adapters/fastify';
|
||||
import type { PrismaClient } from '@prisma/client';
|
||||
@@ -16,8 +16,20 @@ import { KakaoOAuthClient } from './auth/kakaoClient.js';
|
||||
import { RedisOAuthSessionStore } from './auth/oauthSessionStore.js';
|
||||
import { createPostgresUserRepository } from './auth/postgresUserRepository.js';
|
||||
import { RedisGatewaySessionService } from './auth/redisSessionService.js';
|
||||
import { createGatewayProfileRepository } from './orchestrator/profileRepository.js';
|
||||
import { GatewayOrchestrator } from './orchestrator/gatewayOrchestrator.js';
|
||||
import { Pm2ProcessManager } from './orchestrator/pm2ProcessManager.js';
|
||||
import { PnpmBuildRunner } from './orchestrator/buildRunner.js';
|
||||
import { resolveWorkspaceRoot } from './orchestrator/workspaceRoot.js';
|
||||
import { appRouter } from './router.js';
|
||||
|
||||
const buildEnvMap = (env: NodeJS.ProcessEnv): Record<string, string> => {
|
||||
const entries = Object.entries(env).filter(
|
||||
(entry): entry is [string, string] => typeof entry[1] === 'string'
|
||||
);
|
||||
return Object.fromEntries(entries);
|
||||
};
|
||||
|
||||
export const createGatewayApiServer = async () => {
|
||||
const config = resolveGatewayApiConfigFromEnv();
|
||||
const postgres = createPostgresConnector(resolvePostgresConfigFromEnv());
|
||||
@@ -45,6 +57,28 @@ export const createGatewayApiServer = async () => {
|
||||
config.oauthSessionTtlSeconds
|
||||
);
|
||||
|
||||
const profiles = createGatewayProfileRepository(
|
||||
postgres.prisma as PrismaClient
|
||||
);
|
||||
const workspaceRoot = resolveWorkspaceRoot(config.workspaceRootHint);
|
||||
const processManager = new Pm2ProcessManager();
|
||||
const buildRunner = new PnpmBuildRunner();
|
||||
const baseEnv = buildEnvMap(process.env);
|
||||
const orchestrator = new GatewayOrchestrator({
|
||||
repository: profiles,
|
||||
processManager,
|
||||
buildRunner,
|
||||
processConfig: {
|
||||
workspaceRoot,
|
||||
redisKeyPrefix: config.redisKeyPrefix,
|
||||
gameTokenSecret: config.gameTokenSecret,
|
||||
baseEnv,
|
||||
},
|
||||
reconcileIntervalMs: config.orchestratorReconcileIntervalMs,
|
||||
scheduleIntervalMs: config.orchestratorScheduleIntervalMs,
|
||||
buildIntervalMs: config.orchestratorBuildIntervalMs,
|
||||
});
|
||||
|
||||
const app = fastify({
|
||||
logger: true,
|
||||
});
|
||||
@@ -58,7 +92,7 @@ export const createGatewayApiServer = async () => {
|
||||
prefix: config.trpcPath,
|
||||
trpcOptions: {
|
||||
router: appRouter,
|
||||
createContext: () =>
|
||||
createContext: ({ req }: { req: FastifyRequest }) =>
|
||||
createGatewayApiContext({
|
||||
users,
|
||||
sessions,
|
||||
@@ -68,6 +102,10 @@ export const createGatewayApiServer = async () => {
|
||||
kakaoClient,
|
||||
oauthSessions,
|
||||
publicBaseUrl: config.publicBaseUrl,
|
||||
profiles,
|
||||
orchestrator,
|
||||
adminToken: config.adminToken,
|
||||
requestHeaders: req.headers,
|
||||
}),
|
||||
},
|
||||
});
|
||||
@@ -76,7 +114,12 @@ export const createGatewayApiServer = async () => {
|
||||
ok: true,
|
||||
}));
|
||||
|
||||
if (config.orchestratorEnabled) {
|
||||
orchestrator.start();
|
||||
}
|
||||
|
||||
app.addHook('onClose', async () => {
|
||||
await orchestrator.stop();
|
||||
await redis.disconnect();
|
||||
await postgres.disconnect();
|
||||
});
|
||||
|
||||
@@ -36,6 +36,26 @@ const buildCaller = () => {
|
||||
}),
|
||||
sendTalkMessage: async () => {},
|
||||
};
|
||||
const profiles = {
|
||||
listProfiles: async () => [],
|
||||
getProfile: async () => null,
|
||||
upsertProfile: async () => {
|
||||
throw new Error('not used');
|
||||
},
|
||||
updateStatus: async () => null,
|
||||
updateBuildStatus: async () => null,
|
||||
listReservedToStart: async () => [],
|
||||
findQueuedBuild: async () => null,
|
||||
updateLastError: async () => {},
|
||||
};
|
||||
const orchestrator = {
|
||||
start: () => {},
|
||||
stop: async () => {},
|
||||
reconcileNow: async () => {},
|
||||
runScheduleNow: async () => {},
|
||||
runBuildQueueNow: async () => {},
|
||||
listRuntimeStates: async () => [],
|
||||
};
|
||||
const caller = appRouter.createCaller(
|
||||
createGatewayApiContext({
|
||||
users,
|
||||
@@ -46,6 +66,9 @@ const buildCaller = () => {
|
||||
kakaoClient,
|
||||
oauthSessions,
|
||||
publicBaseUrl: 'http://localhost',
|
||||
profiles,
|
||||
orchestrator,
|
||||
requestHeaders: {},
|
||||
})
|
||||
);
|
||||
return { caller, oauthSessions };
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { planProfileReconcile } from '../src/orchestrator/gatewayOrchestrator.js';
|
||||
|
||||
describe('planProfileReconcile', () => {
|
||||
it('starts missing processes for running profiles', () => {
|
||||
expect(
|
||||
planProfileReconcile('RUNNING', {
|
||||
apiRunning: true,
|
||||
daemonRunning: false,
|
||||
})
|
||||
).toEqual({ shouldStart: true, shouldStop: false });
|
||||
});
|
||||
|
||||
it('does nothing when running profile is healthy', () => {
|
||||
expect(
|
||||
planProfileReconcile('RUNNING', {
|
||||
apiRunning: true,
|
||||
daemonRunning: true,
|
||||
})
|
||||
).toEqual({ shouldStart: false, shouldStop: false });
|
||||
});
|
||||
|
||||
it('stops processes for non-running profiles', () => {
|
||||
expect(
|
||||
planProfileReconcile('STOPPED', {
|
||||
apiRunning: false,
|
||||
daemonRunning: true,
|
||||
})
|
||||
).toEqual({ shouldStart: false, shouldStop: true });
|
||||
});
|
||||
|
||||
it('keeps reserved profiles off', () => {
|
||||
expect(
|
||||
planProfileReconcile('RESERVED', {
|
||||
apiRunning: false,
|
||||
daemonRunning: false,
|
||||
})
|
||||
).toEqual({ shouldStart: false, shouldStop: false });
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user