Saga+Outbox로 중복결제 막는 멱등키·Kafka 패턴

-- 결제 엔티티
create table payments (
  payment_id bigint primary key,
  order_id bigint not null,
  idempotency_key varchar(64) not null,
  status varchar(32) not null, -- PENDING, AUTHORIZING, AUTHORIZED, CAPTURED, FAILED, CANCELED
  amount bigint not null,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),
  unique (idempotency_key),
  unique (order_id) -- 주문당 결제 1회 정책이라면
);

-- 승인 요청(외부 PG 호출 단위)
create table payment_authorizations (
  authorization_id bigint primary key,
  payment_id bigint not null,
  authorization_key varchar(64) not null,
  status varchar(32) not null, -- REQUESTED, SUCCEEDED, FAILED
  pg_transaction_id varchar(128),
  created_at timestamptz not null default now(),
  unique (authorization_key),
  unique (pg_transaction_id)
);

-- Outbox
create table outbox_events (
  event_id uuid primary key,
  aggregate_type varchar(32) not null, -- PAYMENT
  aggregate_id bigint not null,        -- payment_id
  event_type varchar(64) not null,     -- PaymentAuthorized, PaymentFailed...
  payload jsonb not null,
  status varchar(16) not null default 'NEW', -- NEW, PUBLISHED
  created_at timestamptz not null default now(),
  published_at timestamptz
);

create index on outbox_events (status, created_at);

// Express/Fastify 스타일 의사코드
async function createOrGetPayment(req) {
  const idemKey = req.headers['idempotency-key'];
  const { orderId, amount } = req.body;

  // 1) 멱등키로 upsert 시도
  // 충돌 시 기존 payment를 조회해서 반환
  const payment = await db.tx(async (tx) => {
    try {
      return await tx.payments.insert({
        order_id: orderId,
        idempotency_key: idemKey,
        status: 'PENDING',
        amount
      });
    } catch (e) {
      if (isUniqueViolation(e)) {
        return await tx.payments.findByIdempotencyKey(idemKey);
      }
      throw e;
    }
  });

  return payment;
}

async function authorizePayment(req) {
  const { paymentId } = req.params;
  const authKey = req.headers['authorization-key'];

  // 2) 승인 요청 레코드부터 멱등하게 만들기
  const authorization = await db.tx(async (tx) => {
    try {
      return await tx.payment_authorizations.insert({
        payment_id: paymentId,
        authorization_key: authKey,
        status: 'REQUESTED'
      });
    } catch (e) {
      if (isUniqueViolation(e)) {
        return await tx.payment_authorizations.findByAuthorizationKey(authKey);
      }
      throw e;
    }
  });

  // 이미 성공한 승인이라면 그대로 반환
  if (authorization.status === 'SUCCEEDED') {
    return { ok: true, paymentId };
  }

  // 3) 상태 전이 가드: PENDING/FAILED 등에서만 AUTHORIZING으로
  const updated = await db.payments.updateStatusIf(paymentId, {
    from: ['PENDING', 'FAILED'],
    to: 'AUTHORIZING'
  });

  if (!updated) {
    // 누군가 이미 진행 중이거나 완료
    return { ok: true, paymentId };
  }

  // 4) 외부 PG 호출(타임아웃/재시도 설계 필요)
  const pgRes = await pg.authorize({ paymentId, amount: req.body.amount, authKey });

  // 5) 결과를 DB+Outbox에 원자적으로 기록
  await db.tx(async (tx) => {
    if (pgRes.approved) {
      await tx.payments.update(paymentId, { status: 'AUTHORIZED' });
      await tx.payment_authorizations.markSucceeded(authorization.authorization_id, pgRes.txId);

      await tx.outbox_events.insert({
        event_id: uuidv4(),
        aggregate_type: 'PAYMENT',
        aggregate_id: paymentId,
        event_type: 'PaymentAuthorized',
        payload: { paymentId, pgTxId: pgRes.txId }
      });
    } else {
      await tx.payments.update(paymentId, { status: 'FAILED' });
      await tx.payment_authorizations.markFailed(authorization.authorization_id);

      await tx.outbox_events.insert({
        event_id: uuidv4(),
        aggregate_type: 'PAYMENT',
        aggregate_id: paymentId,
        event_type: 'PaymentAuthorizationFailed',
        payload: { paymentId, reason: pgRes.reason }
      });
    }
  });

  return { ok: pgRes.approved, paymentId };
}

async function publishOutboxLoop() {
  while (true) {
    const batch = await db.outbox_events.findNew({ limit: 100 });

    for (const ev of batch) {
      try {
        // Kafka key는 aggregate_id로 두는 경우가 많음(같은 payment 순서 보장)
        await kafkaProducer.send({
          topic: 'payment-events',
          messages: [{
            key: String(ev.aggregate_id),
            value: JSON.stringify({
              eventId: ev.event_id,
              type: ev.event_type,
              payload: ev.payload,
              occurredAt: ev.created_at
            })
          }]
        });

        await db.outbox_events.markPublished(ev.event_id);
      } catch (e) {
        // 발행 실패 시 그대로 두고 다음 루프에서 재시도
        // 관측성(로그/메트릭/알람) 필수
      }
    }

    await sleep(200);
  }
}

create table consumed_events (
  consumer_group varchar(128) not null,
  event_id uuid not null,
  consumed_at timestamptz not null default now(),
  primary key (consumer_group, event_id)
);

async function handlePaymentEvent(message) {
  const { eventId, type, payload } = JSON.parse(message.value.toString());

  await db.tx(async (tx) => {
    // 1) 중복 체크를 유니크 제약으로
    try {
      await tx.consumed_events.insert({
        consumer_group: 'ledger-service',
        event_id: eventId
      });
    } catch (e) {
      if (isUniqueViolation(e)) {
        return; // 이미 처리함
      }
      throw e;
    }

    // 2) 도메인 반영(예: 원장 기록)
    if (type === 'PaymentAuthorized') {
      await tx.ledger_entries.insert({
        payment_id: payload.paymentId,
        pg_tx_id: payload.pgTxId,
        amount: payload.amount
      });
    }

    if (type === 'PaymentCanceled') {
      await tx.ledger_entries.insert({
        payment_id: payload.paymentId,
        amount: -payload.amount
      });
    }
  });

  // 3) 트랜잭션 성공 후 오프셋 커밋
}