feat: 턴 데몬 명령 처리 및 결과 응답 기능 추가

2026-01-04 16:05:22 +00:00
parent b2a625d4e7
commit 2753b85a26
12 changed files with 716 additions and 63 deletions
@@ -0,0 +1,236 @@
+import type {
+    TurnDaemonCommand,
+    TurnDaemonCommandResponder,
+    TurnDaemonControlQueue,
+    TurnDaemonStatus,
+    TurnDaemonCommandResult,
+} from './types.js';
+
+export interface TurnDaemonStreamKeys {
+    commandStream: string;
+    eventStream: string;
+}
+
+export const buildTurnDaemonStreamKeys = (
+    profileName: string
+): TurnDaemonStreamKeys => ({
+    commandStream: `sammo:${profileName}:turn-daemon:commands`,
+    eventStream: `sammo:${profileName}:turn-daemon:events`,
+});
+
+interface RedisStreamClient {
+    xAdd(stream: string, id: string, message: Record<string, string>): Promise<string>;
+    xRead(
+        streams: { key: string; id: string },
+        options?: { BLOCK?: number; COUNT?: number }
+    ): Promise<unknown>;
+}
+
+type RedisStreamReadResponse = Array<{
+    name: string;
+    messages: Array<{ id: string; message: Record<string, string> }>;
+}>;
+
+type TurnDaemonEvent =
+    | { type: 'status'; requestId?: string; status: TurnDaemonStatus }
+    | { type: 'commandResult'; result: TurnDaemonCommandResult };
+
+type TurnDaemonCommandEnvelope = {
+    requestId: string;
+    sentAt: string;
+    command: TurnDaemonCommand;
+};
+
+type TurnDaemonEventEnvelope = {
+    requestId?: string;
+    sentAt: string;
+    event: TurnDaemonEvent;
+};
+
+const parseCommandEnvelope = (raw: string): TurnDaemonCommandEnvelope | null => {
+    try {
+        const parsed = JSON.parse(raw) as Partial<TurnDaemonCommandEnvelope>;
+        if (!parsed || typeof parsed !== 'object') {
+            return null;
+        }
+        if (!parsed.command || typeof parsed.command !== 'object') {
+            return null;
+        }
+        if (typeof parsed.requestId !== 'string') {
+            return null;
+        }
+        if (typeof parsed.sentAt !== 'string') {
+            return null;
+        }
+        return parsed as TurnDaemonCommandEnvelope;
+    } catch {
+        return null;
+    }
+};
+
+const normalizeCommand = (
+    envelope: TurnDaemonCommandEnvelope
+): TurnDaemonCommand | null => {
+    const command = envelope.command as TurnDaemonCommand & {
+        requestId?: string;
+    };
+    switch (command.type) {
+        case 'troopJoin': {
+            if (
+                typeof command.generalId !== 'number' ||
+                typeof command.troopId !== 'number'
+            ) {
+                return null;
+            }
+            return {
+                type: 'troopJoin',
+                requestId: envelope.requestId,
+                generalId: command.generalId,
+                troopId: command.troopId,
+            };
+        }
+        case 'troopExit': {
+            if (typeof command.generalId !== 'number') {
+                return null;
+            }
+            return {
+                type: 'troopExit',
+                requestId: envelope.requestId,
+                generalId: command.generalId,
+            };
+        }
+        case 'getStatus': {
+            const requestId =
+                typeof command.requestId === 'string'
+                    ? command.requestId
+                    : envelope.requestId;
+            return { type: 'getStatus', requestId };
+        }
+        case 'run':
+        case 'pause':
+        case 'resume':
+        case 'shutdown':
+            return command;
+        default:
+            return null;
+    }
+};
+
+export class RedisTurnDaemonCommandStream
+    implements TurnDaemonControlQueue, TurnDaemonCommandResponder
+{
+    private readonly client: RedisStreamClient;
+    private readonly keys: TurnDaemonStreamKeys;
+    private readonly localQueue: TurnDaemonCommand[] = [];
+    private lastId: string;
+
+    constructor(
+        client: RedisStreamClient,
+        options: {
+            keys: TurnDaemonStreamKeys;
+            startId?: string;
+        }
+    ) {
+        this.client = client;
+        this.keys = options.keys;
+        this.lastId = options.startId ?? '$';
+    }
+
+    enqueue(command: TurnDaemonCommand): void {
+        this.localQueue.push(command);
+    }
+
+    async drain(): Promise<TurnDaemonCommand[]> {
+        const drained = this.localQueue.splice(0, this.localQueue.length);
+        const remote = await this.readRemoteCommands(1);
+        return drained.concat(remote);
+    }
+
+    async waitUntil(deadlineMs: number | null): Promise<TurnDaemonCommand | null> {
+        if (this.localQueue.length > 0) {
+            return this.localQueue.shift() ?? null;
+        }
+
+        const blockMs =
+            deadlineMs === null
+                ? 0
+                : Math.max(0, deadlineMs - Date.now());
+        if (deadlineMs !== null && blockMs === 0) {
+            return null;
+        }
+
+        const remote = await this.readRemoteCommands(blockMs);
+        if (remote.length === 0) {
+            return null;
+        }
+        const [first, ...rest] = remote;
+        if (rest.length > 0) {
+            this.localQueue.push(...rest);
+        }
+        return first ?? null;
+    }
+
+    getDepth(): number {
+        return this.localQueue.length;
+    }
+
+    async publishStatus(
+        requestId: string,
+        status: TurnDaemonStatus
+    ): Promise<void> {
+        await this.publishEvent({ type: 'status', requestId, status }, requestId);
+    }
+
+    async publishCommandResult(
+        requestId: string,
+        result: TurnDaemonCommandResult
+    ): Promise<void> {
+        await this.publishEvent({ type: 'commandResult', result }, requestId);
+    }
+
+    private async publishEvent(
+        event: TurnDaemonEvent,
+        requestId?: string
+    ): Promise<void> {
+        const envelope: TurnDaemonEventEnvelope = {
+            requestId,
+            sentAt: new Date().toISOString(),
+            event,
+        };
+        await this.client.xAdd(this.keys.eventStream, '*', {
+            payload: JSON.stringify(envelope),
+        });
+    }
+
+    private async readRemoteCommands(blockMs: number): Promise<TurnDaemonCommand[]> {
+        const response = (await this.client.xRead(
+            { key: this.keys.commandStream, id: this.lastId },
+            { BLOCK: blockMs, COUNT: 100 }
+        )) as RedisStreamReadResponse | null;
+
+        if (!response) {
+            return [];
+        }
+
+        const commands: TurnDaemonCommand[] = [];
+        for (const stream of response) {
+            for (const message of stream.messages) {
+                this.lastId = message.id;
+                const payload = message.message.payload;
+                if (!payload) {
+                    continue;
+                }
+                const envelope = parseCommandEnvelope(payload);
+                if (!envelope) {
+                    continue;
+                }
+                const command = normalizeCommand(envelope);
+                if (!command) {
+                    continue;
+                }
+                commands.push(command);
+            }
+        }
+        return commands;
+    }
+}
@@ -11,6 +11,9 @@ import type {
    TurnProcessor,
    Clock,
    TurnCheckpoint,
+    TurnDaemonCommandHandler,
+    TurnDaemonCommandResponder,
+    TurnDaemonCommandResult,
 } from './types.js';

 type PendingRun = {
@@ -31,6 +34,8 @@ export interface TurnDaemonLifecycleDeps {
    stateStore: TurnStateStore;
    processor: TurnProcessor;
    hooks?: TurnDaemonHooks;
+    commandHandler?: TurnDaemonCommandHandler;
+    commandResponder?: TurnDaemonCommandResponder;
    pauseGate?: () => Promise<boolean>;
 }

@@ -42,6 +47,8 @@ export class TurnDaemonLifecycle {
    private readonly stateStore: TurnStateStore;
    private readonly processor: TurnProcessor;
    private readonly hooks?: TurnDaemonHooks;
+    private readonly commandHandler?: TurnDaemonCommandHandler;
+    private readonly commandResponder?: TurnDaemonCommandResponder;
    private readonly pauseGate?: () => Promise<boolean>;
    private readonly options: TurnDaemonLifecycleOptions;

@@ -59,6 +66,8 @@ export class TurnDaemonLifecycle {
        this.stateStore = deps.stateStore;
        this.processor = deps.processor;
        this.hooks = deps.hooks;
+        this.commandHandler = deps.commandHandler;
+        this.commandResponder = deps.commandResponder;
        this.pauseGate = deps.pauseGate;
        this.options = options;
        this.status = {
@@ -218,6 +227,13 @@ export class TurnDaemonLifecycle {
                this.status.state = 'stopping';
                this.stopping = true;
                return;
+            case 'getStatus': {
+                await this.commandResponder?.publishStatus(
+                    command.requestId,
+                    this.getStatus()
+                );
+                return;
+            }
            case 'run':
                this.pendingRun = {
                    reason: command.reason,
@@ -226,6 +242,54 @@ export class TurnDaemonLifecycle {
                };
                this.status.pendingReason = command.reason;
                return;
+            case 'troopJoin':
+            case 'troopExit':
+                await this.handleMutationCommand(command);
+                return;
+        }
+    }
+
+    private async handleMutationCommand(
+        command: Extract<TurnDaemonCommand, { type: 'troopJoin' | 'troopExit' }>
+    ): Promise<void> {
+        let result: TurnDaemonCommandResult | null = null;
+        try {
+            result = this.commandHandler
+                ? await this.commandHandler.handle(command)
+                : null;
+            if (!result) {
+                result = {
+                    type: command.type,
+                    ok: false,
+                    generalId: command.generalId,
+                    ...(command.type === 'troopJoin'
+                        ? {
+                              troopId: command.troopId,
+                              reason: '턴 데몬이 명령을 처리할 수 없습니다.',
+                          }
+                        : {
+                              reason: '턴 데몬이 명령을 처리할 수 없습니다.',
+                          }),
+                } as TurnDaemonCommandResult;
+            }
+        } catch (error) {
+            const reason =
+                error instanceof Error ? error.message : 'Unknown command error.';
+            result = {
+                type: command.type,
+                ok: false,
+                generalId: command.generalId,
+                ...(command.type === 'troopJoin'
+                    ? { troopId: command.troopId, reason }
+                    : { reason }),
+            } as TurnDaemonCommandResult;
+        }
+
+        if (this.commandResponder) {
+            await this.commandResponder.publishCommandResult(
+                command.requestId,
+                result
+            );
        }
    }

@@ -42,7 +42,49 @@ export type TurnDaemonCommand =
    | { type: 'run'; reason: RunReason; targetTime?: string; budget?: TurnRunBudget }
    | { type: 'pause'; reason?: string }
    | { type: 'resume'; reason?: string }
-    | { type: 'shutdown'; reason?: string };
+    | { type: 'shutdown'; reason?: string }
+    | { type: 'getStatus'; requestId: string }
+    | { type: 'troopJoin'; requestId: string; generalId: number; troopId: number }
+    | { type: 'troopExit'; requestId: string; generalId: number };
+
+export type TurnDaemonCommandResult =
+    | {
+          type: 'troopJoin';
+          ok: true;
+          generalId: number;
+          troopId: number;
+      }
+    | {
+          type: 'troopJoin';
+          ok: false;
+          generalId: number;
+          troopId: number;
+          reason: string;
+      }
+    | {
+          type: 'troopExit';
+          ok: true;
+          generalId: number;
+          wasLeader: boolean;
+      }
+    | {
+          type: 'troopExit';
+          ok: false;
+          generalId: number;
+          reason: string;
+      };
+
+export interface TurnDaemonCommandHandler {
+    handle(command: TurnDaemonCommand): Promise<TurnDaemonCommandResult | null>;
+}
+
+export interface TurnDaemonCommandResponder {
+    publishStatus(requestId: string, status: TurnDaemonStatus): Promise<void>;
+    publishCommandResult(
+        requestId: string,
+        result: TurnDaemonCommandResult
+    ): Promise<void>;
+}

 export type { Clock } from '@sammo-ts/common';