1
package idempotency
2

3
import (
4
  "context"
5
  "net/http"
6
  "sync"
7
  "time"
8
)
9

10
// Status 幂等记录的状态
11
type Status string
12

13
const (
14
  StatusProcessing Status = "PROCESSING"
15
  StatusCompleted  Status = "COMPLETED"
16
)
17

18
// Record 幂等记录
19
type Record struct {
20
  Status    Status
21
  Result    *Response
22
  CreatedAt time.Time
23
  ExpiresAt time.Time
24
}
25

26
// Response 缓存的响应
27
type Response struct {
28
  StatusCode int
29
  Body       []byte
30
  Headers    http.Header
31
}
32

33
// Store 幂等键存储（内存版，生产环境应替换为 Redis / 数据库）
34
type Store struct {
35
  mu      sync.RWMutex
36
  records map[string]*Record
37
  ttl     time.Duration
38
}
39

40
func NewStore(ttl time.Duration) *Store {
41
  return &Store{
42
    records: make(map[string]*Record),
43
    ttl:     ttl,
44
  }
45
}
46

47
// Get 查询幂等记录
48
func (s *Store) Get(key string) (*Record, bool) {
49
  s.mu.RLock()
50
  defer s.mu.RUnlock()
51

52
  rec, exists := s.records[key]
53
  if !exists {
54
    return nil, false
55
  }
56
  // 过期记录视为不存在
57
  if time.Now().After(rec.ExpiresAt) {
58
    return nil, false
59
  }
60
  return rec, true
61
}
62

63
// SetProcessing 标记为处理中（防止并发重复请求）
64
func (s *Store) SetProcessing(key string) bool {
65
  s.mu.Lock()
66
  defer s.mu.Unlock()
67

68
  rec, exists := s.records[key]
69
  if exists && time.Now().Before(rec.ExpiresAt) {
70
    // 键已存在且未过期，不能重复处理
71
    return false
72
  }
73

74
  s.records[key] = &Record{
75
    Status:    StatusProcessing,
76
    CreatedAt: time.Now(),
77
    ExpiresAt: time.Now().Add(s.ttl),
78
  }
79
  return true
80
}
81

82
// SetCompleted 标记为完成，缓存结果
83
func (s *Store) SetCompleted(key string, result *Response) {
84
  s.mu.Lock()
85
  defer s.mu.Unlock()
86

87
  if rec, exists := s.records[key]; exists {
88
    rec.Status = StatusCompleted
89
    rec.Result = result
90
  }
91
}
92

93
// Cleanup 清理过期记录（定期调用）
94
func (s *Store) Cleanup() {
95
  s.mu.Lock()
96
  defer s.mu.Unlock()
97

98
  now := time.Now()
99
  for k, rec := range s.records {
100
    if now.After(rec.ExpiresAt) {
101
      delete(s.records, k)
102
    }
103
  }
104
}
105

106
// Middleware HTTP 中间件，包装任意 Handler
107
func Middleware(store *Store, headerKey string) func(http.Handler) http.Handler {
108
  return func(next http.Handler) http.Handler {
109
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
110
      key := r.Header.Get(headerKey)
111
      if key == "" {
112
        // 没有幂等键，直接放行
113
        next.ServeHTTP(w, r)
114
        return
115
      }
116

117
      // 查询已有记录
118
      rec, exists := store.Get(key)
119
      if exists {
120
        switch rec.Status {
121
        case StatusCompleted:
122
          // 已完成，直接返回缓存结果
123
          for k, v := range rec.Result.Headers {
124
            for _, vv := range v {
125
              w.Header().Add(k, vv)
126
            }
127
          }
128
          w.WriteHeader(rec.Result.StatusCode)
129
          w.Write(rec.Result.Body)
130
          return
131
        case StatusProcessing:
132
          // 正在处理中，返回 202 让客户端稍后重试
133
          w.WriteHeader(http.StatusAccepted)
134
          w.Write([]byte(`{"error":"request is being processed"}`))
135
          return
136
        }
137
      }
138

139
      // 标记为 PROCESSING
140
      if !store.SetProcessing(key) {
141
        // 并发冲突，另一个请求已经在处理
142
        w.WriteHeader(http.StatusConflict)
143
        w.Write([]byte(`{"error":"concurrent request detected"}`))
144
        return
145
      }
146

147
      // 捕获响应
148
      rw := &responseWriter{ResponseWriter: w}
149
      next.ServeHTTP(rw, r)
150

151
      // 缓存结果
152
      store.SetCompleted(key, &Response{
153
        StatusCode: rw.statusCode,
154
        Body:       rw.body,
155
        Headers:    rw.Header(),
156
      })
157
    })
158
  }
159
}
160

161
// responseWriter 包装 http.ResponseWriter，捕获响应内容
162
type responseWriter struct {
163
  http.ResponseWriter
164
  statusCode int
165
  body       []byte
166
}
167

168
func (rw *responseWriter) WriteHeader(code int) {
169
  rw.statusCode = code
170
  rw.ResponseWriter.WriteHeader(code)
171
}
172

173
func (rw *responseWriter) Write(b []byte) (int, error) {
174
  rw.body = append(rw.body, b...)
175
  return rw.ResponseWriter.Write(b)
176
}

1
import express from "express";
2
import Redis from "ioredis";
3

4
// 幂等键中间件，使用 Redis 作为后端存储
5
function idempotencyMiddleware(
6
  redis: Redis,
7
  headerKey = "Idempotency-Key",
8
  ttlSeconds = 86400, // 默认 24 小时过期
9
) {
10
  return async (
11
    req: express.Request,
12
    res: express.Response,
13
    next: express.NextFunction,
14
  ) => {
15
    const key = req.headers[headerKey.toLowerCase()] as string;
16
    if (!key) {
17
      // 没有幂等键，直接放行
18
      next();
19
      return;
20
    }
21

22
    const redisKey = `idempotency:${key}`;
23

24
    // 查询已有记录
25
    const existing = await redis.get(redisKey);
26
    if (existing) {
27
      const record = JSON.parse(existing);
28
      if (record.status === "COMPLETED") {
29
        // 已完成，直接返回缓存结果
30
        res.set(record.headers || {});
31
        res.status(record.statusCode).json(record.body);
32
        return;
33
      }
34
      if (record.status === "PROCESSING") {
35
        // 正在处理中，返回 202
36
        res.status(202).json({ error: "request is being processed" });
37
        return;
38
      }
39
    }
40

41
    // SETNX 抢占 PROCESSING 状态，防止并发
42
    const acquired = await redis.set(
43
      redisKey,
44
      JSON.stringify({ status: "PROCESSING", createdAt: Date.now() }),
45
      "EX",
46
      ttlSeconds,
47
      "NX",
48
    );
49

50
    if (!acquired) {
51
      // 另一个请求抢先了
52
      res.status(409).json({ error: "concurrent request detected" });
53
      return;
54
    }
55

56
    // 捕获响应
57
    const originalJson = res.json.bind(res);
58
    const originalStatus = res.status.bind(res);
59
    let capturedStatus = 200;
60
    let capturedBody: unknown;
61

62
    res.status = function (code: number) {
63
      capturedStatus = code;
64
      return originalStatus(code);
65
    };
66

67
    res.json = function (body: unknown) {
68
      capturedBody = body;
69
      // 缓存结果到 Redis
70
      redis.set(
71
        redisKey,
72
        JSON.stringify({
73
          status: "COMPLETED",
74
          statusCode: capturedStatus,
75
          body,
76
          headers: res.getHeaders(),
77
          completedAt: Date.now(),
78
        }),
79
        "EX",
80
        ttlSeconds,
81
      );
82
      return originalJson(body);
83
    };
84

85
    next();
86
  };
87
}
88

89
// 使用示例
90
const app = express();
91
const redis = new Redis({ host: "localhost", port: 6379 });
92

93
app.use(express.json());
94
app.use(idempotencyMiddleware(redis));
95

96
app.post("/pay", async (req, res) => {
97
  // 业务逻辑——即使重试，也只会执行一次
98
  const { amount, userId } = req.body;
99
  const orderId = await processPayment(userId, amount);
100
  res.json({ orderId });
101
});
102

103
async function processPayment(userId: string, amount: number): Promise<string> {
104
  // 模拟支付处理
105
  return `ord-${Date.now()}`;
106
}
107

108
app.listen(3000);