ลำดับ event ไม่การันตี — version guard

อ่านหน้านี้ให้จบก่อนเขียน switch (event.type) บรรทัดแรก

webhook ของเรามีหน้าต่าง retry ยาว ~92 ชั่วโมง ⇒ event ที่สร้างเมื่อวันจันทร์มาถึงคุณวันพฤหัสได้ หลัง event ของวันพุธ

ไม่มี version guard = settled โดน under_review เขียนทับ = ปล่อยของผิด หรือไม่ปล่อยของที่ควรปล่อย

เรารับประกันแค่ 3 ข้อ

1. sequence เพิ่มขึ้นเสมอตาม เวลาที่สร้าง event (ไม่ใช่เวลาที่ส่งถึง)
2. data.object.version เพิ่มขึ้นตามลำดับที่สถานะเปลี่ยนจริงที่ฐานข้อมูลของเรา
3. 🔴 ไม่มีสอง event ของ object เดียวกันที่ version เท่ากัน — รวมถึง event ที่ไม่เปลี่ยนสถานะ (attribute-change) ด้วย

เราไม่รับประกัน ว่ามันจะมาถึงคุณตามลำดับนั้น · ไม่รับประกันว่าจะมาถึงครบก่อนใบถัดไป · และไม่รับประกันว่าจะไม่มาซ้ำ

ข้อ 3 คือสิ่งที่ทำให้วิธีในหน้านี้ปลอดภัย: ถ้า UPDATE ... WHERE version < ใหม่ แล้วได้ rowCount = 0 แปลว่า event ใบนี้เก่ากว่าที่คุณมีแน่นอน ไม่มีทางเป็นอย่างอื่น

---

เคส A — expired มาถึงหลัง settled (เกิดบ่อยที่สุด)

ผู้ใช้โอนช้า รายการหมดอายุไปก่อน แล้วอัปสลิปย้อนหลัง — เราเปิดหน้าต่างรับสลิปย้อนหลังให้ 24 ชั่วโมง ⇒ รายการที่ expired ไปแล้ว กลับมา settle ได้

เวลาที่เราสร้าง event              เวลาที่ถึง endpoint ของคุณ
10:04:41  deposit.expired   v6     10:04:41  attempt 1 → endpoint ล่ม 3 นาที → fail
10:31:02  deposit.slip_received v7 10:31:02  ✓ 200
10:33:15  deposit.settled   v9     10:33:15  ✓ 200  ← เครดิต 5,000 ปล่อยของ ถูกต้อง
                                   10:35:10  deposit.expired v6 ← retry รอบ 4 เพิ่งมาถึง!

โค้ดที่พัง:

switch (event.type) {
  case 'deposit.expired':
    await cancelOrder(orderId);        // 💥 ยกเลิกออร์เดอร์ที่ settle ไปแล้ว
    break;
}

ผลจริง: ผู้ใช้โอนเงินแล้ว ได้ของแล้ว แต่ระบบคุณ mark ออร์เดอร์เป็นยกเลิก · ยอดในระบบคุณไม่ตรงกับยอดในกระเป๋าที่เรา ⇒ รู้ตอนกระทบยอดสิ้นเดือน

เคส B — settled มาถึงหลัง reversed (เสียหายหนักที่สุด)

09:38:02  deposit.settled  v9   attempt 1 → คุณตอบ 504 (DB ช้า)
09:41:20  deposit.reversed v12  attempt 1 → ✓ 200 · คุณ mark ออร์เดอร์เป็น failed ถูกต้อง
09:44:12  deposit.settled  v9   attempt 3 → ✓ 200  💥 เครดิต + ปล่อยของ ทั้งที่ถูกกลับรายการไปแล้ว

ผลจริง: ของออกจากคลังไปแล้ว เงินถูกดึงกลับจากกระเป๋าคุณไปแล้ว · ตามเก็บไม่ได้

ทำไม settled ถึงมาช้ากว่า reversed ได้ตั้ง 6 นาที

เพราะรอบ retry ที่ 3 ห่างจากรอบแรก 1 นาที 10 วินาที และรอบที่ 4 ห่าง 6 นาที 10 วินาที — ในระหว่างนั้น event ใหม่ที่สร้างทีหลังส่งสำเร็จตั้งแต่รอบแรกไปแล้ว

ยิ่ง endpoint คุณ flaky เป็นช่วง ๆ ยิ่งเจอบ่อย

เคส C — withdrawal.filled สลับลำดับ

รายการถอน 10,000 ถูกเติมเป็น 3 ก้อน:

ก้อนที่ 1  4,000  filled_minor=400000  remaining_minor=600000  v7
ก้อนที่ 2  3,000  filled_minor=700000  remaining_minor=300000  v9
ก้อนที่ 3  3,000  filled_minor=1000000 remaining_minor=0       v11

มาถึงคุณตามลำดับ v9 → v11 → v7 · โค้ดที่พัง:

await setRemaining(w.id, event.data.object.remaining_minor);   // 💥 จบด้วยค่า 600000

ผลจริง: หน้าจอผู้ใช้แสดงว่ายังเหลือ 6,000 ทั้งที่จ่ายครบแล้ว · ถ้าระบบคุณเอาค่านี้ไปคำนวณต่อ ผู้ใช้จะถอนเกิน

---

วิธีที่ถูก — 3 หลักการ

จำแค่นี้

1. สถานะมาจาก data.object ไม่ใช่จาก event.type — event.type บอกแค่ว่า "มีอะไรเกิดขึ้น" · data.object.status คือความจริง ณ ตอนสร้าง event
2. guard ด้วย version ในคำสั่ง SQL เดียว ไม่ใช่ SELECT แล้ว if แล้ว UPDATE (สอง request ที่มาพร้อมกันจะ race กันเอง)
3. money/external side effect ผูกกับผลของ UPDATE ไม่ใช่ผูกกับ event type — rowCount = 0 แปลว่า event เก่า ⇒ ไม่เครดิต/ย้อน/ปล่อยของ; ยกเว้นการ append immutable overpayment flag ที่ยังไม่เคยเห็น

โครงตาราง DB ที่แนะนำ

-- 1) transactional inbox — schema เต็มพร้อม outbox อยู่ในหน้าตรวจลายเซ็น
--    เก็บอย่างน้อย 31 วัน เพราะ manual resend ทำได้ถึง 30 วัน
CREATE TABLE pg_webhook_inbox (
  event_id     text PRIMARY KEY,
  event_type   text NOT NULL,
  event        jsonb NOT NULL,
  received_at  timestamptz NOT NULL DEFAULT now(),
  processed_at timestamptz
);
CREATE INDEX ON pg_webhook_inbox (received_at);

-- 2) mapping ฝั่งเรา — บันทึกจาก response 201/202; ห้ามพึ่ง metadata ที่ optional
CREATE TABLE pg_deposit_correlations (
  pg_deposit_id       text PRIMARY KEY,
  merchant_reference text NOT NULL UNIQUE,
  our_order_id        text NOT NULL,
  our_user_id         text NOT NULL
);
CREATE INDEX ON pg_deposit_correlations (our_order_id);

-- 3) กระจกเงาของรายการฝากฝั่งเรา
CREATE TABLE pg_deposits (
  pg_deposit_id          text PRIMARY KEY,       -- dep_...  🔴 text ไม่ใช่ varchar(20)
  pg_version             bigint NOT NULL,        -- ← หัวใจของ version guard
  status                 text   NOT NULL,
  is_terminal            boolean,
  late_slip_accepted_until timestamptz,
  ordered_amount_minor   bigint,
  transferred_amount_minor bigint,                -- ยอดที่ธนาคารยืนยันว่าโอนจริง
  amount_received_minor  bigint,                 -- เครดิตผู้ใช้ปลายทางด้วยตัวนี้
  fee_total_minor        bigint,
  credited_amount_minor  bigint,                 -- ยอดที่เข้ากระเป๋า merchant
  fee_breakdown          jsonb,                  -- เก็บดิบไว้ตอนเถียงเรื่องค่าธรรมเนียม
  overpayment            jsonb,                  -- latest immutable flag projection; อ่าน Claim แยกเพื่อสถานะล่าสุด
  balance_transaction_id text,
  clearing_available_at  timestamptz,
  our_order_id           text NOT NULL,
  our_user_id            text NOT NULL,
  credited_user_at       timestamptz,            -- non-null = เครดิตผู้ใช้แล้ว กันเครดิตซ้ำ
  reversed_at            timestamptz,
  updated_at             timestamptz NOT NULL DEFAULT now(),
  CONSTRAINT pg_deposit_money_ck
    CHECK (status NOT IN ('settled','reversed')
           OR (ordered_amount_minor IS NOT NULL
               AND transferred_amount_minor IS NOT NULL
               AND amount_received_minor IS NOT NULL
               AND fee_total_minor IS NOT NULL
               AND credited_amount_minor IS NOT NULL
               AND ordered_amount_minor > 0
               AND transferred_amount_minor > 0
               AND amount_received_minor > 0
               AND amount_received_minor = LEAST(transferred_amount_minor,ordered_amount_minor)
               AND fee_total_minor BETWEEN 0 AND amount_received_minor
               AND amount_received_minor - fee_total_minor = credited_amount_minor))
);
CREATE INDEX ON pg_deposits (our_order_id);

-- 4) flag ยอดเกินเป็น append-only — jsonb ใน pg_deposits เป็นเพียง latest projection
CREATE TABLE pg_deposit_overpayment_flags (
  claim_id                text PRIMARY KEY,
  pg_deposit_id           text NOT NULL REFERENCES pg_deposits(pg_deposit_id),
  observed_deposit_version bigint NOT NULL,
  cause                   text NOT NULL CHECK (cause IN ('same_transfer_excess','additional_transfer')),
  amount_minor            bigint NOT NULL CHECK (amount_minor > 0),
  slip_received_at        timestamptz NOT NULL,
  status_snapshot         text NOT NULL CHECK (status_snapshot = 'open'),
  action_required_snapshot text NOT NULL CHECK (action_required_snapshot = 'contact_support'),
  expected_response_at    timestamptz,
  raw_flag                jsonb NOT NULL,
  received_at             timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX ON pg_deposit_overpayment_flags (pg_deposit_id, observed_deposit_version);
CREATE FUNCTION pg_reject_overpayment_flag_mutation() RETURNS trigger
LANGUAGE plpgsql AS $$ BEGIN
  RAISE EXCEPTION 'pg_deposit_overpayment_flags is append-only' USING ERRCODE = '55000';
  RETURN OLD; -- unreachable; ระบุ return contract ของ trigger ให้ครบ
END $$;
CREATE TRIGGER pg_deposit_overpayment_flags_immutable
  BEFORE UPDATE OR DELETE ON pg_deposit_overpayment_flags
  FOR EACH ROW EXECUTE FUNCTION pg_reject_overpayment_flag_mutation();

-- 5) กระจกเงาของรายการถอน
CREATE TABLE pg_withdrawals (
  pg_withdrawal_id     text PRIMARY KEY,         -- wdr_...
  pg_version           bigint NOT NULL,
  status               text   NOT NULL,
  amount_total_minor   bigint NOT NULL,
  filled_minor         bigint NOT NULL DEFAULT 0,   -- 🔴 มาจาก payload เท่านั้น ห้าม SUM ห้าม +=
  remaining_minor      bigint NOT NULL,
  fee_total_minor      bigint,
  fee_recognized_minor bigint,
  refunded_minor       bigint,
  settled_splits       smallint,
  our_payout_id        text NOT NULL,
  closed_at            timestamptz,
  updated_at           timestamptz NOT NULL DEFAULT now()
);

-- 6) บรรทัด fill — timeline/audit เท่านั้น 🔴 ห้ามเอาไปขับ business logic
CREATE TABLE pg_withdrawal_fills (
  leg_ref              text PRIMARY KEY,         -- ← id ของก้อน ใช้กันซ้ำได้เลย
  pg_withdrawal_id     text NOT NULL REFERENCES pg_withdrawals(pg_withdrawal_id),
  sequence             smallint,
  amount_minor         bigint NOT NULL,
  fee_recognized_minor bigint,
  settled_at           timestamptz,
  auto_confirm_at      timestamptz,
  balance_transaction_id text,
  reversed_at          timestamptz,
  received_at          timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX ON pg_withdrawal_fills (pg_withdrawal_id, sequence);

-- 7) balance_transaction — immutable ไม่มี version ⇒ dedupe ด้วย id แล้วสะสม
CREATE TABLE pg_balance_txns (
  btxn_id      text PRIMARY KEY,
  type         text NOT NULL,
  direction    text NOT NULL,
  amount_minor bigint NOT NULL,
  bucket_from  text,
  bucket_to    text,
  source_id    text,
  created_at   timestamptz NOT NULL
);
CREATE INDEX ON pg_balance_txns (source_id);

pg_deposit_id ต้องเป็น text

id ของเราไม่ใช่รูปแบบ reference สั้น ๆ แบบที่เจ้าอื่นใช้ · คอลัมน์ varchar(20) จะตัดปลาย id เงียบ ๆ แล้วคุณจะ reconcile ไม่เจอ

และ 🔴 ห้าม parse ห้ามเดาความหมายจาก prefix — แยกประเภทด้วย event.type และ data.object.object เท่านั้น

บันทึก pg_deposit_correlations ทันทีที่ได้ response 201/202 จากการสร้างรายการ โดยใช้ mapping ที่ระบบคุณรู้อยู่แล้ว. ถ้า webhook มาก่อน mapping ให้ transaction ล้มและตอบ non-2xx เพื่อรับ retry — ห้าม fallback ไปอ่าน metadata.user_id/metadata.order_id เพราะ metadata เป็น optional และอาจไม่มีทั้ง object

โค้ด — ฝั่งรายการฝาก

async function applyDepositEvent(event) {
  const o = event.data.object;

  // object ที่ไม่มี version (balance_transaction) ใช้เส้นทางอื่น
  if (o.version === undefined) return applyImmutable(event);

  // version guard + เงินใน DB + outbox + processed_at ต้อง commit พร้อมกัน
  await db.tx(async (t) => {
  const { rows: [correlation] } = await t.query(
    `SELECT our_order_id, our_user_id FROM pg_deposit_correlations WHERE pg_deposit_id = $1`,
    [o.id]);
  if (!correlation) throw new Error('missing deposit correlation; retry webhook after create response is stored');

  const { rowCount } = await t.query(`
    INSERT INTO pg_deposits (pg_deposit_id, pg_version, status, is_terminal,
                             late_slip_accepted_until, ordered_amount_minor,
                             transferred_amount_minor, amount_received_minor,
                             fee_total_minor, credited_amount_minor,
                             fee_breakdown, overpayment, balance_transaction_id, clearing_available_at,
                             our_order_id, our_user_id)
    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)
    ON CONFLICT (pg_deposit_id) DO UPDATE
      SET pg_version = EXCLUDED.pg_version,
          status     = EXCLUDED.status,
          is_terminal = EXCLUDED.is_terminal,
          late_slip_accepted_until = EXCLUDED.late_slip_accepted_until,
          ordered_amount_minor  = EXCLUDED.ordered_amount_minor,
          transferred_amount_minor = EXCLUDED.transferred_amount_minor,
          amount_received_minor = EXCLUDED.amount_received_minor,
          fee_total_minor       = EXCLUDED.fee_total_minor,
          credited_amount_minor = EXCLUDED.credited_amount_minor,
          fee_breakdown         = EXCLUDED.fee_breakdown,
          overpayment           = EXCLUDED.overpayment,
          balance_transaction_id = EXCLUDED.balance_transaction_id,
          clearing_available_at  = EXCLUDED.clearing_available_at,
          updated_at = now()
      WHERE pg_deposits.pg_version < EXCLUDED.pg_version    -- ← หัวใจ
  `, [o.id, o.version, o.status, o.is_terminal ?? null,
      o.late_slip_accepted_until ?? null, o.ordered_amount_minor ?? null,
      o.transferred_amount_minor ?? null, o.amount_received_minor ?? null,
      o.fee_total_minor ?? null, o.credited_amount_minor ?? null,
      JSON.stringify(o.fee_breakdown ?? null), JSON.stringify(o.overpayment ?? null),
      o.balance_transaction_id ?? null, o.clearing_available_at ?? null,
      correlation.our_order_id, correlation.our_user_id]);

  // append ก่อนเช็ก rowCount: event v11 ที่มาหลัง v12 ยังต้องเก็บ claim/flag ของ v11
  if (o.overpayment) {
    const flagParams = [
      o.overpayment.claim_id, o.id, o.version, o.overpayment.cause,
      o.overpayment.amount_minor, o.overpayment.slip_received_at,
      o.overpayment.status, o.overpayment.action_required,
      o.overpayment.expected_response_at ?? null, JSON.stringify(o.overpayment),
    ];
    const appended = await t.query(`
      INSERT INTO pg_deposit_overpayment_flags
        (claim_id, pg_deposit_id, observed_deposit_version, cause, amount_minor,
         slip_received_at, status_snapshot, action_required_snapshot,
         expected_response_at, raw_flag)
      VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
      ON CONFLICT (claim_id) DO NOTHING
      RETURNING claim_id
    `, flagParams);

    // claim เดิมอาจติดมากับ snapshot version ใหม่: เก็บ version ที่เห็นครั้งแรก
    // แต่ห้ามกลืน claim_id collision ที่ชี้คนละ deposit หรือ payload ไม่ตรงกัน
    if (appended.rowCount === 0) {
      const duplicate = await t.query(`
        SELECT 1 FROM pg_deposit_overpayment_flags
         WHERE claim_id = $1
           AND pg_deposit_id = $2
           AND cause = $3
           AND amount_minor = $4
           AND slip_received_at = $5::timestamptz
           AND status_snapshot = $6
           AND action_required_snapshot = $7
           AND expected_response_at IS NOT DISTINCT FROM $8::timestamptz
           AND raw_flag = $9::jsonb
      `, [o.overpayment.claim_id, o.id, o.overpayment.cause,
          o.overpayment.amount_minor, o.overpayment.slip_received_at,
          o.overpayment.status, o.overpayment.action_required,
          o.overpayment.expected_response_at ?? null, JSON.stringify(o.overpayment)]);
      if (duplicate.rowCount !== 1)
        throw new Error('overpayment claim_id collision or immutable flag mismatch');
    }
  }

  if (rowCount === 0) {
    await t.query(`UPDATE pg_webhook_inbox SET processed_at = now()
                    WHERE event_id = $1 AND processed_at IS NULL`, [event.id]);
    return;                     // event เก่ากว่าที่เรามีอยู่ → ทิ้งเงียบ ๆ
  }

  switch (o.status) {
    case 'settled':
      // 🔴 เครดิตผู้ใช้ปลายทางด้วย amount_received_minor ไม่ใช่ credited_amount_minor
      const credited = await t.query(
        `UPDATE pg_deposits SET credited_user_at = now()
          WHERE pg_deposit_id = $1 AND status = 'settled' AND credited_user_at IS NULL`,
        [o.id]);
      if (credited.rowCount === 1) {
        await t.query(
          `UPDATE users SET balance_minor = balance_minor + $1 WHERE id = $2`,
          [o.amount_received_minor, correlation.our_user_id]);
        await enqueueEffect(t, `deposit:${o.id}`, 'release_goods', {
          order_id: correlation.our_order_id,
          partial:  o.partial === true,
          received: o.amount_received_minor,
          ordered:  o.ordered_amount_minor,
        });
      }
      break;

    case 'reversed':
      const reversed = await t.query(
        `UPDATE pg_deposits SET reversed_at = now()
          WHERE pg_deposit_id = $1 AND status = 'reversed'
            AND credited_user_at IS NOT NULL AND reversed_at IS NULL
          RETURNING amount_received_minor`, [o.id]);
      if (reversed.rowCount === 1) {
        await t.query(
          `UPDATE users SET balance_minor = balance_minor - $1 WHERE id = $2`,
          [reversed.rows[0].amount_received_minor, correlation.our_user_id]);
        await enqueueEffect(t, `deposit:${o.id}`, 'hold_goods', {
          order_id: correlation.our_order_id });
      }
      break;

    case 'expired':
    case 'canceled':
      // 🔴 ไม่ใช่จุดจบถ้ายังรับสลิปย้อนหลังได้
      if (o.is_terminal === false && o.late_slip_accepted_until)
        await enqueueEffect(t, event.id, 'show_late_slip', {
          order_id: correlation.our_order_id, until: o.late_slip_accepted_until });
      else
        await enqueueEffect(t, event.id, 'fail_order', {
          order_id: correlation.our_order_id, reason: o.cancel_reason_code ?? 'expired' });
      break;

    case 'under_review':
      await enqueueEffect(t, event.id, 'show_pending_review', {
        order_id: correlation.our_order_id, message: o.review.reasons[0].display_th });
      break;

    case 'rejected':
      await enqueueEffect(t, event.id, 'fail_order', {
        order_id: correlation.our_order_id, reason: o.rejection?.display_th });
      break;

    default:
      // 🔴 status ของ deposit เป็น enum ปิดก็จริง แต่ default กันโค้ดพังตอนเราเพิ่มค่าในอนาคต
      await enqueueEffect(t, event.id, 'update_order_status', {
        order_id: correlation.our_order_id, status: o.status });
  }
  await t.query(`UPDATE pg_webhook_inbox SET processed_at = now()
                  WHERE event_id = $1 AND processed_at IS NULL`, [event.id]);
  });
}

ผลลัพธ์กับ 3 เคสข้างบน:

เคส	เกิดอะไร
A	expired (v6) มาถึงตอน pg_version = 9 → WHERE 9 < 6 เป็นเท็จ → rowCount = 0 → ไม่มี side effect → ตอบ 200
B	settled (v9) มาถึงตอน pg_version = 12 → ไม่เครดิตซ้ำ ไม่ปล่อยของ
C	remaining_minor เขียนทับได้เฉพาะเมื่อ version สูงกว่า → ยอดถูกเสมอ

Side effect ต้องออกผ่าน transactional outbox

version guard กัน event เก่า ได้ แต่ยังไม่ได้กัน worker crash กลางทาง ดังนั้นการอัปเดต version, การเพิ่ม/หักยอดใน DB, การสร้าง outbox และการตั้ง processed_at ต้องอยู่ใน transaction เดียวกัน

UPDATE pg_deposits
   SET credited_user_at = now()
 WHERE pg_deposit_id = $1
   AND status = 'settled'
   AND credited_user_at IS NULL;
-- rowCount = 1 เท่านั้นถึงจะบวกยอดให้ผู้ใช้ · ทำใน transaction เดียวกับการบวกยอด

async function enqueueEffect(t, dedupeScope, type, payload) {
  await t.query(`
    INSERT INTO pg_outbox (topic, dedupe_key, payload)
    VALUES ($1, $2, $3::jsonb)
    ON CONFLICT (dedupe_key) DO NOTHING
  `, [type, `${dedupeScope}:${type}`, JSON.stringify(payload)]);
}

งานธุรกิจที่ต้องเกิดครั้งเดียวใช้ scope ที่ผูกกับ resource เช่น deposit:${o.id}; งานอัปเดตตาม snapshot ใช้ event.id ได้ ดังนั้น event ใหม่ที่ version สูงขึ้นแต่ยังมี status settled จะไม่ปล่อยของซ้ำ

outbox worker เป็นผู้เรียก API ภายนอก เช่น release goods, email หรือ ERP โดยส่ง dedupe_key เป็น idempotency key ทุกครั้ง ห้ามเรียก service ภายนอกขณะ transaction DB เปิดอยู่

โค้ด — ฝั่งรายการถอน

สองอย่างในใบเดียวกัน: ยอดรวม เขียนทับเฉพาะเมื่อ version สูงกว่า · บรรทัด fill เพิ่มด้วย leg_ref

async function applyWithdrawalEvent(event) {
  const o = event.data.object;
  if (o.version === undefined) return applyImmutable(event);

  await db.tx(async (t) => {
    // 1) ยอดรวม — version guard
    const { rowCount } = await t.query(`
      INSERT INTO pg_withdrawals (pg_withdrawal_id, pg_version, status,
                                  amount_total_minor, filled_minor, remaining_minor,
                                  fee_total_minor, fee_recognized_minor, refunded_minor,
                                  settled_splits, our_payout_id)
      VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
      ON CONFLICT (pg_withdrawal_id) DO UPDATE
        SET pg_version = EXCLUDED.pg_version, status = EXCLUDED.status,
            filled_minor = EXCLUDED.filled_minor,           -- 🔴 เขียนทับ ไม่ใช่ +=
            remaining_minor = EXCLUDED.remaining_minor,
            fee_recognized_minor = EXCLUDED.fee_recognized_minor,
            refunded_minor = EXCLUDED.refunded_minor,
            settled_splits = EXCLUDED.settled_splits,
            updated_at = now()
        WHERE pg_withdrawals.pg_version < EXCLUDED.pg_version
    `, [o.id, o.version, o.status, o.amount_total_minor, o.filled_minor ?? 0,
        o.remaining_minor, o.fee_total_minor ?? null, o.fee_recognized_minor ?? null,
        o.refunded_minor ?? null, o.settled_splits ?? null, o.metadata.payout_id]);

    // 2) บรรทัด fill — idempotent ด้วย leg_ref · ทำได้แม้ยอดรวมจะเก่ากว่า
    //    (บรรทัดเป็น append-only ไม่ขัดกัน — ต่างจากยอดรวมที่ต้อง guard)
    const fill = event.data.fill;   // fill อยู่ข้าง data.object ไม่ได้อยู่ใน object
    if (fill) {
      await t.query(`
        INSERT INTO pg_withdrawal_fills (leg_ref, pg_withdrawal_id, sequence, amount_minor,
                                         fee_recognized_minor, settled_at, auto_confirm_at,
                                         balance_transaction_id)
        VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
        ON CONFLICT (leg_ref) DO NOTHING
      `, [fill.leg_ref, o.id, fill.sequence, fill.amount_minor,
          fill.fee_recognized_minor, fill.settled_at, fill.auto_confirm_at,
          fill.balance_transaction_id]);
    }

    // 3) ปิดรายการได้ทางเดียวเท่านั้น — และต้องอ่านค่าที่ผ่าน guard แล้วจาก DB ของเราเอง
    if (rowCount === 1 && o.status === 'settled' && o.remaining_minor === 0) {
      const closed = await t.query(`UPDATE pg_withdrawals SET closed_at = now()
                      WHERE pg_withdrawal_id = $1 AND status = 'settled'
                        AND remaining_minor = 0 AND closed_at IS NULL`, [o.id]);
      if (closed.rowCount === 1) {
        await enqueueEffect(t, `withdrawal:${o.id}`, 'close_payout', {
          payout_id: o.metadata.payout_id });
      }
    }

    await t.query(`UPDATE pg_webhook_inbox SET processed_at = now()
                    WHERE event_id = $1 AND processed_at IS NULL`, [event.id]);
  });
}

สามข้อห้ามของฝั่งถอน

• 🔴 ห้าม SUM(pg_withdrawal_fills.amount_minor) ขับ business logic — event อาจยังมาไม่ครบ คุณจะได้ค่าต่ำกว่าความจริง แล้วสรุปว่า "ยังจ่ายไม่ครบ" ทั้งที่ครบแล้ว · ตาราง fill มีไว้ทำ timeline/audit เท่านั้น
• 🔴 ห้าม filled_minor += fill.amount_minor — delivery ซ้ำจะบวกเกิน · เขียนทับด้วยค่าจาก payload ที่ผ่าน version guard เท่านั้น
• 🔴 ห้ามปิดรายการตอนได้ withdrawal.filled ใบแรก — ปิดเมื่อ status = 'settled' และ remaining_minor = 0

ทำไมบรรทัด fill ไม่ต้อง version guard

เพราะมันเป็น บรรทัดใหม่ ไม่ใช่การเขียนทับค่าเดิม · leg_ref เป็น primary key ⇒ ON CONFLICT DO NOTHING ทำให้ delivery ซ้ำไม่สร้างบรรทัดซ้ำ

ต่างจากยอดรวมที่เป็น "ค่าเดียวที่ถูกเขียนทับเรื่อย ๆ" ซึ่งเป็นจุดที่ลำดับมีความหมาย

Object immutable — dedupe ด้วย id แล้วสะสม

balance_transaction (btxn_*) ไม่มี version ไม่มี previous_attributes เพราะมันเป็น append-only

async function applyImmutable(event) {
  const b = event.data.object;   // btxn_...
  await db.tx(async (t) => {
  await t.query(`
    INSERT INTO pg_balance_txns (btxn_id, type, direction, amount_minor,
                                 bucket_from, bucket_to, source_id, created_at)
    VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
    ON CONFLICT (btxn_id) DO NOTHING
  `, [b.id, b.type, b.direction, b.amount_minor, b.bucket_from, b.bucket_to,
      b.source?.id ?? null, b.created_at]);
  await t.query(`UPDATE pg_webhook_inbox SET processed_at = now()
                  WHERE event_id = $1 AND processed_at IS NULL`, [event.id]);
  });
}

ห้ามใช้ balance_after เป็นยอดคงเหลือของระบบคุณ

balance_after เป็น snapshot ณ ตอนที่ transaction นั้นเกิด · ถ้า event สลับลำดับ การเขียนทับจะได้ยอดเก่า

ใช้เพื่อ ตรวจทาน เท่านั้น (เทียบกับผลรวมที่คุณคำนวณเอง แล้ว alert ถ้าไม่ตรง) · ยอดจริงดึงจาก GET /v1/pay/balance

และอย่าลืมว่า balance.cleared คือการ ย้าย bucket ไม่ใช่เงินเข้าใหม่ — บวกยอดตอนนั้นอีกครั้ง = ยอดสองเท่า

Gap ของ sequence

if (event.sequence > maxSeen + 1) {
  metrics.gauge('pg.webhook.gap', event.sequence - maxSeen - 1);   // เตือนเฉย ๆ
}
maxSeen = Math.max(maxSeen, event.sequence);

gap ไม่ใช่หลักฐานว่าหาย

event ที่ยังอยู่ในรอบ retry ทำให้เกิด gap ชั่วคราวได้นานถึง 92 ชั่วโมง

🔴 ห้ามหยุดประมวลผลเพื่อรอ event ที่หายไป — จะค้างถาวร · gap ที่ยังค้างเกิน 4 วันเท่านั้นที่แปลว่าหายจริง ให้ไปดึงด้วย GET /v1/pay/events?since_sequence=

เช็กลิสต์ของหน้านี้

• [ ] ทุกการเขียนสถานะเป็น atomic UPDATE ที่มี WHERE version < ใหม่ ไม่ใช่ SELECT-then-UPDATE
• [ ] side effect ทุกตัวผูกกับ rowCount ของ UPDATE นั้น ไม่ใช่ผูกกับ event.type
• [ ] version guard + business DB effect + outbox + processed_at commit ใน transaction เดียว
• [ ] service ภายนอกถูกเรียกโดย outbox worker พร้อม idempotency key เท่านั้น
• [ ] rowCount = 0 → ตอบ 200 ไม่ทำอะไรต่อ
• [ ] การเครดิตผู้ใช้ผูกกับ credited_user_at IS NULL อีกชั้น และใช้ amount_received_minor
• [ ] บรรทัด fill ใช้ leg_ref เป็น primary key + ON CONFLICT DO NOTHING
• [ ] filled_minor / remaining_minor มาจาก payload ที่ผ่าน guard — ไม่ SUM ไม่ +=
• [ ] balance_transaction dedupe ด้วย btxn_id และไม่ใช้ balance_after เป็นยอดจริง
• [ ] expired / canceled เช็ค is_terminal + late_slip_accepted_until ก่อนปิดออร์เดอร์
• [ ] มี default: ในทุก switch

ขั้นต่อไป

อ่านต่อ	เมื่อ
ทดสอบก่อนขึ้น production	เขียนเสร็จแล้ว — มีเทสต์ที่ยิง event สลับลำดับให้ทดสอบข้อนี้โดยเฉพาะ
Retry และ endpoint ล่ม	อยากรู้ว่าทำไม event เก่าถึงมาช้าได้ถึง 92 ชั่วโมง
แคตตาล็อก event	ต้องการดูว่า event ไหนมี field อะไรให้ guard
สถานะรายการฝาก	ต้องการ map สถานะ → หน้าจอผู้ใช้
รับ fill หลายก้อน	ทำฝั่งถอนต่อ