113 lines
3.8 KiB
Go
113 lines
3.8 KiB
Go
package middleware
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"errors"
|
|
apierrors "git.yuntuo.hk/k2safe/go-omni/errors"
|
|
"git.yuntuo.hk/k2safe/go-omni/httpx"
|
|
"git.yuntuo.hk/k2safe/go-omni/logx"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
type StoredResponse struct {
|
|
Status int `json:"status"`
|
|
ErrorCode string `json:"errorCode,omitempty"`
|
|
Headers http.Header `json:"headers"`
|
|
Body []byte `json:"body"`
|
|
}
|
|
type IdempotencyRecord struct {
|
|
Fingerprint string `json:"fingerprint"`
|
|
Owner string `json:"owner"`
|
|
State string `json:"state"`
|
|
Response *StoredResponse `json:"response,omitempty"`
|
|
}
|
|
|
|
// IdempotencyStore must atomically claim absent keys, compare owners when
|
|
// finishing and retain in-flight claims until explicit resolution. Only completed
|
|
// or failed records expire; an uncertain outcome must never become a new claim.
|
|
type IdempotencyStore interface {
|
|
Claim(context.Context, string, string, string) (IdempotencyRecord, bool, error)
|
|
Complete(context.Context, string, string, *StoredResponse, time.Duration) error
|
|
}
|
|
|
|
func (i *Idempotency) shared(c *httpx.Context, next httpx.Handler, id, fingerprint [32]byte) error {
|
|
key, fp := hex.EncodeToString(id[:]), hex.EncodeToString(fingerprint[:])
|
|
token := make([]byte, 16)
|
|
if _, err := rand.Read(token); err != nil {
|
|
return err
|
|
}
|
|
owner := hex.EncodeToString(token)
|
|
record, claimed, err := i.config.Store.Claim(c.Context(), key, fp, owner)
|
|
if err != nil {
|
|
return apierrors.New(503, "IDEMPOTENCY_UNAVAILABLE", "idempotency store unavailable").WithCause(err)
|
|
}
|
|
if !claimed {
|
|
if record.Fingerprint != fp {
|
|
return apierrors.New(409, "IDEMPOTENCY_CONFLICT", "key was used with a different request")
|
|
}
|
|
if record.State != "completed" || record.Response == nil {
|
|
return apierrors.New(409, "IDEMPOTENCY_UNAVAILABLE", "request is in progress or outcome requires reconciliation")
|
|
}
|
|
response := record.Response
|
|
headers := response.Headers.Clone()
|
|
headers.Set("Idempotency-Replayed", "true")
|
|
if response.ErrorCode != "" {
|
|
c.RecordError(apierrors.New(response.Status, response.ErrorCode, ""))
|
|
}
|
|
return c.WriteResponse(response.Status, headers, response.Body)
|
|
}
|
|
completed := false
|
|
finish := func(response *StoredResponse) error {
|
|
ctx, cancel := context.WithTimeout(context.WithoutCancel(c.Context()), time.Second)
|
|
defer cancel()
|
|
return i.config.Store.Complete(ctx, key, owner, response, i.config.TTL)
|
|
}
|
|
defer func() {
|
|
if !completed {
|
|
if err := finish(nil); err != nil {
|
|
logx.ErrorContext(c.Context(), "idempotency.persistFailed", "error", err)
|
|
}
|
|
}
|
|
}()
|
|
buffer := &boundedResponse{headers: c.Raw().Header().Clone(), limit: i.config.MaxResponseBytes}
|
|
child := c.Clone(c.Context(), buffer)
|
|
err = next(child)
|
|
if ctxErr := child.Context().Err(); ctxErr != nil {
|
|
return ctxErr
|
|
}
|
|
if err != nil {
|
|
child.WriteError(err)
|
|
}
|
|
if buffer.err != nil {
|
|
return apierrors.Internal(buffer.err)
|
|
}
|
|
if buffer.status == 0 {
|
|
buffer.status = 200
|
|
}
|
|
if failure := child.Failure(); failure != nil {
|
|
c.RecordError(failure)
|
|
}
|
|
var response *StoredResponse
|
|
if buffer.status < 500 && buffer.headers.Get("Set-Cookie") == "" && !strings.Contains(strings.ToLower(buffer.headers.Get("Cache-Control")), "no-store") {
|
|
response = &StoredResponse{Status: buffer.status, Headers: buffer.headers.Clone(), Body: append([]byte(nil), buffer.body.Bytes()...)}
|
|
if failure := child.Failure(); failure != nil {
|
|
response.ErrorCode = failure.Code
|
|
}
|
|
}
|
|
if err = finish(response); err != nil {
|
|
return apierrors.New(503, "IDEMPOTENCY_UNAVAILABLE", "outcome requires reconciliation").WithCause(err)
|
|
}
|
|
completed = true
|
|
return c.WriteResponse(buffer.status, buffer.headers, buffer.body.Bytes())
|
|
}
|
|
|
|
var ErrLimitCapacity = errors.New("middleware: shared rate limiter capacity")
|
|
|
|
type RateLimitBackend interface {
|
|
Take(context.Context, string) (time.Duration, error)
|
|
}
|