package httpapi import ( "bytes" "encoding/json" "errors" "fmt" "net/http" "sync" "time" "git.yeij.top/AskaEth/Scoreboard/server/internal/middleware" ) // errHandled 表示响应已经由 handler 自己写出(例如幂等回放),s.handle 不再重复写。 var errHandled = errors.New("httpapi: response already written") // idempotencyStore 缓存写操作的首次响应(见 devdocs/05 第 5 节:重复提交返回首次结果)。 // 迁移里没有幂等表,且不允许改 migrations,所以用进程内 TTL 缓存; // 进程重启后客户端重试会重新执行,属于可接受的降级。 type idempotencyStore struct { mu sync.Mutex entries map[string]idempotentResponse inflight map[string]chan struct{} ttl time.Duration } type idempotentResponse struct { body []byte expires time.Time } func newIdempotencyStore(ttl time.Duration) *idempotencyStore { if ttl <= 0 { ttl = 10 * time.Minute } return &idempotencyStore{ entries: make(map[string]idempotentResponse), inflight: make(map[string]chan struct{}), ttl: ttl, } } // begin 声明某个 key 的执行权:leader=true 表示由本请求执行, // 否则返回等待通道——等首次执行结束后回放其结果,避免同一 requestId 并发双执行。 func (s *idempotencyStore) begin(key string) (leader bool, wait <-chan struct{}) { s.mu.Lock() defer s.mu.Unlock() if ch, ok := s.inflight[key]; ok { return false, ch } ch := make(chan struct{}) s.inflight[key] = ch return true, ch } // end 释放执行权并唤醒等待者。 func (s *idempotencyStore) end(key string) { s.mu.Lock() defer s.mu.Unlock() if ch, ok := s.inflight[key]; ok { delete(s.inflight, key) close(ch) } } func (s *idempotencyStore) get(key string) ([]byte, bool) { s.mu.Lock() defer s.mu.Unlock() e, ok := s.entries[key] if !ok || time.Now().After(e.expires) { delete(s.entries, key) return nil, false } return e.body, true } func (s *idempotencyStore) put(key string, body []byte) { s.mu.Lock() defer s.mu.Unlock() now := time.Now() if len(s.entries) > 4096 { for k, e := range s.entries { if now.After(e.expires) { delete(s.entries, k) } } } s.entries[key] = idempotentResponse{body: body, expires: now.Add(s.ttl)} } // idempotent 以 requestId 为键执行写操作;重复请求直接回放首次响应。 // 返回值恒为 errHandled 或真正的错误——调用方写成 `return nil, s.idempotent(...)`。 func (s *server) idempotent(w http.ResponseWriter, r *http.Request, requestID string, fn func() (any, error)) error { if requestID == "" || s.idem == nil { data, err := fn() if err != nil { return err } writeJSON(w, http.StatusOK, map[string]any{"ok": true, "data": data}) return errHandled } key := fmt.Sprintf("%s|%d|%s", r.URL.Path, middlewareUserID(r), requestID) if body, ok := s.idem.get(key); ok { writeCached(w, body) return errHandled } data, err := fn() if err != nil { return err } body, err := json.Marshal(map[string]any{"ok": true, "data": data}) if err != nil { return internal(err) } s.idem.put(key, body) writeCached(w, body) return errHandled } // cached 给所有写路由加一层基于 X-Request-Id 的幂等缓存: // handler 自己写响应,这里负责在成功后留存响应体、重复请求时回放。 func (s *server) cached(inner http.Handler) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { requestID := r.Header.Get(middleware.HeaderRequestID) if requestID == "" || s.idem == nil { inner.ServeHTTP(w, r) return } key := fmt.Sprintf("%s|%d|%s", r.URL.Path, middlewareUserID(r), requestID) if body, ok := s.idem.get(key); ok { writeCached(w, body) return } // 占位锁只放在最外层:内层 idempotent 与这里共用同一把 key, // 若两处都抢占会自我等待。并发同一 requestId 由这里串行化并回放首次结果。 leader, wait := s.idem.begin(key) for !leader { <-wait if body, ok := s.idem.get(key); ok { writeCached(w, body) return } leader, wait = s.idem.begin(key) } defer s.idem.end(key) cw := &captureWriter{ResponseWriter: w, status: http.StatusOK} inner.ServeHTTP(cw, r) if cw.status >= 200 && cw.status < 300 && cw.body.Len() > 0 { s.idem.put(key, cw.body.Bytes()) } } } func writeCached(w http.ResponseWriter, body []byte) { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.Header().Set("X-Idempotent-Replay", "1") w.WriteHeader(http.StatusOK) _, _ = w.Write(body) } type captureWriter struct { http.ResponseWriter status int body bytes.Buffer } func (w *captureWriter) WriteHeader(code int) { w.status = code w.ResponseWriter.WriteHeader(code) } func (w *captureWriter) Write(b []byte) (int, error) { w.body.Write(b) return w.ResponseWriter.Write(b) }