Transactional Outbox (트랜잭셔널 아웃박스)

직원이 직접 우체국에 가는 방식         사내 발송함을 쓰는 방식
──────────────────────────             ──────────────────────────
1. 서류 작성 완료                       1. 서류 작성 + 발송함에 편지 넣기 (한 번에)
2. 우체국으로 출발                      2. 우편 담당자가 주기적으로 발송함을 비우며
   (가는 길에 일이 생기면                  우체국에 보낸다
    편지는 영영 안 간다)                   (담당자가 아파도 편지는 발송함에 남아 있다)

async placeOrder(cmd: PlaceOrderCommand) {
  const order = Order.place(cmd);
  await this.orderRepo.save(order);                     // ① DB
  await this.kafka.emit('ordering.events', event);      // ② 브로커
}

sequenceDiagram
    participant S as 주문 서비스
    participant DB as PostgreSQL
    participant K as Kafka

    S->>DB: BEGIN; INSERT order; COMMIT
    DB-->>S: OK
    Note over S: 여기서 프로세스 종료
    S--xK: emit(OrderPlaced) 실행되지 않음
    Note over DB,K: DB에는 주문이 있지만<br/>아무도 그 사실을 모른다

flowchart LR
    subgraph TX[하나의 DB 트랜잭션]
        A[orders 테이블<br/>INSERT] --- B[outbox 테이블<br/>INSERT]
    end
    S[주문 서비스] --> TX
    TX --> R[Message Relay<br/>별도 프로세스]
    R -->|publish| K[[Kafka]]
    R -->|발행 완료 표시| B

CREATE TABLE outbox (
  id             UUID        PRIMARY KEY,          -- 이벤트 ID (소비자 멱등 처리에 사용)
  aggregate_type TEXT        NOT NULL,             -- 'Order'
  aggregate_id   TEXT        NOT NULL,             -- '1001' → Kafka 파티션 키
  type           TEXT        NOT NULL,             -- 'ordering.order-placed.v1'
  payload        JSONB       NOT NULL,             -- 이벤트 본문
  headers        JSONB       NOT NULL DEFAULT '{}',-- correlationId 등
  created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
  published_at   TIMESTAMPTZ                       -- NULL이면 아직 발행 안 됨
);

-- 미발행 이벤트만 빠르게 찾기 위한 부분 인덱스
CREATE INDEX outbox_unpublished_idx ON outbox (created_at) WHERE published_at IS NULL;

// shared/outbox/outbox.entity.ts
@Entity('outbox')
export class OutboxEntity {
  @PrimaryColumn('uuid') id: string;
  @Column() aggregateType: string;
  @Column() aggregateId: string;
  @Column() type: string;
  @Column('jsonb') payload: object;
  @Column('jsonb', { default: {} }) headers: object;
  @CreateDateColumn() createdAt: Date;
  @Column({ type: 'timestamptz', nullable: true }) publishedAt: Date | null;
}

// ordering/application/place-order.handler.ts
@CommandHandler(PlaceOrderCommand)
export class PlaceOrderHandler implements ICommandHandler<PlaceOrderCommand> {
  constructor(private readonly dataSource: DataSource) {}

  async execute(cmd: PlaceOrderCommand): Promise<string> {
    const order = Order.place(OrderId.generate(), CustomerId.of(cmd.customerId), cmd.lines);

    await this.dataSource.transaction(async (tx) => {
      // ① 비즈니스 데이터
      await tx.getRepository(OrderOrmEntity).save(OrderMapper.toOrm(order));

      // ② 같은 트랜잭션에 outbox 레코드
      const events = order.pullEvents().map((e) => toIntegrationEvent(e));
      await tx.getRepository(OutboxEntity).insert(
        events.map((e) => ({
          id: e.eventId,
          aggregateType: 'Order',
          aggregateId: order.id.value,
          type: e.type,
          payload: e.data,
          headers: { correlationId: cmd.correlationId },
        })),
      );
    }); // 커밋: 둘 다 저장되거나 둘 다 롤백된다

    return order.id.value;
  }
}

// Repository가 Aggregate 저장과 outbox 기록을 함께 처리한다
async save(order: Order): Promise<void> {
  await this.dataSource.transaction(async (tx) => {
    await tx.getRepository(OrderOrmEntity).save(OrderMapper.toOrm(order));
    await this.outbox.append(tx, 'Order', order.id.value, order.pullEvents());
  });
}

// shared/outbox/outbox-relay.ts
@Injectable()
export class OutboxRelay {
  private readonly logger = new Logger(OutboxRelay.name);

  constructor(
    private readonly dataSource: DataSource,
    @Inject('KAFKA') private readonly kafka: ClientKafka,
  ) {}

  @Interval(500) // @nestjs/schedule
  async relay() {
    await this.dataSource.transaction(async (tx) => {
      // FOR UPDATE SKIP LOCKED: 릴레이 인스턴스가 여러 개여도 같은 행을 동시에 집지 않는다
      const rows: OutboxEntity[] = await tx.query(
        `SELECT * FROM outbox
          WHERE published_at IS NULL
          ORDER BY created_at
          LIMIT 100
          FOR UPDATE SKIP LOCKED`,
      );

      for (const row of rows) {
        try {
          await lastValueFrom(
            this.kafka.emit(`${row.aggregate_type.toLowerCase()}.events`, {
              key: row.aggregate_id, // 같은 주문 → 같은 파티션 → 순서 보장
              value: JSON.stringify({
                eventId: row.id,
                type: row.type,
                occurredAt: row.created_at,
                data: row.payload,
              }),
              headers: row.headers,
            }),
          );
          await tx.query(`UPDATE outbox SET published_at = now() WHERE id = $1`, [row.id]);
        } catch (err) {
          this.logger.error(`outbox ${row.id} 발행 실패, 다음 주기에 재시도`, err);
          break; // 같은 Aggregate의 뒤 이벤트가 먼저 나가지 않도록 이번 배치는 멈춘다
        }
      }
    });
  }
}

flowchart LR
    S[주문 서비스] -->|INSERT orders + outbox| DB[(PostgreSQL)]
    DB -->|WAL 스트림| D[Debezium<br/>Kafka Connect]
    D -->|Outbox Event Router| K[[Kafka<br/>order.events]]

-- 모니터링: 가장 오래된 미발행 이벤트가 몇 초 전에 생겼나
SELECT EXTRACT(EPOCH FROM now() - MIN(created_at)) AS oldest_unpublished_seconds
  FROM outbox WHERE published_at IS NULL;

-- 정리: 7일 지난 발행 완료 이벤트 삭제
DELETE FROM outbox WHERE published_at < now() - interval '7 days';