商城订单状态错乱：状态机、幂等消费与本地消息表的最终一致实践

// 显式定义合法流转：key 为当前状态，value 为允许到达的状态集合
const TRANSITIONS = {
  WAIT_PAY:    ['PAID', 'CANCELED'],
  PAID:        ['WAIT_SHIP', 'REFUNDING'],
  WAIT_SHIP:   ['SHIPPED', 'REFUNDING'],
  SHIPPED:     ['COMPLETED', 'REFUNDING'],
  COMPLETED:   ['REFUNDING'],
  REFUNDING:   ['REFUNDED', 'PAID', 'WAIT_SHIP'], // 退款驳回回到原态
  CANCELED:    [],
  REFUNDED:    [],
};

function canTransit(from, to) {
  return (TRANSITIONS[from] || []).includes(to);
}

// 唯一的状态变更入口：带条件更新，数据库层兜底防并发覆盖
async function changeStatus(conn, orderId, expectFrom, to, extra = {}) {
  if (!canTransit(expectFrom, to)) {
    throw new Error(`非法状态流转: ${expectFrom} -> ${to}`);
  }
  // 条件更新：只有当前状态仍是 expectFrom 才允许改，天然防旧状态覆盖
  const res = await conn.execute(
    'UPDATE orders SET status=?, updated_at=NOW() WHERE id=? AND status=?',
    [to, orderId, expectFrom]
  );
  if (res.affectedRows === 0) {
    // 状态已被别的流程改过，本次放弃（幂等返回，不报错刷屏）
    return { changed: false };
  }
  return { changed: true };
}

// 以支付结果消息为例：msgId 为消息唯一键，orderId+pay流水 为业务幂等键
async function consumePayMessage(msg) {
  const dedupKey = `consumed:${msg.msgId}`;
  // SET NX 抢占：只有第一个消费者能设置成功
  const first = await redis.set(dedupKey, '1', 'NX', 'EX', 86400);
  if (!first) {
    return ack(); // 重复消息，直接确认，不再处理
  }
  try {
    await db.transaction(async (conn) => {
      const order = await conn.query('SELECT status FROM orders WHERE id=? FOR UPDATE', [msg.orderId]);
      const r = await changeStatus(conn, msg.orderId, 'WAIT_PAY', 'PAID');
      if (r.changed) {
        await conn.execute('INSERT INTO pay_records(...) VALUES(...)', [/* 支付流水，唯一索引兜底 */]);
      }
    });
    ack();
  } catch (e) {
    await redis.del(dedupKey); // 处理失败释放幂等键，允许下次重试
    nack(true);                // 抛回让 MQ 重投
  }
}

-- 1. 本地消息表：和业务表在同一个库，靠本地事务保证原子
CREATE TABLE local_message (
  id           BIGINT PRIMARY KEY AUTO_INCREMENT,
  biz_id       VARCHAR(64)  NOT NULL,      -- 关联订单号
  topic        VARCHAR(64)  NOT NULL,
  payload      TEXT         NOT NULL,
  status       TINYINT      NOT NULL DEFAULT 0, -- 0待发送 1已发送 2已确认
  retry_count  INT          NOT NULL DEFAULT 0,
  next_retry   DATETIME     NOT NULL,
  created_at   DATETIME     NOT NULL,
  UNIQUE KEY uk_biz_topic (biz_id, topic)
);

// 2. 业务操作 + 写消息表，同一事务，要么都成要么都不成
async function paySuccess(conn, order) {
  await changeStatus(conn, order.id, 'WAIT_PAY', 'PAID');
  await conn.execute(
    'INSERT INTO local_message(biz_id,topic,payload,next_retry,created_at) VALUES(?,?,?,?,NOW())',
    [order.id, 'ORDER_PAID', JSON.stringify({ orderId: order.id }), now()]);
}

// 3. 后台投递器：轮询待发送消息，发送成功才标记，失败按退避重试
async function messageRelay() {
  const pending = await db.query(
    'SELECT * FROM local_message WHERE status=0 AND next_retry<=NOW() LIMIT 100 FOR UPDATE SKIP LOCKED');
  for (const m of pending) {
    try {
      await mq.publish(m.topic, m.payload);
      await db.execute('UPDATE local_message SET status=1 WHERE id=?', [m.id]);
    } catch (e) {
      await db.execute(
        'UPDATE local_message SET retry_count=retry_count+1, next_retry=DATE_ADD(NOW(),INTERVAL POW(2,LEAST(retry_count,5)) SECOND) WHERE id=?',
        [m.id]); // 指数退避，避免坏消息疯狂重试
    }
  }
}