Scoreboard/server/internal/httpapi/idempotency.go

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)
}