7장: 이벤트 소싱과 CQRS 패턴

[전통적 CRUD]
사용자 테이블: {id: 1, name: "홍길동", email: "hong@mail.com"}
→ 이메일 변경 → {id: 1, name: "홍길동", email: "new@mail.com"}
→ 이전 이메일은 사라짐
 
[이벤트 소싱]
이벤트 1: UserCreated {name: "홍길동", email: "hong@mail.com"}
이벤트 2: EmailChanged {oldEmail: "hong@mail.com", newEmail: "new@mail.com"}
→ 모든 변경 이력이 보존됨

// 대화 이벤트 타입 정의
type ConversationEvent =
  | {
      type: "ConversationStarted";
      timestamp: number;
      userId: string;
      sessionId: string;
      model: string;
    }
  | {
      type: "UserMessageSent";
      timestamp: number;
      content: string;
      attachments?: string[];
    }
  | {
      type: "AssistantStreamStarted";
      timestamp: number;
      model: string;
      promptTokens: number;
    }
  | {
      type: "AssistantTokenGenerated";
      timestamp: number;
      token: string;
      index: number;
    }
  | {
      type: "AssistantStreamCompleted";
      timestamp: number;
      totalTokens: number;
      finishReason: string;
      latencyMs: number;
    }
  | {
      type: "GenerationCancelled";
      timestamp: number;
      tokensGenerated: number;
      reason: "user" | "timeout" | "error";
    }
  | {
      type: "FeedbackGiven";
      timestamp: number;
      messageIndex: number;
      rating: "positive" | "negative";
      comment?: string;
    }
  | {
      type: "ToolCallRequested";
      timestamp: number;
      toolName: string;
      arguments: Record<string, unknown>;
    }
  | {
      type: "ToolCallCompleted";
      timestamp: number;
      toolName: string;
      result: unknown;
      durationMs: number;
    };

interface ConversationState {
  sessionId: string;
  messages: Array<{
    role: "user" | "assistant";
    content: string;
    metadata: Record<string, unknown>;
  }>;
  isStreaming: boolean;
  currentStreamBuffer: string;
  totalTokensUsed: number;
  toolCalls: Array<{
    name: string;
    status: "pending" | "completed";
  }>;
}
 
function projectConversation(
  events: ConversationEvent[]
): ConversationState {
  const state: ConversationState = {
    sessionId: "",
    messages: [],
    isStreaming: false,
    currentStreamBuffer: "",
    totalTokensUsed: 0,
    toolCalls: [],
  };
 
  for (const event of events) {
    switch (event.type) {
      case "ConversationStarted":
        state.sessionId = event.sessionId;
        break;
 
      case "UserMessageSent":
        state.messages.push({
          role: "user",
          content: event.content,
          metadata: { attachments: event.attachments },
        });
        break;
 
      case "AssistantStreamStarted":
        state.isStreaming = true;
        state.currentStreamBuffer = "";
        break;
 
      case "AssistantTokenGenerated":
        state.currentStreamBuffer += event.token;
        break;
 
      case "AssistantStreamCompleted":
        state.isStreaming = false;
        state.messages.push({
          role: "assistant",
          content: state.currentStreamBuffer,
          metadata: {
            tokens: event.totalTokens,
            latencyMs: event.latencyMs,
          },
        });
        state.currentStreamBuffer = "";
        state.totalTokensUsed += event.totalTokens;
        break;
 
      case "GenerationCancelled":
        state.isStreaming = false;
        if (state.currentStreamBuffer) {
          state.messages.push({
            role: "assistant",
            content: state.currentStreamBuffer + " [중단됨]",
            metadata: { cancelled: true },
          });
        }
        state.currentStreamBuffer = "";
        break;
 
      case "ToolCallRequested":
        state.toolCalls.push({
          name: event.toolName,
          status: "pending",
        });
        break;
 
      case "ToolCallCompleted": {
        const call = state.toolCalls.find(
          (c) =>
            c.name === event.toolName && c.status === "pending"
        );
        if (call) call.status = "completed";
        break;
      }
    }
  }
 
  return state;
}

type AgentEvent =
  | {
      type: "TaskReceived";
      timestamp: number;
      taskDescription: string;
      constraints: string[];
    }
  | {
      type: "PlanCreated";
      timestamp: number;
      steps: string[];
      estimatedDuration: number;
    }
  | {
      type: "StepStarted";
      timestamp: number;
      stepIndex: number;
      description: string;
    }
  | {
      type: "ToolInvoked";
      timestamp: number;
      tool: string;
      input: Record<string, unknown>;
    }
  | {
      type: "ToolResultReceived";
      timestamp: number;
      tool: string;
      output: unknown;
      success: boolean;
    }
  | {
      type: "ReplanningTriggered";
      timestamp: number;
      reason: string;
      newSteps: string[];
    }
  | {
      type: "TaskCompleted";
      timestamp: number;
      result: unknown;
      totalSteps: number;
      totalDurationMs: number;
    };
 
// 에이전트 실행 추적기
class AgentTracer {
  private events: AgentEvent[] = [];
  private eventStore: EventStore;
 
  constructor(
    private agentId: string,
    eventStore: EventStore
  ) {
    this.eventStore = eventStore;
  }
 
  async record(event: AgentEvent) {
    this.events.push(event);
    await this.eventStore.append(
      `agent-${this.agentId}`,
      event
    );
  }
 
  // 특정 시점의 에이전트 상태 조회
  getStateAt(timestamp: number): AgentState {
    const relevantEvents = this.events.filter(
      (e) => e.timestamp <= timestamp
    );
    return projectAgentState(relevantEvents);
  }
}

class TimeTravelDebugger {
  constructor(private eventStore: EventStore) {}
 
  /**
   * 특정 시점의 대화 상태를 재현합니다.
   * AI가 왜 특정 응답을 했는지 분석할 때 사용합니다.
   */
  async replayTo(
    streamId: string,
    targetTimestamp: number
  ): Promise<ConversationState> {
    const allEvents = await this.eventStore.readStream(streamId);
 
    // 목표 시점까지의 이벤트만 선택
    const eventsUntil = allEvents.filter(
      (e) => e.timestamp <= targetTimestamp
    );
 
    // 이벤트를 순서대로 재생하여 상태 복원
    return projectConversation(eventsUntil);
  }
 
  /**
   * 두 시점 사이에 발생한 이벤트를 조회합니다.
   * "이 시간대에 무슨 일이 있었는지" 분석할 때 사용합니다.
   */
  async getEventsBetween(
    streamId: string,
    from: number,
    to: number
  ): Promise<ConversationEvent[]> {
    const allEvents = await this.eventStore.readStream(streamId);
    return allEvents.filter(
      (e) => e.timestamp >= from && e.timestamp <= to
    );
  }
 
  /**
   * 에이전트의 의사결정 과정을 시각화합니다.
   */
  async traceDecision(
    agentId: string,
    decisionTimestamp: number
  ): Promise<DecisionTrace> {
    const events = await this.eventStore.readStream(
      `agent-${agentId}`
    );
 
    // 해당 결정 직전의 컨텍스트 수집
    const contextEvents = events.filter(
      (e) => e.timestamp < decisionTimestamp
    );
 
    // 결정 이후의 결과 수집
    const outcomeEvents = events.filter(
      (e) =>
        e.timestamp >= decisionTimestamp &&
        e.timestamp < decisionTimestamp + 60000
    );
 
    return {
      context: contextEvents,
      decision: events.find(
        (e) => e.timestamp === decisionTimestamp
      ),
      outcomes: outcomeEvents,
    };
  }
}

conversations-events     — 대화 이벤트 (파티션: sessionId 기준)
agent-events             — 에이전트 실행 이벤트
inference-metrics        — 추론 성능 메트릭
feedback-events          — 사용자 피드백

// 감사 로그 쿼리 예시
 
// 1. 특정 사용자의 모든 AI 상호작용 조회
const userInteractions = await eventStore.readStream(
  `user-${userId}`
);
 
// 2. 특정 기간의 생성 중단 이벤트 집계
const cancellations = await eventStore.query({
  eventType: "GenerationCancelled",
  fromTimestamp: startOfDay,
  toTimestamp: endOfDay,
});
 
// 3. 도구 호출 실패 패턴 분석
const failedToolCalls = await eventStore.query({
  eventType: "ToolResultReceived",
  filter: { success: false },
  fromTimestamp: lastWeek,
});
 
// 4. 특정 모델 버전의 응답 품질 추적
const modelResponses = await eventStore.query({
  eventType: "AssistantStreamCompleted",
  metadata: { model: "claude-sonnet-4-20250514" },
});