던플 개발기 (7): 메시징 백본 — 명령은 RabbitMQ, 이벤트는 Kafka, 그 사이 transactional outbox

RabbitMQ = do this command/job
Kafka    = this market fact happened

flowchart LR
  SCHED[운영 tick / dispatcher] -->|명령| CMDX[RabbitMQ<br/>market.command exchange]
  CMDX --> WORK[market.command.work 큐]
  WORK --> MW[market-worker]
  MW -->|호출| NEO[Neople Open API]
  MW -->|같은 트랜잭션| DB[(market PostgreSQL<br/>snapshot + outbox_events)]
  DB --> RELAY[outbox relay]
  RELAY -->|이벤트| KAFKA[Kafka<br/>market.* topics]
  KAFKA --> RT[realtime-gateway]
  KAFKA --> RANK[ranking worker]
  KAFKA --> ALERT[alert worker]
  RT -->|WebSocket/SSE| U[Browser]

// libs/shared/messaging-rabbitmq/src/lib/envelope.ts
export const COMMAND_PAYLOAD_SCHEMAS = {
  'market.collect-item.requested': z.object({
    itemId: z.string().min(1),
    // 발행 시점의 수집 tier (도메인 의존을 피하려 free-form)
    tier: z.string().min(1),
    reason,
    // next_collect_at에서 복사한 ISO-8601 due time; advisory only
    dueAt: z.string().datetime({ offset: true }),
  }),
  'provider.probe.requested': z.object({
    providerId: z.string().min(1),
    reason,
  }),
  // ...item.catalog.refresh / market.rollup / market.purge ...
} as const satisfies Record<CommandType, z.ZodTypeAny>;

// libs/shared/messaging-rabbitmq/src/lib/topology.ts
function domainTopology(domain, routingKeys) {
  return {
    domain,
    exchange: `${domain}.command`, // topic exchange
    workQueue: `${domain}.command.work`, // consumer가 읽는 durable 큐
    retryExchange: `${domain}.command.retry`,
    retryQueue: `${domain}.command.retry.delay`, // TTL 후 work로 재전달
    deadLetterExchange: `${domain}.command.dlx`,
    deadLetterQueue: `${domain}.command.dlq`, // poison·소진 메시지 parking
    routingKeys,
  };
}

flowchart LR
  P[publisher<br/>confirm 대기] --> EX[market.command<br/>topic exchange]
  EX -->|routing key = commandType| WQ[market.command.work]
  WQ --> C{consumer<br/>disposition}
  C -->|ACK / RESCHEDULE| DONE[channel.ack]
  C -->|RETRY · 예산 남음| RX[market.command.retry]
  RX -->|x-message-ttl 후| EX
  C -->|DROP · 예산 소진 · poison| DLX[market.command.dlx]
  DLX --> DLQ[market.command.dlq<br/>parking]

// libs/shared/messaging-rabbitmq/src/lib/retry-policy.ts
export function decideRetryAction(input: RetryDecisionInput): RetryAction {
  switch (input.disposition) {
    case 'ACK':
    case 'RESCHEDULE':
      // 점검(RESCHEDULE)은 핸들러가 이미 next_collect_at을 미뤘으므로
      // 그냥 ack. 절대 dead-letter하지 않는다.
      return { kind: 'ack' };
    case 'DROP':
      return { kind: 'park', reason: 'poison' };
    case 'RETRY': {
      if (input.attempt + 1 >= input.maxAttempts) {
        return { kind: 'park', reason: 'attempts-exhausted' };
      }
      return { kind: 'retry', nextAttempt: input.attempt + 1 };
    }
  }
}

// libs/market/events/src/lib/topics.ts
export const MARKET_TOPICS = [
  'item.catalog.events', // item.catalog.updated
  'market.snapshot.events', // market.snapshot.collected
  'market.price.events', // market.price.updated
  'market.popularity.events', // market.popularity.recalculated
  'provider.status.events', // provider.status.changed
] as const;
 
// 스트림을 독립적으로 읽는 초기 consumer group들
export const CONSUMER_GROUPS = {
  realtimeGateway: 'realtime-gateway',
  marketRankingWorker: 'market-ranking-worker',
  marketAlertWorker: 'market-alert-worker',
} as const;

-- libs/market/data-access/prisma/migrations/.../create_outbox_events/migration.sql
CREATE TABLE "outbox_events" (
  "id"             UUID           NOT NULL DEFAULT gen_random_uuid(),
  "event_id"       UUID           NOT NULL,
  "event_type"     VARCHAR(128)   NOT NULL,
  "topic"          VARCHAR(128)   NOT NULL,
  "key"            VARCHAR(128)   NOT NULL,
  "dedupe_key"     VARCHAR(255)   NOT NULL,
  "payload"        JSONB          NOT NULL,
  "published_at"   TIMESTAMPTZ(6),           -- 발행 후 도장
  "attempts"       INTEGER        NOT NULL DEFAULT 0,
  "last_error"     TEXT,
  "created_at"     TIMESTAMPTZ(6) NOT NULL DEFAULT now(),
  PRIMARY KEY ("id")
);
 
-- 한 사실당 outbox row 하나: 재시도된 생산 트랜잭션이 이중 발행 못 함
CREATE UNIQUE INDEX "outbox_events_dedupe_key_key" ON "outbox_events" ("dedupe_key");
 
-- relay는 미발행 row를 오래된 순서로 스캔
CREATE INDEX "idx_outbox_events_unpublished" ON "outbox_events" ("published_at", "created_at");

// libs/market/data-access/src/lib/repositories/prisma-auction-collector.repository.ts
// ... 스냅샷 / 매물 / 집계 write를 ops에 push한 뒤 ...
 
// Transactional outbox: 이벤트를 스냅샷 write와 같은 트랜잭션에 넣어
// DB 커밋과 Kafka 발행 사이 유실을 원천 차단한다.
ops.push(
  this.prisma.outboxEvent.createMany({
    data: this.buildOutboxEvents(input), // snapshot.collected + price.updated
    skipDuplicates: true,
  })
);
 
await this.prisma.$transaction(ops);

// libs/market/collection/src/lib/use-cases/relay-outbox.use-case.ts
async execute(input: RelayOutboxInput): Promise<RelayOutboxResult> {
  const rows = await this.repo.fetchUnpublished(input.batchSize);
  const publishedIds: string[] = [];
  let failed = 0;
 
  for (const row of rows) {
    try {
      await this.publisher.publish(row);       // → Kafka
      publishedIds.push(row.id);
    } catch (error: unknown) {
      failed += 1;
      await this.repo.markFailed(row.id, /* ... */); // attempts++, 미발행 유지
    }
  }
  if (publishedIds.length > 0) {
    await this.repo.markPublished(publishedIds, input.now);
  }
  return { fetched: rows.length, published: publishedIds.length, failed };
}

sequenceDiagram
  participant W as market-worker
  participant DB as PostgreSQL
  participant R as outbox relay
  participant K as Kafka
  participant C as consumer
 
  W->>DB: BEGIN
  W->>DB: snapshot / aggregate write
  W->>DB: INSERT outbox_events (dedupe_key UNIQUE)
  W->>DB: COMMIT
  Note over W,DB: 한 트랜잭션 — 둘 다 커밋되거나 둘 다 롤백
 
  loop relay 폴링
    R->>DB: SELECT WHERE published_at IS NULL ORDER BY created_at
    R->>K: publish(topic, key=itemId, envelope)
    R->>DB: UPDATE published_at = now()
  end
 
  K->>C: deliver
  Note over C: eventId로 dedupe (at-least-once)

// libs/shared/messaging-kafka/src/lib/ports.ts
export function idempotent(
  handler: EventHandler,
  keyOf: (event: ReceivedEvent) => string | undefined,
  seen: SeenStore = createBoundedSeenStore() // FIFO, 기본 10_000 keys
): EventHandler {
  return {
    async handle(event: ReceivedEvent): Promise<void> {
      const key = keyOf(event);
      if (key !== undefined && seen.has(key)) return; // 이미 본 이벤트면 skip
      await handler.handle(event);
      if (key !== undefined) seen.add(key);
    },
  };
}

// apps/market-cron/src/app/relay/kafka-outbox-event.publisher.ts
async publish(record: OutboxEventRecord): Promise<void> {
  await this.producer.publish({
    topic: record.topic,
    key: record.envelope.key,   // = itemId → 한 아이템은 한 partition에 고정
    value: record.envelope,
  });
}

// apps/realtime-gateway/src/app/gateway.ts
let event;
try {
  event = parseMarketEvent(received.value);
} catch (error: unknown) {
  // poison 이벤트가 partition을 막으면 안 된다. DB/outbox가 진실이니 skip.
  log(`skipping unparseable event at ${received.topic}@${received.offset}: ...`);
  return Promise.resolve();
}