幂等键不是防重开关：重放要原样返回上一次的响应

CREATE TABLE idempotency_keys (
    id              BIGSERIAL PRIMARY KEY,
    user_id         BIGINT      NOT NULL,
    op              TEXT        NOT NULL,       -- 操作类型：order.create / payment.callback
    idem_key        TEXT        NOT NULL,       -- 客户端提供的键
    request_hash    TEXT        NOT NULL,       -- 请求体指纹：同键不同参的检测依据
    status          TEXT        NOT NULL DEFAULT 'processing',
    response_status INT,                        -- 重放时还原的状态码
    response_body   JSONB,                      -- 重放时还原的响应体
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    expires_at      TIMESTAMPTZ NOT NULL        -- TTL 依据，必须长于最大重试窗口
);

-- 作用域 = 用户 + 操作 + 键；全局唯一键会让 A 用户的键挡住 B 用户的同名键
CREATE UNIQUE INDEX uq_idempotency ON idempotency_keys (user_id, op, idem_key);

package idemrepo

import (
    "context"

    "github.com/jackc/pgx/v5"
)

type Record struct {
    Status         string
    ResponseStatus int
    ResponseBody   []byte
    RequestHash    string
}

// Acquire 在事务内抢占幂等键。
// first=true 表示我们是第一个到达的请求，业务继续在本事务内执行
func (r *Repo) Acquire(ctx context.Context, tx pgx.Tx, userID int64, op, key, reqHash string) (rec Record, first bool, err error) {
    tag, err := tx.Exec(ctx, `
        INSERT INTO idempotency_keys (user_id, op, idem_key, request_hash, expires_at)
        VALUES ($1, $2, $3, $4, now() + interval '72 hours')
        ON CONFLICT (user_id, op, idem_key) DO NOTHING`,
        userID, op, key, reqHash)
    if err != nil {
        return rec, false, err
    }
    first = tag.RowsAffected() == 1

    // 冲突时读出来的是先到请求已提交的结果
    err = tx.QueryRow(ctx, `
        SELECT status, COALESCE(response_status, 0), response_body, request_hash
        FROM idempotency_keys
        WHERE user_id = $1 AND op = $2 AND idem_key = $3`,
        userID, op, key).Scan(&rec.Status, &rec.ResponseStatus, &rec.ResponseBody, &rec.RequestHash)
    return rec, first, err
}

// Finish 把最终响应写回幂等记录，与业务变更同一个事务提交
func (r *Repo) Finish(ctx context.Context, tx pgx.Tx, userID int64, op, key, status string, httpStatus int, body []byte) error {
    _, err := tx.Exec(ctx, `
        UPDATE idempotency_keys
        SET status = $4, response_status = $5, response_body = $6
        WHERE user_id = $1 AND op = $2 AND idem_key = $3`,
        userID, op, key, status, httpStatus, body)
    return err
}

func (s *Server) handleCreateOrder(w http.ResponseWriter, r *http.Request) {
    key := r.Header.Get("Idempotency-Key")
    if key == "" || len(key) > 128 {
        http.Error(w, `{"error":"missing Idempotency-Key"}`, http.StatusBadRequest)
        return
    }
    body, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20))
    reqHash := sha256Hex(body) // 对原始 body 取指纹，见「同键不同 body」一节
    userID := currentUser(r)

    // 同键请求正在执行时，第二个请求会在唯一索引上等第一个事务提交；
    // 给这一步单独设短超时，等待超限就让路，别耗尽整个请求的预算
    acqCtx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
    defer cancel()

    tx, err := s.pool.Begin(acqCtx)
    if err != nil {
        http.Error(w, `{"error":"internal"}`, http.StatusInternalServerError)
        return
    }
    defer tx.Rollback(context.Background()) // 成功路径 Commit 之后它是 no-op

    rec, first, err := s.idem.Acquire(acqCtx, tx, userID, "order.create", key, reqHash)
    if err != nil { // 多半是同键请求还在执行，等不到它提交
        w.Header().Set("Retry-After", "2")
        http.Error(w, `{"error":"idempotency_key_in_flight"}`, http.StatusConflict)
        return
    }

    if !first {
        if rec.Status == "processing" { // 两阶段变体下会出现：已有请求在途
            w.Header().Set("Retry-After", "2")
            http.Error(w, `{"error":"idempotency_key_in_flight"}`, http.StatusConflict)
            return
        }
        if rec.RequestHash != reqHash { // 键被复用，但请求不是同一个
            http.Error(w, `{"error":"key_reused_with_different_body"}`, http.StatusUnprocessableEntity)
            return
        }
        // 重放：原样还回上次的完整响应，包括上次的失败
        w.Header().Set("Content-Type", "application/json")
        w.WriteHeader(rec.ResponseStatus)
        _, _ = w.Write(rec.ResponseBody)
        return
    }

    order, bizErr := s.orders.CreateInTx(acqCtx, tx, userID, body)
    if bizErr != nil {
        if isDeterministic(bizErr) { // 参数错、余额不足：重放等价，失败也记下来
            resp := errorBody(bizErr)
            _ = s.idem.Finish(acqCtx, tx, userID, "order.create", key, "failed", bizErr.HTTPStatus, resp)
            _ = tx.Commit(acqCtx)
            w.WriteHeader(bizErr.HTTPStatus)
            _, _ = w.Write(resp)
            return
        }
        _ = tx.Rollback(r.Context()) // 瞬时故障：不留记录，客户端重试等价于首次执行
        http.Error(w, `{"error":"internal"}`, http.StatusInternalServerError)
        return
    }

    resp, _ := json.Marshal(order)
    _ = s.idem.Finish(acqCtx, tx, userID, "order.create", key, "succeeded", http.StatusCreated, resp)
    if err := tx.Commit(acqCtx); err != nil { // 提交失败：订单与幂等记录一起回滚，可安全重试
        http.Error(w, `{"error":"internal"}`, http.StatusInternalServerError)
        return
    }
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(http.StatusCreated)
    _, _ = w.Write(resp)
}

// 定时任务分批删除，避免大事务和长锁；expires_at 上要有索引
func (s *Server) cleanupExpired(ctx context.Context) error {
    _, err := s.pool.Exec(ctx, `
        DELETE FROM idempotency_keys
        WHERE id IN (
            SELECT id FROM idempotency_keys
            WHERE status <> 'processing' AND expires_at < now()
            LIMIT 5000
        )`)
    return err
}