diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 7d985404..fd9fe0f7 100644 --- a/multiplayer-todo.md +++ b/multiplayer-todo.md @@ -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 | diff --git a/server/api/service.go b/server/api/service.go new file mode 100644 index 00000000..f3dac66b --- /dev/null +++ b/server/api/service.go @@ -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) +} diff --git a/server/api/service_test.go b/server/api/service_test.go new file mode 100644 index 00000000..65e01df4 --- /dev/null +++ b/server/api/service_test.go @@ -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() +}