Kafka Saga에서 보상 누락·중복 방지 구현법

-- 비즈니스 테이블 예: 주문
create table orders (
  order_id varchar(50) primary key,
  status varchar(30) not null,
  updated_at timestamptz not null default now()
);

-- Outbox 테이블
create table outbox_events (
  id bigserial primary key,
  aggregate_type varchar(50) not null,
  aggregate_id varchar(50) not null,
  event_type varchar(100) not null,
  payload jsonb not null,
  saga_id varchar(50),
  step varchar(50),
  created_at timestamptz not null default now(),
  published_at timestamptz,
  publish_attempts int not null default 0
);

create index idx_outbox_unpublished on outbox_events(published_at) where published_at is null;

// TypeScript 예시 (의사코드)
async function markOrderFailedAndEnqueueEvent(db, { orderId, sagaId, reason }) {
  await db.tx(async (tx) => {
    await tx.query(
      `update orders set status = $1, updated_at = now() where order_id = $2`,
      ['FAILED', orderId]
    );

    await tx.query(
      `insert into outbox_events(aggregate_type, aggregate_id, event_type, payload, saga_id, step)
       values ($1, $2, $3, $4::jsonb, $5, $6)`,
      ['Order', orderId, 'OrderFailed', JSON.stringify({ orderId, reason }), sagaId, 'ORDER']
    );
  });
}

async function outboxPublisherLoop(db, kafkaProducer) {
  while (true) {
    const rows = await db.query(
      `select id, event_type, payload, saga_id, step
       from outbox_events
       where published_at is null
       order by id
       limit 100`
    );

    for (const r of rows) {
      try {
        await kafkaProducer.send({
          topic: 'saga-events',
          messages: [
            {
              key: r.saga_id ?? String(r.id),
              value: JSON.stringify({
                eventType: r.event_type,
                payload: r.payload,
                sagaId: r.saga_id,
                step: r.step,
                outboxId: r.id
              })
            }
          ]
        });

        await db.query(
          `update outbox_events
           set published_at = now(), publish_attempts = publish_attempts + 1
           where id = $1`,
          [r.id]
        );
      } catch (e) {
        await db.query(
          `update outbox_events
           set publish_attempts = publish_attempts + 1
           where id = $1`,
          [r.id]
        );
      }
    }

    await new Promise((r) => setTimeout(r, 300));
  }
}

create table inbox_processed (
  consumer_group varchar(200) not null,
  message_id varchar(100) not null,
  processed_at timestamptz not null default now(),
  primary key (consumer_group, message_id)
);

async function handleSagaEvent(db, consumerGroup, msg) {
  const parsed = JSON.parse(msg.value.toString());
  const messageId = String(parsed.outboxId ?? parsed.eventId);

  await db.tx(async (tx) => {
    // 1) 이미 처리했는지 먼저 기록 시도
    try {
      await tx.query(
        `insert into inbox_processed(consumer_group, message_id)
         values ($1, $2)`,
        [consumerGroup, messageId]
      );
    } catch (e) {
      // PK 충돌이면 이미 처리한 메시지
      return;
    }

    // 2) 여기부터는 “정확히 한 번 처리” 구간
    await applyBusinessLogicAndMaybeCompensate(tx, parsed);
  });
}

create table saga_instances (
  saga_id varchar(50) primary key,
  state varchar(30) not null,
  current_step varchar(50) not null,
  last_event_id varchar(100),
  updated_at timestamptz not null default now()
);

create table saga_steps (
  saga_id varchar(50) not null,
  step varchar(50) not null,
  status varchar(30) not null,
  updated_at timestamptz not null default now(),
  primary key (saga_id, step)
);

async function onShipmentFailed(tx, { sagaId, reason }) {
  // 현재까지 완료된 step 조회
  const steps = await tx.query(
    `select step, status from saga_steps where saga_id = $1`,
    [sagaId]
  );

  const done = new Set(steps.rows.filter(r => r.status === 'DONE').map(r => r.step));

  // 보상은 역순으로
  if (done.has('INVENTORY')) {
    await tx.query(
      `update saga_steps set status = 'COMPENSATING', updated_at = now()
       where saga_id = $1 and step = 'INVENTORY' and status = 'DONE'`,
      [sagaId]
    );

    await tx.query(
      `insert into outbox_events(aggregate_type, aggregate_id, event_type, payload, saga_id, step)
       values ('Saga', $1, 'CompensateInventory', $2::jsonb, $1, 'INVENTORY')`,
      [sagaId, JSON.stringify({ sagaId, reason, idempotencyKey: `comp-inv-${sagaId}` })]
    );
  }

  if (done.has('PAYMENT')) {
    await tx.query(
      `update saga_steps set status = 'COMPENSATING', updated_at = now()
       where saga_id = $1 and step = 'PAYMENT' and status = 'DONE'`,
      [sagaId]
    );

    await tx.query(
      `insert into outbox_events(aggregate_type, aggregate_id, event_type, payload, saga_id, step)
       values ('Saga', $1, 'CompensatePayment', $2::jsonb, $1, 'PAYMENT')`,
      [sagaId, JSON.stringify({ sagaId, reason, idempotencyKey: `comp-pay-${sagaId}` })]
    );
  }
}

create table compensation_executions (
  idempotency_key varchar(120) primary key,
  saga_id varchar(50) not null,
  step varchar(50) not null,
  status varchar(30) not null,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

async function compensateInventory(db, { sagaId, idempotencyKey }) {
  await db.tx(async (tx) => {
    // 1) 멱등키 선점
    const inserted = await tx.query(
      `insert into compensation_executions(idempotency_key, saga_id, step, status)
       values ($1, $2, 'INVENTORY', 'STARTED')
       on conflict (idempotency_key) do nothing`,
      [idempotencyKey, sagaId]
    );

    if (inserted.rowCount === 0) {
      // 이미 실행(또는 실행 중)
      return;
    }

    // 2) 실제 보상 로직 (재고 복구 등)
    await tx.query(
      `update inventory_reservations
       set status = 'CANCELLED'
       where saga_id = $1 and status = 'RESERVED'`,
      [sagaId]
    );

    // 3) 완료 마킹
    await tx.query(
      `update compensation_executions
       set status = 'DONE', updated_at = now()
       where idempotency_key = $1`,
      [idempotencyKey]
    );
  });
}