feat: add admin scenario operations

2026-07-25 11:38:58 +00:00
parent a83f7a44e3
commit 02d22de72c
19 changed files with 2020 additions and 22 deletions
@@ -7,7 +7,12 @@ import { isRecord } from '@sammo-ts/common';

 import type { BuildCommand, BuildRunner } from './buildRunner.js';
 import type { ProcessManager } from './processManager.js';
-import type { GatewayProfileRecord, GatewayProfileRepository, GatewayProfileStatus } from './profileRepository.js';
+import type {
+    GatewayOperationRecord,
+    GatewayProfileRecord,
+    GatewayProfileRepository,
+    GatewayProfileStatus,
+} from './profileRepository.js';
 import type { GitWorkspaceManager } from './workspaceManager.js';
 import { seedProfileDatabase, type AdminSeedUser } from './seedProfileDatabase.js';

@@ -46,6 +51,7 @@ export interface GatewayOrchestratorHandle {
    reconcileNow(): Promise<void>;
    runScheduleNow(): Promise<void>;
    runBuildQueueNow(): Promise<void>;
+    runOperationsNow(): Promise<void>;
    cleanupStaleWorkspaces(): Promise<{
        removed: string[];
        skipped: string[];
@@ -376,6 +382,7 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle {
    private scheduleInFlight = false;
    private buildInFlight = false;
    private adminActionInFlight = false;
+    private operationInFlight = false;
    private readonly resetInFlight = new Set<string>();

    constructor(options: GatewayOrchestratorOptions) {
@@ -393,11 +400,15 @@ 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.adminActionTimer = setInterval(() => void this.runAdminActionsNow(), this.adminActionIntervalMs);
+        this.adminActionTimer = setInterval(() => {
+            void this.runOperationsNow();
+            void this.runAdminActionsNow();
+        }, this.adminActionIntervalMs);
    }

    async stop(): Promise<void> {
@@ -548,6 +559,83 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle {
        }
    }

+    async runOperationsNow(): Promise<void> {
+        if (this.operationInFlight || this.buildInFlight) {
+            return;
+        }
+        this.operationInFlight = true;
+        try {
+            const operation = await this.repository.claimNextOperation(this.now());
+            if (!operation) {
+                return;
+            }
+            await this.handleOperation(operation);
+        } finally {
+            this.operationInFlight = false;
+        }
+    }
+
+    private async handleOperation(operation: GatewayOperationRecord): Promise<void> {
+        const profile = await this.repository.getProfile(operation.profileName);
+        if (!profile) {
+            await this.repository.completeOperation(operation.id, 'FAILED', {
+                error: 'Profile not found.',
+            });
+            return;
+        }
+        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);
+                if (!started) {
+                    throw new Error('Failed to start profile processes.');
+                }
+                await this.repository.completeOperation(operation.id, 'SUCCEEDED', { error: null });
+                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 });
+                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 payload = normalizeMeta(operation.payload);
+            const install = isRecord(payload.install) ? payload.install : {};
+            const resetAction: GatewayAdminActionRecord = {
+                action: operation.scheduledAt ? 'RESET_SCHEDULED' : 'RESET_NOW',
+                requestedAt: operation.createdAt,
+                scheduledAt: operation.scheduledAt ?? null,
+                reason: operation.reason ?? null,
+                install,
+            };
+            const result = await this.handleResetAction(profile, resetAction, commitSha);
+            if (result.status === 'REQUESTED') {
+                const retryAt = new Date(this.now().getTime() + this.adminActionIntervalMs).toISOString();
+                await this.repository.requeueOperation(operation.id, result.detail, retryAt);
+                return;
+            }
+            if (result.status !== 'APPLIED') {
+                throw new Error(result.detail ?? 'Reset failed.');
+            }
+            await this.repository.completeOperation(operation.id, 'SUCCEEDED', {
+                resolvedCommitSha: commitSha,
+                error: null,
+            });
+        } catch (error) {
+            const detail = error instanceof Error ? error.message : String(error);
+            await this.repository.completeOperation(operation.id, 'FAILED', { error: detail });
+        }
+    }
+
    private async runAdminActionsNow(): Promise<void> {
        if (this.adminActionInFlight) {
            return;
@@ -632,7 +720,8 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle {

    private async handleResetAction(
        profile: GatewayProfileRecord,
-        action: GatewayAdminActionRecord
+        action: GatewayAdminActionRecord,
+        commitShaOverride?: string
    ): Promise<GatewayAdminActionResult> {
        // 리셋 요청을 빌드+재기동 흐름으로 처리한다.
        if (this.resetInFlight.has(profile.profileName)) {
@@ -651,7 +740,7 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle {
            }
        }

-        const commitSha = profile.buildCommitSha;
+        const commitSha = commitShaOverride ?? profile.buildCommitSha;
        if (!commitSha) {
            return { status: 'FAILED', detail: 'buildCommitSha is missing' };
        }
@@ -734,7 +823,10 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle {
                scheduledStartAt: action.scheduledAt ?? null,
            });
            const builtProfile = (await this.repository.getProfile(profile.profileName)) ?? activeProfile;
-            await this.startProfile(builtProfile);
+            const started = await this.startProfile(builtProfile);
+            if (!started) {
+                return { status: 'FAILED', detail: 'reset completed but profile processes failed to start' };
+            }
            return { status: 'APPLIED', detail: 'reset completed via rebuild' };
        } catch (error) {
            const detail = error instanceof Error ? error.message : String(error);
@@ -852,32 +944,39 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle {
        return cutoff;
    }

-    private async startProfile(profile: GatewayProfileRecord): Promise<void> {
+    private async startProfile(profile: GatewayProfileRecord): Promise<boolean> {
        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);
+            return true;
        } catch (error) {
            await this.repository.updateLastError(
                profile.profileName,
                error instanceof Error ? error.message : 'Failed to start processes.'
            );
+            return false;
        }
    }

    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);
+        const failures: string[] = [];
+        for (const name of [apiName, daemonName]) {
+            try {
+                await this.processManager.stop(name);
+            } catch {
+                try {
+                    await this.processManager.delete(name);
+                } catch (error) {
+                    failures.push(`${name}: ${error instanceof Error ? error.message : String(error)}`);
+                }
+            }
        }
-        try {
-            await this.processManager.stop(daemonName);
-        } catch {
-            await this.processManager.delete(daemonName);
+        if (failures.length > 0) {
+            throw new Error(`Failed to stop profile processes: ${failures.join('; ')}`);
        }
    }

@@ -14,6 +14,45 @@ 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 const GATEWAY_OPERATION_TYPES = ['RESET', 'START', 'STOP'] as const;
+export type GatewayOperationType = (typeof GATEWAY_OPERATION_TYPES)[number];
+
+export const GATEWAY_OPERATION_STATUSES = ['QUEUED', 'RUNNING', 'SUCCEEDED', 'FAILED', 'CANCELLED'] as const;
+export type GatewayOperationStatus = (typeof GATEWAY_OPERATION_STATUSES)[number];
+
+export const GATEWAY_SOURCE_MODES = ['BRANCH', 'COMMIT'] as const;
+export type GatewaySourceMode = (typeof GATEWAY_SOURCE_MODES)[number];
+
+export interface GatewayOperationRecord {
+    id: string;
+    profileName: string;
+    type: GatewayOperationType;
+    status: GatewayOperationStatus;
+    sourceMode?: GatewaySourceMode;
+    sourceRef?: string;
+    resolvedCommitSha?: string;
+    payload: GatewayPrisma.JsonObject;
+    reason?: string;
+    requestedBy: string;
+    scheduledAt?: string;
+    startedAt?: string;
+    completedAt?: string;
+    error?: string;
+    createdAt: string;
+    updatedAt: string;
+}
+
+export interface GatewayOperationCreateInput {
+    profileName: string;
+    type: GatewayOperationType;
+    sourceMode?: GatewaySourceMode;
+    sourceRef?: string;
+    payload?: GatewayPrisma.JsonObject;
+    reason?: string;
+    requestedBy: string;
+    scheduledAt?: string;
+}
+
 export interface GatewayProfileRecord {
    profileName: string;
    profile: string;
@@ -82,6 +121,18 @@ export interface GatewayProfileRepository {
    updateLastError(profileName: string, lastError: string | null): Promise<void>;
    updateWorkspaceUsage(profileName: string, workspace: string, lastUsedAt: string): Promise<void>;
    clearWorkspaceUsage(profileNames: string[]): Promise<void>;
+    listOperations(options?: { profileName?: string; limit?: number }): Promise<GatewayOperationRecord[]>;
+    getOperation(id: string): Promise<GatewayOperationRecord | null>;
+    createOperation(input: GatewayOperationCreateInput): Promise<GatewayOperationRecord>;
+    claimNextOperation(now: Date): Promise<GatewayOperationRecord | null>;
+    completeOperation(
+        id: string,
+        status: Extract<GatewayOperationStatus, 'SUCCEEDED' | 'FAILED'>,
+        fields?: { resolvedCommitSha?: string | null; error?: string | null }
+    ): Promise<GatewayOperationRecord>;
+    requeueOperation(id: string, detail?: string, retryAt?: string): Promise<GatewayOperationRecord>;
+    cancelOperation(id: string): Promise<boolean>;
+    retryOperation(id: string, requestedBy: string): Promise<GatewayOperationRecord | null>;
 }

 const toIso = (value: Date | null): string | undefined => (value ? value.toISOString() : undefined);
@@ -109,6 +160,25 @@ type GatewayProfileRow = {
    updatedAt: Date;
 };

+type GatewayOperationRow = {
+    id: string;
+    profileName: string;
+    type: GatewayOperationType;
+    status: GatewayOperationStatus;
+    sourceMode: GatewaySourceMode | null;
+    sourceRef: string | null;
+    resolvedCommitSha: string | null;
+    payload: GatewayPrisma.JsonValue;
+    reason: string | null;
+    requestedBy: string;
+    scheduledAt: Date | null;
+    startedAt: Date | null;
+    completedAt: Date | null;
+    error: string | null;
+    createdAt: Date;
+    updatedAt: Date;
+};
+
 const mapProfile = (row: GatewayProfileRow): GatewayProfileRecord => ({
    profileName: row.profileName,
    profile: row.profile,
@@ -134,6 +204,25 @@ const mapProfile = (row: GatewayProfileRow): GatewayProfileRecord => ({

 const buildProfileName = (profile: string, scenario: string): string => `${profile}:${scenario}`;

+const mapOperation = (row: GatewayOperationRow): GatewayOperationRecord => ({
+    id: row.id,
+    profileName: row.profileName,
+    type: row.type,
+    status: row.status,
+    sourceMode: row.sourceMode ?? undefined,
+    sourceRef: row.sourceRef ?? undefined,
+    resolvedCommitSha: row.resolvedCommitSha ?? undefined,
+    payload: (row.payload ?? {}) as GatewayPrisma.JsonObject,
+    reason: row.reason ?? undefined,
+    requestedBy: row.requestedBy,
+    scheduledAt: toIso(row.scheduledAt),
+    startedAt: toIso(row.startedAt),
+    completedAt: toIso(row.completedAt),
+    error: row.error ?? undefined,
+    createdAt: row.createdAt.toISOString(),
+    updatedAt: row.updatedAt.toISOString(),
+});
+
 export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): GatewayProfileRepository => ({
    async listProfiles(): Promise<GatewayProfileRecord[]> {
        const rows = await prisma.gatewayProfile.findMany({
@@ -327,4 +416,111 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat
            },
        });
    },
+    async listOperations(options?: { profileName?: string; limit?: number }): Promise<GatewayOperationRecord[]> {
+        const rows = await prisma.gatewayOperation.findMany({
+            where: options?.profileName ? { profileName: options.profileName } : undefined,
+            orderBy: { createdAt: 'desc' },
+            take: Math.min(Math.max(options?.limit ?? 50, 1), 200),
+        });
+        return rows.map(mapOperation);
+    },
+    async getOperation(id: string): Promise<GatewayOperationRecord | null> {
+        const row = await prisma.gatewayOperation.findUnique({ where: { id } });
+        return row ? mapOperation(row) : null;
+    },
+    async createOperation(input: GatewayOperationCreateInput): Promise<GatewayOperationRecord> {
+        const row = await prisma.gatewayOperation.create({
+            data: {
+                profileName: input.profileName,
+                type: input.type,
+                sourceMode: input.sourceMode,
+                sourceRef: input.sourceRef,
+                payload: (input.payload ?? {}) as GatewayPrisma.JsonObject,
+                reason: input.reason,
+                requestedBy: input.requestedBy,
+                scheduledAt: input.scheduledAt ? new Date(input.scheduledAt) : null,
+            },
+        });
+        return mapOperation(row);
+    },
+    async claimNextOperation(now: Date): Promise<GatewayOperationRecord | null> {
+        const row = await prisma.$transaction(async (tx) => {
+            const candidate = 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 },
+            });
+            if (claimed.count !== 1) {
+                return null;
+            }
+            return tx.gatewayOperation.findUnique({ where: { id: candidate.id } });
+        });
+        return row ? mapOperation(row) : null;
+    },
+    async completeOperation(
+        id: string,
+        status: Extract<GatewayOperationStatus, 'SUCCEEDED' | 'FAILED'>,
+        fields?: { resolvedCommitSha?: string | null; error?: string | null }
+    ): Promise<GatewayOperationRecord> {
+        const row = await prisma.gatewayOperation.update({
+            where: { id },
+            data: {
+                status,
+                completedAt: new Date(),
+                resolvedCommitSha:
+                    fields?.resolvedCommitSha === undefined ? undefined : fields.resolvedCommitSha,
+                error: fields?.error === undefined ? undefined : fields.error,
+            },
+        });
+        return mapOperation(row);
+    },
+    async requeueOperation(id: string, detail?: string, retryAt?: string): Promise<GatewayOperationRecord> {
+        const row = await prisma.gatewayOperation.update({
+            where: { id },
+            data: {
+                status: 'QUEUED',
+                startedAt: null,
+                error: detail,
+                scheduledAt: retryAt ? new Date(retryAt) : undefined,
+            },
+        });
+        return mapOperation(row);
+    },
+    async cancelOperation(id: string): Promise<boolean> {
+        const result = await prisma.gatewayOperation.updateMany({
+            where: { id, status: 'QUEUED' },
+            data: { status: 'CANCELLED', completedAt: new Date() },
+        });
+        return result.count === 1;
+    },
+    async retryOperation(id: string, requestedBy: string): Promise<GatewayOperationRecord | null> {
+        const row = await prisma.$transaction(async (tx) => {
+            const previous = await tx.gatewayOperation.findUnique({ where: { id } });
+            if (!previous || (previous.status !== 'FAILED' && previous.status !== 'CANCELLED')) {
+                return null;
+            }
+            return tx.gatewayOperation.create({
+                data: {
+                    profileName: previous.profileName,
+                    type: previous.type,
+                    sourceMode: previous.sourceMode,
+                    sourceRef: previous.sourceRef,
+                    payload: previous.payload as GatewayPrisma.JsonObject,
+                    reason: previous.reason,
+                    requestedBy,
+                    scheduledAt: null,
+                },
+            });
+        });
+        return row ? mapOperation(row) : null;
+    },
 });
@@ -40,6 +40,15 @@ const ensureDir = (dir: string): void => {
 };

 const hasInstallMarker = (dir: string): boolean => fs.existsSync(path.join(dir, 'node_modules', '.pnpm'));
+const GIT_REF_PATTERN = /^[0-9A-Za-z._/-]+$/;
+
+const assertGitRef = (value: string): string => {
+    const ref = value.trim();
+    if (!ref || ref.startsWith('-') || ref.includes('..') || !GIT_REF_PATTERN.test(ref)) {
+        throw new Error('Invalid git ref.');
+    }
+    return ref;
+};

 export class GitWorkspaceManager {
    private readonly repoRoot: string;
@@ -52,6 +61,28 @@ export class GitWorkspaceManager {
        this.baseEnv = options.baseEnv;
    }

+    async resolveCommit(sourceMode: 'BRANCH' | 'COMMIT', sourceRef: string): Promise<string> {
+        const ref = assertGitRef(sourceRef);
+        if (sourceMode === 'BRANCH') {
+            const fetched = await runGit(['fetch', '--all', '--prune'], this.repoRoot, this.baseEnv);
+            if (!fetched.ok) {
+                throw new Error(fetched.output || 'Failed to fetch git branches.');
+            }
+        }
+        const candidates =
+            sourceMode === 'BRANCH'
+                ? [`refs/remotes/origin/${ref}^{commit}`, `refs/heads/${ref}^{commit}`]
+                : [`${ref}^{commit}`];
+        for (const candidate of candidates) {
+            const result = await runGit(['rev-parse', '--verify', candidate], this.repoRoot, this.baseEnv);
+            const commitSha = result.output.trim().split('\n')[0];
+            if (result.ok && /^[0-9a-f]{40}$/i.test(commitSha)) {
+                return commitSha;
+            }
+        }
+        throw new Error(`${sourceMode === 'BRANCH' ? 'Branch' : 'Commit'} not found.`);
+    }
+
    async prepare(commitSha: string): Promise<WorkspaceInfo> {
        const workspacePath = path.join(this.worktreeRoot, commitSha);
        ensureDir(this.worktreeRoot);