06 — 분산 트랜잭션 & 사가

// 단일 DataSource 안에서 여러 테이블을 묶음
await AppDataSource.transaction(async (manager) => {
  await manager.save(Order, /* ... */);
  await manager.save(OrderItem, /* ... */);
  await manager.save(Payment, /* ... */);
});

// 두 DataSource에 걸친 트랜잭션 — TypeORM은 이걸 *합치지 않는다*
async function transferAcrossDBs() {
  await OrderDB.transaction(async (m1) => {
    await m1.save(Order, /* ... */);
 
    // ❌ 이 트랜잭션은 OrderDB와 *무관한 별개* 트랜잭션이다
    await InventoryDB.transaction(async (m2) => {
      await m2.update(Inventory, /* ... */);
    });
    // ❌ InventoryDB가 COMMIT된 후 OrderDB가 ROLLBACK되면 — 재고만 빠짐
  });
}

1. CreateOrder      (DB-A에 INSERT)
2. ReserveInventory (DB-B에 UPDATE)
3. ChargePayment    (외부 PG)
4. SendNotification (외부 메시징)

3 실패 → 2 보상(재고 원복) → 1 보상(주문 취소)
4 실패 → 3 보상(환불) → 2 보상(재고 원복) → 1 보상(주문 취소)

// 중앙에서 단계를 호출하고 실패 시 보상을 호출
class OrderSaga {
  async execute(input: OrderInput) {
    const steps: SagaStep[] = [];
 
    try {
      const order = await this.createOrder(input);
      steps.push({ compensate: () => this.cancelOrder(order.id) });
 
      const reservation = await this.reserveInventory(order);
      steps.push({ compensate: () => this.releaseInventory(reservation.id) });
 
      const payment = await this.chargePayment(order);
      steps.push({ compensate: () => this.refundPayment(payment.id) });
 
      return order;
    } catch (err) {
      // 역순으로 보상
      for (const step of steps.reverse()) {
        await step.compensate();
      }
      throw err;
    }
  }
}

// 각 서비스가 다른 서비스의 이벤트를 듣고 자기 단계를 실행
 
// Order Service
on('OrderCreated', () => publish('ReserveInventoryRequested'));
 
// Inventory Service
on('ReserveInventoryRequested', async () => {
  try {
    await reserveInventory();
    publish('InventoryReserved');
  } catch {
    publish('InventoryReservationFailed');
  }
});
 
// Payment Service
on('InventoryReserved', () => chargePayment());
on('InventoryReservationFailed', () => publish('OrderShouldBeCancelled'));

// ❌ 안티패턴
await AppDataSource.transaction(async (manager) => {
  await manager.save(Order, order);
 
  // ⚠️ DB 트랜잭션은 *Kafka에 발행한 사실*을 모른다
  await kafka.publish('OrderCreated', order);
 
  // 만약 여기서 throw가 나면 DB는 ROLLBACK되지만 Kafka는 *이미 발행됨*
  validateSomething();
});

@Entity()
class OutboxEvent {
  @PrimaryGeneratedColumn('uuid')
  id: string;
 
  @Column()
  aggregateType: string;
 
  @Column()
  aggregateId: string;
 
  @Column()
  eventType: string;
 
  @Column({ type: 'jsonb' })
  payload: any;
 
  @CreateDateColumn()
  createdAt: Date;
 
  @Column({ type: 'timestamp', nullable: true })
  sentAt: Date | null;
}
 
// 비즈니스 트랜잭션
async function createOrder(input: OrderInput) {
  return AppDataSource.transaction(async (manager) => {
    const order = await manager.save(Order, input);
 
    // ✅ 같은 트랜잭션에 outbox INSERT
    await manager.save(OutboxEvent, {
      aggregateType: 'Order',
      aggregateId: order.id,
      eventType: 'OrderCreated',
      payload: { orderId: order.id, total: order.total },
    });
 
    return order;
  });
}
 
// 별도 워커 — polling 또는 logical replication
async function outboxWorker() {
  setInterval(async () => {
    const unsent = await AppDataSource.getRepository(OutboxEvent).find({
      where: { sentAt: IsNull() },
      order: { createdAt: 'ASC' },
      take: 100,
    });
 
    for (const event of unsent) {
      try {
        await kafka.publish(event.eventType, event.payload);
        event.sentAt = new Date();
        await AppDataSource.getRepository(OutboxEvent).save(event);
      } catch (err) {
        // retry on next tick
      }
    }
  }, 1000);
}

PostgreSQL WAL → Debezium → Kafka

1. CreateOrder      ✅
2. ReserveInventory ✅
3. ChargePayment    ❌ FAILED
→ 2 보상: ReleaseInventory  → ❌ FAILED (재고 서비스 죽음)
→ 1 보상: CancelOrder       → 시도 못 함

// ❌ 멱등하지 않음
async function refundPayment(paymentId: string) {
  const p = await paymentRepo.findOne({ where: { id: paymentId } });
  await pg.refund(p.externalId, p.amount); // 두 번 부르면 두 번 환불
}
 
// ✅ 멱등
async function refundPayment(paymentId: string) {
  const p = await paymentRepo.findOne({ where: { id: paymentId } });
  if (p.refundedAt) return; // 이미 환불됨
  await pg.refund(p.externalId, p.amount, { idempotencyKey: paymentId });
  await paymentRepo.update({ id: paymentId }, { refundedAt: new Date() });
}