顾客储值卡余额和后台流水对不上：余额扣减的幂等、流水对账与补偿实践

CREATE TABLE card_ledger (
  id          BIGINT PRIMARY KEY AUTO_INCREMENT,
  card_no     VARCHAR(32) NOT NULL,
  biz_type    VARCHAR(16) NOT NULL,      -- RECHARGE/PAY/REFUND/ADJUST
  amount      DECIMAL(10,2) NOT NULL,    -- 正为入、负为出
  balance_after DECIMAL(10,2) NOT NULL,  -- 该笔后的余额快照，方便核对
  store_id    BIGINT NOT NULL,
  operator_id BIGINT NOT NULL,
  biz_no      VARCHAR(40) NOT NULL,      -- 业务单号，幂等键
  created_at  DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  UNIQUE KEY uk_biz (biz_type, biz_no)   -- 同类型同单号只允许一笔
);

def card_pay(card_no, amount, biz_no):
    # 幂等：同一业务单号只入账一次，重复提交直接返回原结果
    if db.exists("SELECT 1 FROM card_ledger WHERE biz_type='PAY' AND biz_no=%s", biz_no):
        return {"dedup": True}
    # 原子条件更新：余额足够才扣，防止超扣成负数
    updated = db.execute(
        "UPDATE card_balance SET balance=balance-%s "
        "WHERE card_no=%s AND balance>=%s",
        (amount, card_no, amount)
    )
    if not updated:
        return {"error": "insufficient_balance"}
    db.insert("card_ledger", card_no=card_no, biz_type="PAY",
              amount=-amount, biz_no=biz_no, balance_after=query_balance(card_no))
    return {"ok": True}

def card_pay_tx(card_no, amount, biz_no):
    with db.transaction():
        # 行级锁：SELECT FOR UPDATE 串行化同一张卡的并发扣减
        bal = db.fetchone("SELECT balance FROM card_balance WHERE card_no=%s FOR UPDATE", card_no)
        if bal.balance < amount:
            raise InsufficientBalance()
        db.execute("UPDATE card_balance SET balance=balance-%s WHERE card_no=%s", (amount, card_no))
        db.insert("card_ledger", card_no=card_no, biz_type="PAY", amount=-amount, biz_no=biz_no,
                  balance_after=bal.balance - amount)

def daily_reconcile():
    rows = db.query(
        "SELECT card_no, balance, "
        "(SELECT COALESCE(SUM(amount),0) FROM card_ledger l WHERE l.card_no=b.balance.card_no) AS ledger_sum "
        "FROM card_balance b"
    )
    for r in rows:
        if abs(r.balance - r.ledger_sum) > 0.005:
            # 差额超过一分钱：记对账异常，先冻结该卡交易，人工复核后补正
            db.insert("reconcile_exception", card_no=r.card_no,
                      diff=r.balance - r.ledger_sum, status="OPEN")
            db.execute("UPDATE card_balance SET frozen=1 WHERE card_no=%s", r.card_no)