Saga 패턴 심화: Choreography vs Orchestration, 그리고 Temporal로 보상 로직 구현하기

// order-service/order.consumer.ts
import { Kafka } from 'kafkajs';
 
const kafka = new Kafka({ brokers: ['localhost:9092'] });
const producer = kafka.producer();
const consumer = kafka.consumer({ groupId: 'order-service' });
 
export async function createOrderAndPublish(ctx: OrderContext): Promise<void> {
  await orderRepository.create({ id: ctx.orderId, status: 'PENDING' });
  await producer.send({
    topic: 'order-events',
    messages: [{ key: ctx.orderId, value: JSON.stringify({ type: 'OrderCreated', ...ctx }) }],
  });
}
 
// 보상 이벤트 구독: 결제 환불이 완료된 뒤 주문 취소
await consumer.subscribe({ topics: ['payment-events'] });
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value!.toString());
    if (event.type === 'PaymentRefunded') {
      await orderRepository.updateStatus(event.orderId, 'CANCELLED');
    }
  },
});

// payment-service/payment.consumer.ts
import { Kafka } from 'kafkajs';
 
const consumer = kafka.consumer({ groupId: 'payment-service' });
const producer = kafka.producer();
 
// 정상 흐름 이벤트와 보상 이벤트를 모두 구독
await consumer.subscribe({ topics: ['order-events', 'inventory-events'] });
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value!.toString());
 
    if (event.type === 'OrderCreated') {
      const result = await paymentGateway.charge(event);
      await producer.send({
        topic: 'payment-events',
        messages: [{
          key: event.orderId,
          value: JSON.stringify(
            result.success
              ? { type: 'PaymentProcessed', orderId: event.orderId, transactionId: result.transactionId }
              : { type: 'PaymentFailed', orderId: event.orderId, reason: result.reason }
          ),
        }],
      });
    }
 
    // 재고 실패 시 환불 (보상 트랜잭션)
    if (event.type === 'InventoryFailed') {
      await paymentGateway.refund({ orderId: event.orderId, transactionId: event.transactionId });
      await producer.send({
        topic: 'payment-events',
        messages: [{
          key: event.orderId,
          value: JSON.stringify({ type: 'PaymentRefunded', orderId: event.orderId }),
        }],
      });
    }
  },
});

// TypeORM + PostgreSQL 예시
async function createOrderWithOutbox(
  ctx: OrderContext,
  queryRunner: QueryRunner
): Promise<void> {
  await queryRunner.startTransaction();
  try {
    await queryRunner.manager.save(Order, {
      id: ctx.orderId,
      status: 'PENDING',
      amount: ctx.amount,
    });
 
    // Kafka에 직접 발행하는 대신 outbox 테이블에 기록 (같은 트랜잭션)
    await queryRunner.manager.save(OutboxEvent, {
      id: uuidv4(),
      aggregateType: 'Order',
      aggregateId: ctx.orderId,
      eventType: 'OrderCreated',
      payload: JSON.stringify(ctx),
      createdAt: new Date(),
    });
 
    await queryRunner.commitTransaction();
  } catch (err) {
    await queryRunner.rollbackTransaction();
    throw err;
  }
  // Debezium 같은 CDC 도구가 outbox 테이블 변경을 감지 → Kafka 발행
}

pnpm add @temporalio/workflow @temporalio/activity @temporalio/worker @temporalio/client

// activities.ts
import { log, ApplicationFailure } from '@temporalio/activity';
 
export interface OrderContext {
  orderId: string;
  userId: string;
  amount: number;
  items: Array<{ productId: string; quantity: number }>;
}
 
export async function createOrder(ctx: OrderContext): Promise<void> {
  await orderRepository.create({ id: ctx.orderId, userId: ctx.userId, status: 'PENDING', amount: ctx.amount });
  log.info('주문 생성 완료', { orderId: ctx.orderId });
}
 
export async function cancelOrder(orderId: string): Promise<void> {
  // 이미 취소된 주문이면 건너뜀 (멱등성 보장)
  const order = await orderRepository.findById(orderId);
  if (order?.status === 'CANCELLED') return;
  await orderRepository.updateStatus(orderId, 'CANCELLED');
  log.info('주문 취소 완료', { orderId });
}
 
export async function processPayment(ctx: OrderContext): Promise<string> {
  const result = await paymentGateway.charge({
    orderId: ctx.orderId,
    userId: ctx.userId,
    amount: ctx.amount,
  });
 
  if (!result.success) {
    // 결제 거절은 재시도해도 결과가 바뀌지 않으므로 non-retryable
    throw ApplicationFailure.nonRetryable(`결제 실패: ${result.reason}`, 'PAYMENT_DECLINED');
  }
 
  return result.transactionId;
}
 
export async function refundPayment(orderId: string, transactionId: string): Promise<void> {
  const existing = await paymentRepository.findRefund(orderId);
  if (existing) {
    log.info('이미 환불 처리됨, 건너뜀', { orderId });
    return;
  }
  await paymentGateway.refund({ orderId, transactionId, idempotencyKey: `refund-${orderId}` });
  await paymentRepository.markRefunded(orderId, transactionId);
  log.info('환불 완료', { orderId });
}
 
export async function updateInventory(ctx: OrderContext): Promise<void> {
  // 주의: 이 Activity가 재시도될 경우 이미 차감된 항목이 다시 차감될 수 있습니다.
  // 프로덕션에서는 각 decrement 호출에 idempotency key를 사용하거나,
  // 차감 전 현재 재고를 조회하여 이미 처리됐는지 확인하는 로직이 필요합니다.
  const decremented: typeof ctx.items = [];
 
  for (const item of ctx.items) {
    const success = await inventoryService.decrement(item.productId, item.quantity);
    if (!success) {
      // 재고 부족 — 이미 차감한 항목은 이 Activity 내에서 복원 후 non-retryable throw
      for (const done of decremented) {
        await inventoryService.increment(done.productId, done.quantity);
      }
      throw ApplicationFailure.nonRetryable(`재고 부족: ${item.productId}`, 'INSUFFICIENT_INVENTORY');
    }
    decremented.push(item);
  }
}
 
export async function restoreInventory(ctx: OrderContext): Promise<void> {
  for (const item of ctx.items) {
    await inventoryService.increment(item.productId, item.quantity);
  }
  log.info('재고 복원 완료', { orderId: ctx.orderId });
}
 
export async function scheduleShipping(orderId: string): Promise<void> {
  await shippingService.schedule({ orderId });
}
 
export async function cancelShipping(orderId: string): Promise<void> {
  await shippingService.cancel({ orderId });
  log.info('배송 취소 완료', { orderId });
}

// workflows.ts
import { proxyActivities, log } from '@temporalio/workflow';
import type * as activities from './activities';
import type { OrderContext } from './activities';
 
const {
  createOrder, cancelOrder,
  processPayment, refundPayment,
  updateInventory, restoreInventory,
  scheduleShipping, cancelShipping,
} = proxyActivities<typeof activities>({
  startToCloseTimeout: '10 minutes',
  retry: {
    maximumAttempts: 3,
    initialInterval: '1 second',
    backoffCoefficient: 2,
  },
});
 
export async function orderSagaWorkflow(ctx: OrderContext): Promise<void> {
  // 보상 스택: unshift()로 맨 앞에 추가하면 for...of 순회 시 자동으로 역순 실행
  const compensations: Array<() => Promise<void>> = [];
 
  try {
    await createOrder(ctx);
    compensations.unshift(() => cancelOrder(ctx.orderId));
 
    const transactionId = await processPayment(ctx);
    // transactionId를 클로저로 캡처 — 보상 시점에 별도 상태 관리 불필요
    compensations.unshift(() => refundPayment(ctx.orderId, transactionId));
 
    await updateInventory(ctx);
    compensations.unshift(() => restoreInventory(ctx));
 
    await scheduleShipping(ctx.orderId);
    compensations.unshift(() => cancelShipping(ctx.orderId));
 
    log.info('주문 처리 완료', { orderId: ctx.orderId });
  } catch (err) {
    log.error('주문 처리 실패, 보상 트랜잭션 실행', { orderId: ctx.orderId, error: String(err) });
 
    for (const compensate of compensations) {
      try {
        await compensate();
      } catch (compensationErr) {
        // 보상 실패 — 시스템이 불일치 상태로 남을 수 있음
        // 알림 발송 또는 DLQ 적재로 수동 복구 경로 확보 필요
        log.error('보상 트랜잭션 실패', { error: String(compensationErr) });
        // 이 보상이 실패해도 나머지 보상은 계속 시도
      }
    }
 
    throw err;
  }
}

// worker.ts
import { Worker } from '@temporalio/worker';
import * as activities from './activities';
 
async function run() {
  const worker = await Worker.create({
    workflowsPath: require.resolve('./workflows'),
    activities,
    taskQueue: 'order-saga',
  });
  await worker.run();
}
 
run().catch(console.error);

// client.ts
import { Client } from '@temporalio/client';
import { orderSagaWorkflow } from './workflows';
import type { OrderContext } from './activities';
 
const client = new Client();
const orderCtx: OrderContext = {
  orderId: 'ORD-20260712-001',
  userId: 'user-123',
  amount: 59000,
  items: [{ productId: 'PROD-456', quantity: 2 }],
};
 
const handle = await client.workflow.start(orderSagaWorkflow, {
  taskQueue: 'order-saga',
  // orderId를 workflowId로 사용: 같은 주문의 중복 실행을 방지
  // 네트워크 재시도로 start()가 두 번 호출되더라도 Temporal이 멱등성을 보장
  workflowId: `order-${orderCtx.orderId}`,
  args: [orderCtx],
});
 
console.log(`Workflow 시작: ${handle.workflowId}`);
await handle.result();

// 멱등성 보장 예시 — 환불 Activity
export async function refundPayment(orderId: string, transactionId: string): Promise<void> {
  const existing = await paymentRepository.findRefund(orderId);
  if (existing) {
    log.info('이미 환불 처리됨, 건너뜀', { orderId });
    return; // 두 번 호출돼도 두 번 환불이 일어나지 않음
  }
 
  await paymentGateway.refund({
    orderId,
    transactionId,
    idempotencyKey: `refund-${orderId}`, // 외부 결제 API에도 멱등성 키 전달
  });
 
  await paymentRepository.markRefunded(orderId, transactionId);
}