textmachine/platform/internal/httpapi/idempotency.go

package httpapi

import (
	"bytes"
	"context"
	"crypto/sha256"
	"errors"
	"net/http"
	"time"

	"textmachine/platform/internal/pgstore"
)

// idempotency.go: `Idempotency-Key` on the three writes that create something (canon
// §IdempotencyKey).
//
// The semantics are entirely the server's duty and are written out here rather than inferred:
// scoped to (principal, method, path) and the same key on another operation is ANOTHER key · the
// same key with the same request replays the ORIGINAL response and does no new work · the same key
// with a DIFFERENT request is `409 key_reused` · a repeat while the first is in flight is
// `409 key_in_flight` with `Retry-After` · the record is kept at least 24 hours · a key over 255
// characters is `400` · omitting the header is legal and means no retry protection.

// IdempotencyKeys is the storage this needs. An interface so the HTTP layer keeps knowing nothing
// about SQL, and so an instance without a database simply has none.
type IdempotencyKeys interface {
	ClaimIdempotency(ctx context.Context, k pgstore.IdempotencyKey, now time.Time) (*pgstore.IdempotentResponse, pgstore.ClaimToken, error)
	CompleteIdempotency(ctx context.Context, k pgstore.IdempotencyKey, token pgstore.ClaimToken, resp pgstore.IdempotentResponse, now time.Time) error
	ReleaseIdempotency(ctx context.Context, k pgstore.IdempotencyKey, token pgstore.ClaimToken) error
}

// maxKeyLength is the contract's own bound; over it the request is refused rather than truncated.
const maxKeyLength = 255

// keyInFlightRetry is what a caller is told to wait when the first attempt under its key is still
// running. Deliberately short: the wait is for one request to finish, not for a translation.
const keyInFlightRetry = 2 * time.Second

// bodyIdentity says whether a request's identity depends on bytes its fingerprint cannot carry.
//
// A JSON write is compared whole, so the fingerprint IS the request. A multipart upload declares its
// parts and not its file, so two DIFFERENT books arrive under one key with the same declaration —
// and a replay decided on the declaration alone answers the second with the first one's book.
type bodyIdentity bool

const (
	fingerprintIsTheRequest bodyIdentity = false
	bodyDecidesIdentity     bodyIdentity = true
)

// idempotent is one claimed key, or nothing at all when the caller sent no header. `active` is
// cleared by whichever of complete/release runs first, so a handler can DEFER the release and every
// exit settles the key — a panic into the recovery middleware included.
type idempotent struct {
	h   *v0
	key pgstore.IdempotencyKey
	// token is this ATTEMPT's claim. Presented on both writes, so an attempt whose claim was taken
	// over can neither record its own receipt nor delete its successor's row.
	token  pgstore.ClaimToken
	active bool
	// replay is a stored answer this request may be given ONCE it has shown it is the same request:
	// set only where the body decides identity, and then the caller owes the comparison.
	replay *pgstore.IdempotentResponse
}

// beginIdempotent claims the key a write was presented with.
//
// The fingerprint is what makes "the same request" decidable, and what goes into it is the caller's
// choice because it differs by route: a JSON body is compared whole, while a multipart upload is
// compared over its DECLARED parts and never over the bytes — a server does not hold a book in
// memory to compare it, and a retry of an interrupted upload re-sends the same file.
//
// ok is false when the answer has already been written: a conflict, or a replay of the first
// attempt's own response.
func (h *v0) beginIdempotent(w http.ResponseWriter, r *http.Request, user string, fingerprint []byte,
	identity bodyIdentity) (*idempotent, bool) {
	raw := r.Header.Get("Idempotency-Key")
	if raw == "" || h.keys == nil {
		return &idempotent{}, true // legal, and it means no retry protection
	}
	if len(raw) > maxKeyLength {
		Invalid(w, r, Item{Pointer: "/Idempotency-Key", Code: ItemTooLong})
		return nil, false
	}
	// The REQUEST's path and not the route's pattern: the scope is (principal, method, path), and two
	// books' runs live at two paths. Keyed by the pattern, one key presented for two books would read
	// as the same key for a different request — a conflict where the client did nothing wrong.
	sum := sha256.Sum256(fingerprint)
	key := pgstore.IdempotencyKey{
		UserID: user, Method: r.Method, Path: r.URL.Path, Key: raw, Fingerprint: sum[:],
	}
	replay, token, err := h.keys.ClaimIdempotency(r.Context(), key, time.Now())
	switch {
	case errors.Is(err, pgstore.ErrKeyReused):
		FailCause(w, r, CodeIdempotencyConflict, CauseKeyReused)
		return nil, false
	case errors.Is(err, pgstore.ErrKeyContended):
		// The claim loop gave up: this key was taken and handed back under us as fast as the racers
		// could fail. The ANSWER is the same as an ordinary in-flight repeat, because the caller's
		// remedy is the same one — wait and present it again — and it must not be a 500: nothing was
		// wrong with the request and there is nothing its client could fix (PD-369). Ordered BEFORE
		// the case below because it wraps it.
		//
		// Loud on OUR side and quiet on the wire, which is the whole mitigation for "the fix hides a
		// storm": repeated contention on one key is an operator's signal, never a caller's fault.
		h.log.WarnContext(r.Context(), "an idempotency key is under repeated contention", "err", err)
		h.keyInFlight(w, r)
		return nil, false
	case errors.Is(err, pgstore.ErrKeyInFlight):
		h.keyInFlight(w, r)
		return nil, false
	case err != nil:
		h.log.ErrorContext(r.Context(), "the idempotency key could not be claimed", "err", err)
		Fail(w, r, CodeInternalError)
		return nil, false
	}
	if replay != nil {
		if identity == bodyDecidesIdentity && len(replay.ContentSHA256) > 0 {
			// NOT replayed here: what was compared is the DECLARATION, and this route's identity is the
			// file. The caller reads the body and `replayIfIdentical` decides from the digest — the one
			// answer that must not be given is the stored one to a request nobody has verified.
			return &idempotent{h: h, key: key, replay: replay}, true
		}
		h.replay(w, r, replay)
		return nil, false
	}
	return &idempotent{h: h, key: key, token: token, active: true}, true
}

// replay writes the first attempt's own answer, verbatim: same status, same body, same Location.
// Doing the work again is exactly what the header exists to prevent.
func (h *v0) replay(w http.ResponseWriter, r *http.Request, stored *pgstore.IdempotentResponse) {
	if stored.Location != "" {
		w.Header().Set("Location", stored.Location)
	}
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(stored.Status)
	if _, err := w.Write(stored.Body); err != nil {
		h.log.DebugContext(r.Context(), "replayed body not delivered", "err", err)
	}
}

// keyInFlight is the contract's answer to "this key is busy, come back": ONE writer, because the two
// paths that reach it — a first attempt still running, and a claim that lost the race too often —
// must not drift into two different words for one remedy.
func (h *v0) keyInFlight(w http.ResponseWriter, r *http.Request) {
	WriteProblem(w, r, Problem{Code: CodeIdempotencyConflict,
		Cause: &Cause{Code: CauseKeyInFlight}, RetryAfter: keyInFlightRetry})
}

// replayIfIdentical answers a repeat whose identity is its BODY: the stored answer when the bytes
// digest to what the first attempt accepted, and `key_reused` when they do not — which is what the
// canon calls a different request under the same key.
//
// The body is read either way, and that is the cost of an identity the declaration cannot carry. It
// buys the thing the declaration could not: a second, different book is never answered with the
// first one's address.
func (i *idempotent) replayIfIdentical(w http.ResponseWriter, r *http.Request, digest []byte) {
	if !bytes.Equal(digest, i.replay.ContentSHA256) {
		FailCause(w, r, CodeIdempotencyConflict, CauseKeyReused)
		return
	}
	i.h.replay(w, r, i.replay)
}

// settleCtx detaches a key's own write from the request: the client that will retry is precisely
// the one that hung up, and on `r.Context()` the key then stays in flight for `ClaimStale` and the
// retry re-does the work. Same rule as the intake's writeCtx, applied to the receipt.
func settleCtx(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
}

// complete records what this attempt answered, so a repeat is given the same thing. `content` is the
// digest of the body it accepted, for a route whose identity the fingerprint cannot settle; nil
// where it can.
func (i *idempotent) complete(ctx context.Context, status int, location string, body, content []byte) {
	if !i.active {
		return
	}
	i.active = false
	ctx, cancel := settleCtx(ctx)
	defer cancel()
	resp := pgstore.IdempotentResponse{Status: status, Location: location, Body: body, ContentSHA256: content}
	if err := i.h.keys.CompleteIdempotency(ctx, i.key, i.token, resp, time.Now()); err != nil {
		// The work IS done and the caller has its answer; what is lost is the ability to replay it, so
		// a retry does the work twice — or, on ErrClaimLost, a later attempt already did. Loud, and
		// not a failure of the request.
		i.h.log.ErrorContext(ctx, "the idempotency key could not be recorded", "err", err)
	}
}

// release gives the key back after an attempt that did not complete: a failed attempt is not a
// completed one, and the same key may be presented again. Safe to defer — see idempotent.
func (i *idempotent) release(ctx context.Context) {
	if !i.active {
		return
	}
	i.active = false
	ctx, cancel := settleCtx(ctx)
	defer cancel()
	if err := i.h.keys.ReleaseIdempotency(ctx, i.key, i.token); err != nil {
		i.h.log.ErrorContext(ctx, "the idempotency key could not be released", "err", err)
	}
}