feat: add authenticated queue HTTP API

2026-09-11 00:14:00 +00:00 · 2026-08-31 21:13:10 +01:00
parent caa875f15c
commit cd30098eca
3 changed files with 305 additions and 1 deletions
@@ -1191,7 +1191,7 @@ the local/CI/community transport, not a silent production fallback.

 | # | Task | Acceptance |
 |---|---|---|
-| 8.14 `[D:8.4,8.5,8.8]` | **IN PROGRESS.** Pure Go queue domain enforces one active ticket per verified player under concurrent mutation, 10 s heartbeat/30 s expiry, retry-safe create/heartbeat/cancel and deterministic candidate projection; store layer adds a rebuildable candidate-cache boundary | `server/domain/queue.go` and `server/store/candidates.go` cover ownership/expiry/idempotency, concurrent create fencing, deterministic projection, cache loss and atomic rebuild; PostgreSQL row adapter, real Redis index/TTLs and restart/failover integration remain |
+| 8.14 `[D:8.4,8.5,8.8]` | **IN PROGRESS.** Pure Go queue domain enforces one active ticket per verified player under concurrent mutation, 10 s heartbeat/30 s expiry, retry-safe create/heartbeat/cancel and deterministic candidate projection; store layer adds a rebuildable candidate-cache boundary and authenticated HTTP queue adapter | `server/domain/queue.go`, `server/store/candidates.go` and `server/api/service.go` cover ownership/expiry/idempotency, concurrent create fencing, server-owned candidate resolution, bounded/strict JSON input, cache loss and atomic rebuild; PostgreSQL row adapter, real Redis index/TTLs and restart/failover integration remain |
 | 8.15 `[D:7.8,8.3]` | **IN PROGRESS.** Pure Go probe validation treats Steam location as opaque, requires nonce/freshness/region and server-computed RTT, and implements discrepancy quarantine/release | `server/domain/probes.go` and adversarial fixtures cover stale/wrong/forged evidence, the 25 ms/30% threshold, three-sample quarantine and five-clean release; Steam coordinator and regional probe adapters remain |
 | 8.16 `[D:8.14,8.15]` | **IN PROGRESS.** Pure Go candidate/team selection implements the <=100 ms ceiling, pairwise widening tolerance, anchor inclusion, deterministic set/region scoring and balanced team partitioning | `server/domain/matcher.go`, `teams.go` and adversarial fixtures cover no-common-region, tolerance boundaries, lexical ties, mean-rating balance and input-order independence; queue-backed candidate loading and full population fixtures remain |
 | 8.17 `[D:8.14,8.16]` | **IN PROGRESS.** Pure Go proposal policy sends a 10-second response window to every selected human, requires unanimous acceptance, applies exact decline/timeout cooldowns and ranked escalation | `server/domain/proposal.go` and adversarial fixtures cover partial/unanimous response, expiry, replay/conflict and ranked six-player/cooldown rules; casual 6→2 composition, queue precedence and allocation integration remain |
@@ -0,0 +1,202 @@
+// Package api exposes the small authenticated HTTP boundary around domain
+// policies. Persistent adapters can replace the in-memory dependencies without
+// changing request authentication or validation rules.
+package api
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+const maxBodyBytes = 8 << 10
+
+type CandidateProvider func(playerID, ticketID string) (domain.Candidate, error)
+
+type Service struct {
+	Sessions  *domain.SessionStore
+	Queue     *domain.Queue
+	Candidate CandidateProvider
+	Now       func() time.Time
+}
+
+func (s *Service) Handler() http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/healthz", s.health)
+	mux.HandleFunc("/v1/queue", s.queueCreate)
+	mux.HandleFunc("/v1/queue/", s.queueMutation)
+	return mux
+}
+
+func (s *Service) health(w http.ResponseWriter, _ *http.Request) {
+	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
+}
+
+type queueCreateRequest struct {
+	TicketID string `json:"ticket_id"`
+}
+type queueResponse struct {
+	TicketID   string    `json:"ticket_id"`
+	PlayerID   string    `json:"player_id"`
+	State      string    `json:"state"`
+	Revision   uint64    `json:"revision"`
+	EnqueuedAt time.Time `json:"enqueued_at"`
+	ExpiresAt  time.Time `json:"expires_at"`
+}
+
+func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
+		return
+	}
+	playerID, ok := s.authenticate(w, r)
+	if !ok {
+		return
+	}
+	if s.Queue == nil || s.Candidate == nil {
+		writeError(w, http.StatusServiceUnavailable, "queue_unavailable")
+		return
+	}
+	var input queueCreateRequest
+	if !decodeBody(w, r, &input) {
+		return
+	}
+	if input.TicketID == "" {
+		writeError(w, http.StatusBadRequest, "invalid_request")
+		return
+	}
+	key := r.Header.Get("Idempotency-Key")
+	if len(key) < 16 || len(key) > 128 {
+		writeError(w, http.StatusBadRequest, "invalid_idempotency_key")
+		return
+	}
+	now := s.now()
+	candidate, err := s.Candidate(playerID, input.TicketID)
+	if err != nil {
+		writeError(w, http.StatusUnprocessableEntity, "candidate_unavailable")
+		return
+	}
+	ticket, err := s.Queue.Create(playerID, input.TicketID, key, candidate, now)
+	if err != nil {
+		writeDomainError(w, err)
+		return
+	}
+	writeJSON(w, http.StatusCreated, toQueueResponse(ticket))
+}
+
+func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
+		return
+	}
+	playerID, ok := s.authenticate(w, r)
+	if !ok {
+		return
+	}
+	if s.Queue == nil {
+		writeError(w, http.StatusServiceUnavailable, "queue_unavailable")
+		return
+	}
+	parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v1/queue/"), "/")
+	if len(parts) != 2 || parts[0] == "" || (parts[1] != "heartbeat" && parts[1] != "cancel") {
+		writeError(w, http.StatusNotFound, "not_found")
+		return
+	}
+	ticketID, key := parts[0], r.Header.Get("Idempotency-Key")
+	if len(key) < 16 || len(key) > 128 {
+		writeError(w, http.StatusBadRequest, "invalid_idempotency_key")
+		return
+	}
+	revision, err := strconv.ParseUint(r.Header.Get("If-Match-Revision"), 10, 64)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "invalid_revision")
+		return
+	}
+	now := s.now()
+	var ticket domain.QueueTicket
+	if parts[1] == "heartbeat" {
+		ticket, err = s.Queue.Heartbeat(playerID, ticketID, key, revision, now)
+	} else {
+		ticket, err = s.Queue.Cancel(playerID, ticketID, key, revision, now)
+	}
+	if err != nil {
+		writeDomainError(w, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, toQueueResponse(ticket))
+}
+
+func (s *Service) authenticate(w http.ResponseWriter, r *http.Request) (string, bool) {
+	if s.Sessions == nil {
+		writeError(w, http.StatusServiceUnavailable, "auth_unavailable")
+		return "", false
+	}
+	parts := strings.Fields(r.Header.Get("Authorization"))
+	if len(parts) != 2 || parts[0] != "Bearer" {
+		writeError(w, http.StatusUnauthorized, "unauthorized")
+		return "", false
+	}
+	separator := strings.IndexByte(parts[1], ':')
+	if separator <= 0 || separator == len(parts[1])-1 {
+		writeError(w, http.StatusUnauthorized, "unauthorized")
+		return "", false
+	}
+	session, err := s.Sessions.Authenticate(parts[1][:separator], parts[1][separator+1:], s.now())
+	if err != nil {
+		writeError(w, http.StatusUnauthorized, "unauthorized")
+		return "", false
+	}
+	return session.PlayerID, true
+}
+
+func (s *Service) now() time.Time {
+	if s.Now != nil {
+		return s.Now()
+	}
+	return time.Now().UTC()
+}
+
+func decodeBody(w http.ResponseWriter, r *http.Request, target any) bool {
+	r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes)
+	decoder := json.NewDecoder(r.Body)
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(target); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid_request")
+		return false
+	}
+	return true
+}
+
+func toQueueResponse(ticket domain.QueueTicket) queueResponse {
+	return queueResponse{TicketID: ticket.TicketID, PlayerID: ticket.PlayerID, State: string(ticket.State), Revision: ticket.Revision, EnqueuedAt: ticket.EnqueuedAt, ExpiresAt: ticket.ExpiresAt}
+}
+
+func writeDomainError(w http.ResponseWriter, err error) {
+	switch {
+	case errors.Is(err, domain.ErrPlayerQueued), errors.Is(err, domain.ErrConflict), errors.Is(err, domain.ErrStaleRevision):
+		writeError(w, http.StatusConflict, "conflict")
+	case errors.Is(err, domain.ErrTicketExpired):
+		writeError(w, http.StatusGone, "expired")
+	case errors.Is(err, domain.ErrNotTicketOwner):
+		writeError(w, http.StatusForbidden, "forbidden")
+	case errors.Is(err, domain.ErrTicketNotFound):
+		writeError(w, http.StatusNotFound, "not_found")
+	default:
+		writeError(w, http.StatusUnprocessableEntity, "invalid_request")
+	}
+}
+
+func writeError(w http.ResponseWriter, status int, code string) {
+	writeJSON(w, status, map[string]string{"error": code})
+}
+
+func writeJSON(w http.ResponseWriter, status int, value any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(value)
+}
@@ -0,0 +1,102 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+func TestAuthenticatedQueueAPIUsesServerCandidateAndRevisionedMutations(t *testing.T) {
+	now := time.Unix(1000, 0).UTC()
+	sessions := domain.NewSessionStore()
+	session, token, err := sessions.Issue("player-1", time.Hour, now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	queue := domain.NewQueue()
+	service := &Service{Sessions: sessions, Queue: queue, Now: func() time.Time { return now }, Candidate: func(playerID, ticketID string) (domain.Candidate, error) {
+		return domain.Candidate{PlayerID: playerID, TicketID: ticketID, EnqueuedAt: now, PredictedRTT: map[string]float64{"EU": 20}}, nil
+	}}
+	server := httptest.NewServer(service.Handler())
+	defer server.Close()
+	request := func(method, path, body string, headers map[string]string) *http.Response {
+		req, _ := http.NewRequest(method, server.URL+path, strings.NewReader(body))
+		for key, value := range headers {
+			req.Header.Set(key, value)
+		}
+		response, requestErr := http.DefaultClient.Do(req)
+		if requestErr != nil {
+			t.Fatal(requestErr)
+		}
+		return response
+	}
+	headers := map[string]string{"Authorization": "Bearer " + session.SessionID + ":" + token, "Idempotency-Key": "create-key-123456"}
+	response := request(http.MethodPost, "/v1/queue", `{"ticket_id":"ticket-1"}`, headers)
+	if response.StatusCode != http.StatusCreated {
+		t.Fatalf("create status = %d", response.StatusCode)
+	}
+	var created queueResponse
+	if err := json.NewDecoder(response.Body).Decode(&created); err != nil {
+		t.Fatal(err)
+	}
+	_ = response.Body.Close()
+	if created.PlayerID != "player-1" || created.State != "QUEUED" || created.Revision != 0 {
+		t.Fatalf("created = %+v", created)
+	}
+	response = request(http.MethodPost, "/v1/queue/ticket-1/heartbeat", `{}`, map[string]string{"Authorization": headers["Authorization"], "Idempotency-Key": "heartbeat-key-123456", "If-Match-Revision": "0"})
+	if response.StatusCode != http.StatusOK {
+		t.Fatalf("heartbeat status = %d", response.StatusCode)
+	}
+	_ = response.Body.Close()
+	response = request(http.MethodPost, "/v1/queue/ticket-1/cancel", `{}`, map[string]string{"Authorization": headers["Authorization"], "Idempotency-Key": "cancel-key-123456", "If-Match-Revision": "0"})
+	if response.StatusCode != http.StatusConflict {
+		t.Fatalf("stale cancel status = %d", response.StatusCode)
+	}
+	_ = response.Body.Close()
+}
+
+func TestQueueAPIRejectsUnauthenticatedUnknownAndOversizedInput(t *testing.T) {
+	service := &Service{Sessions: domain.NewSessionStore(), Queue: domain.NewQueue(), Candidate: func(string, string) (domain.Candidate, error) { return domain.Candidate{}, nil }}
+	server := httptest.NewServer(service.Handler())
+	defer server.Close()
+	request, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/queue", strings.NewReader(`{"ticket_id":"ticket-1","player_id":"attacker"}`))
+	request.Header.Set("Idempotency-Key", "create-key-123456")
+	response, err := http.DefaultClient.Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if response.StatusCode != http.StatusUnauthorized {
+		t.Fatalf("unauthenticated status = %d", response.StatusCode)
+	}
+	_ = response.Body.Close()
+	sessionStore := domain.NewSessionStore()
+	session, token, _ := sessionStore.Issue("player-1", time.Hour, time.Now())
+	service.Sessions = sessionStore
+	request, _ = http.NewRequest(http.MethodPost, server.URL+"/v1/queue", strings.NewReader(`{"ticket_id":"ticket-1","unknown":true}`))
+	request.Header.Set("Authorization", "Bearer "+session.SessionID+":"+token)
+	request.Header.Set("Idempotency-Key", "create-key-123456")
+	response, err = http.DefaultClient.Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if response.StatusCode != http.StatusBadRequest {
+		t.Fatalf("unknown field status = %d", response.StatusCode)
+	}
+	_ = response.Body.Close()
+	request, _ = http.NewRequest(http.MethodPost, server.URL+"/v1/queue", strings.NewReader(`{"ticket_id":`))
+	request.Header.Set("Authorization", "Bearer "+session.SessionID+":"+token)
+	request.Header.Set("Idempotency-Key", "create-key-654321")
+	response, err = http.DefaultClient.Do(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if response.StatusCode != http.StatusBadRequest {
+		t.Fatalf("malformed body status = %d", response.StatusCode)
+	}
+	_ = response.Body.Close()
+}