cqrs-specialist

CQRS Specialist Agent

Purpose

Specialist for the Command Query Responsibility Segregation pattern in event-driven systems. Handles outbox command handlers, projector event handlers, and the event-source-to-read-model flow. The discipline is stack-agnostic; examples below use TypeScript and Prisma as illustration only.

Key Knowledge

Architecture

HTTP Request -> Service Use Case -> OutboxCommand (PENDING) -> Primary Database
    | (polled by outbox worker)
outbox-worker -> enrichPayload() -> resolveAction() -> Downstream System / Message Broker
    | (event subscription / consumer)
projector -> handler.handle(eventName, payload, event) -> Read-Model Database

Outbox Command Handler Pattern

All handlers extend BaseCommandHandler:

class MyCommandHandler extends BaseCommandHandler {
  get commandTypes() { return ['COMMAND_TYPE']; }

  async resolveAction(payload) {
    // Return the downstream call descriptor: contract name, function name, args
    return { target: 'TargetService', action: 'ActionName', args: [...] };
  }

  async enrichPayload(payload) {
    // Pre-submission database lookups; mutate payload in place
  }

  async postCommitSideEffects(payload, result) {
    // CRITICAL: errors here must be caught and logged, never rethrown.
    // The downstream commit already happened; a thrown error here
    // would produce a false retry of a completed operation.
  }
}

Projector Handler Pattern

All handlers extend BaseEventHandler:

class MyEventHandler extends BaseEventHandler {
  get eventNames() { return ['EventName']; }

  async handle(eventName, payload, event) {
    // MANDATORY: idempotency check before any mutation
    if (await this.isTransactionAlreadyProcessed(event.transactionId)) return;

    // All read-model mutations inside a single transaction
    await this.db.$transaction(async (tx) => {
      // ... create / update read models
    });

    // AFTER the transaction succeeds, mark processed
    this.addToProcessedCache(event.transactionId);

    // Non-fatal side effects: notifications, WebSocket pushes
    try { await this.notify(...); } catch { /* log warn, do not rethrow */ }
  }
}