From 59cf29949f6d5f9b91e506837d94a556d885386c Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Mon, 31 Aug 2026 22:17:32 +0100 Subject: [PATCH] feat: wire persistent queue backend into API --- multiplayer-todo.md | 2 +- server/api/service.go | 42 +++++++++++++++++++++++++++++++++----- server/api/service_test.go | 38 ++++++++++++++++++++++++++++++++++ server/store/queue_sql.go | 15 ++++++++++++++ 4 files changed, 91 insertions(+), 6 deletions(-) diff --git a/multiplayer-todo.md b/multiplayer-todo.md index eb71ff89..699f5675 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, owner-only recovery reads and deterministic candidate projection; store layer adds a rebuildable candidate-cache boundary and authenticated HTTP queue adapter with playlist/build/protocol compatibility metadata | `server/domain/queue.go`, `server/store/candidates.go`, `server/store/queue_sql.go` and `server/api/service.go` cover ownership/expiry/idempotency, concurrent create fencing, candidate/player ownership binding, owner/revision-scoped SQL heartbeat/cancel/recovery, authoritative queue-to-cache rebuild, expired recovery as a terminal error, owner-scoped SQL recovery with authoritative expiry handling, server-owned candidate resolution, strict compatibility metadata and cache loss/atomic rebuild; live PostgreSQL row execution, 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, owner-only recovery reads and deterministic candidate projection; store layer adds a rebuildable candidate-cache boundary and authenticated HTTP queue adapter with playlist/build/protocol compatibility metadata | `server/domain/queue.go`, `server/store/candidates.go`, `server/store/queue_sql.go` and `server/api/service.go` cover ownership/expiry/idempotency, concurrent create fencing, candidate/player ownership binding, owner/revision-scoped SQL heartbeat/cancel/recovery, injectable PostgreSQL queue backend selected by the HTTP service, authoritative queue-to-cache rebuild, expired recovery as a terminal error, owner-scoped SQL recovery with authoritative expiry handling, server-owned candidate resolution, strict compatibility metadata and cache loss/atomic rebuild; live PostgreSQL row execution, 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; authenticated HTTP now accepts only opaque location/nonce input through a server-owned probe provider | `server/domain/probes.go`, adversarial fixtures and `server/api/service.go` cover stale/wrong/forged evidence, the 25 ms/30% threshold, three-sample quarantine, five-clean release, authenticated provider arguments and rejection of client RTT fields; 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; queue-backed formation now consumes the server-owned projection, fences duplicate player identities and rejects playlist/build/protocol mixing | `server/domain/matcher.go`, `teams.go` and adversarial fixtures cover no-common-region, tolerance boundaries, lexical ties, mean-rating balance, malformed candidates, duplicate identities, compatibility mismatches and queue-backed oldest-anchor formation; full population fixtures and durable matcher claim integration 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; authenticated API exposes revisioned accept/decline mutations; formed matches now pass through a playlist-aware proposal boundary | `server/domain/proposal.go`, `formation.go` and `server/api/service.go` plus adversarial fixtures cover partial/unanimous response, expiry, replay/conflict, stale API revision, casual lineup preparation and ranked metadata validation; queue precedence and allocation integration remain | diff --git a/server/api/service.go b/server/api/service.go index 3ec40048..0a3dc641 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -4,6 +4,7 @@ package api import ( + "context" "encoding/json" "errors" "io" @@ -22,11 +23,19 @@ type CandidateProvider func(playerID, ticketID string) (domain.Candidate, error) type CandidateProviderV2 func(playerID, ticketID string, spec domain.QueueSpec) (domain.Candidate, error) type ProbeProvider func(playerID, region string, opaqueLocation, nonce []byte, receivedAt time.Time) (domain.ProbeEvidence, []byte, error) +type QueueBackend interface { + Create(context.Context, string, string, string, domain.QueueSpec, time.Time) (domain.QueueTicket, error) + Heartbeat(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error) + Cancel(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error) + Get(context.Context, string, string, time.Time) (domain.QueueTicket, error) +} + type Service struct { Sessions *domain.SessionStore Queue *domain.Queue Candidate CandidateProvider CandidateV2 CandidateProviderV2 + QueueBackend QueueBackend Probe ProbeProvider Now func() time.Time Proposals map[string]*domain.Proposal @@ -75,7 +84,7 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) { if !ok { return } - if s.Queue == nil || (s.Candidate == nil && s.CandidateV2 == nil) { + if (s.Queue == nil && s.QueueBackend == nil) || (s.QueueBackend == nil && s.Candidate == nil && s.CandidateV2 == nil) { writeError(w, http.StatusServiceUnavailable, "queue_unavailable") return } @@ -94,6 +103,15 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) { } now := s.now() spec := domain.QueueSpec{Playlist: domain.Playlist(input.Playlist), ClientBuild: input.ClientBuild, ProtocolVersion: input.ProtocolVersion} + if s.QueueBackend != nil { + ticket, err := s.QueueBackend.Create(r.Context(), playerID, input.TicketID, key, spec, now) + if err != nil { + writeDomainError(w, err) + return + } + writeJSON(w, http.StatusCreated, toQueueResponse(ticket)) + return + } var candidate domain.Candidate var err error if s.CandidateV2 != nil { @@ -131,7 +149,7 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) { if !ok { return } - if s.Queue == nil { + if s.Queue == nil && s.QueueBackend == nil { writeError(w, http.StatusServiceUnavailable, "queue_unavailable") return } @@ -141,7 +159,13 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusNotFound, "not_found") return } - ticket, err := s.Queue.Get(playerID, parts[0], s.now()) + var ticket domain.QueueTicket + var err error + if s.QueueBackend != nil { + ticket, err = s.QueueBackend.Get(r.Context(), playerID, parts[0], s.now()) + } else { + ticket, err = s.Queue.Get(playerID, parts[0], s.now()) + } if err != nil { writeDomainError(w, err) return @@ -166,9 +190,17 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) { now := s.now() var ticket domain.QueueTicket if parts[1] == "heartbeat" { - ticket, err = s.Queue.Heartbeat(playerID, ticketID, key, revision, now) + if s.QueueBackend != nil { + ticket, err = s.QueueBackend.Heartbeat(r.Context(), playerID, ticketID, key, revision, now) + } else { + ticket, err = s.Queue.Heartbeat(playerID, ticketID, key, revision, now) + } } else { - ticket, err = s.Queue.Cancel(playerID, ticketID, key, revision, now) + if s.QueueBackend != nil { + ticket, err = s.QueueBackend.Cancel(r.Context(), playerID, ticketID, key, revision, now) + } else { + ticket, err = s.Queue.Cancel(playerID, ticketID, key, revision, now) + } } if err != nil { writeDomainError(w, err) diff --git a/server/api/service_test.go b/server/api/service_test.go index 68afce88..10afa3d6 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -1,6 +1,7 @@ package api import ( + "context" "encoding/json" "net/http" "net/http/httptest" @@ -11,6 +12,22 @@ import ( "github.com/cosmic-clash/cosmic-clash/server/domain" ) +type queueBackendSpy struct{ createCalls int } + +func (b *queueBackendSpy) Create(_ context.Context, playerID, ticketID, _ string, spec domain.QueueSpec, now time.Time) (domain.QueueTicket, error) { + b.createCalls++ + return domain.QueueTicket{TicketID: ticketID, PlayerID: playerID, Playlist: spec.Playlist, State: domain.Queued, EnqueuedAt: now, ExpiresAt: now.Add(domain.QueueExpiryWindow)}, nil +} +func (*queueBackendSpy) Heartbeat(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error) { + return domain.QueueTicket{}, nil +} +func (*queueBackendSpy) Cancel(context.Context, string, string, string, uint64, time.Time) (domain.QueueTicket, error) { + return domain.QueueTicket{}, nil +} +func (*queueBackendSpy) Get(context.Context, string, string, time.Time) (domain.QueueTicket, error) { + return domain.QueueTicket{}, nil +} + func TestAuthenticatedQueueAPIUsesServerCandidateAndRevisionedMutations(t *testing.T) { now := time.Unix(1000, 0).UTC() sessions := domain.NewSessionStore() @@ -184,6 +201,27 @@ func TestQueueCreateRejectsCandidateMetadataMismatch(t *testing.T) { } } +func TestQueueAPIUsesInjectedPersistentBackendWithoutCandidateProvider(t *testing.T) { + now := time.Unix(1000, 0).UTC() + sessions := domain.NewSessionStore() + session, token, _ := sessions.Issue("player-1", time.Hour, now) + backend := &queueBackendSpy{} + service := &Service{Sessions: sessions, QueueBackend: backend, Now: func() time.Time { return now }} + server := httptest.NewServer(service.Handler()) + defer server.Close() + req, _ := http.NewRequest(http.MethodPost, server.URL+"/v1/queue", strings.NewReader(`{"ticket_id":"ticket-1","playlist":"ranked","client_build":"build-1","protocol_version":1}`)) + req.Header.Set("Authorization", "Bearer "+session.SessionID+":"+token) + req.Header.Set("Idempotency-Key", "create-key-123456") + response, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusCreated || backend.createCalls != 1 { + t.Fatalf("status=%d backend_calls=%d", response.StatusCode, backend.createCalls) + } +} + func TestQueueRecoveryAPIIsAuthenticatedOwnerOnlyAndExpiresStaleTickets(t *testing.T) { now := time.Unix(1000, 0).UTC() sessions := domain.NewSessionStore() diff --git a/server/store/queue_sql.go b/server/store/queue_sql.go index f9391364..9a59ca8b 100644 --- a/server/store/queue_sql.go +++ b/server/store/queue_sql.go @@ -91,6 +91,21 @@ type queueTicketRecord struct { Revision uint64 `json:"revision"` } +type PostgresQueue struct{ DB *sql.DB } + +func (q PostgresQueue) Create(ctx context.Context, playerID, ticketID, idempotencyKey string, spec domain.QueueSpec, now time.Time) (domain.QueueTicket, error) { + return CreateQueueTicket(ctx, q.DB, ticketID, playerID, idempotencyKey, spec, now) +} +func (q PostgresQueue) Heartbeat(ctx context.Context, playerID, ticketID, idempotencyKey string, revision uint64, now time.Time) (domain.QueueTicket, error) { + return HeartbeatQueueTicket(ctx, q.DB, playerID, ticketID, idempotencyKey, revision, now) +} +func (q PostgresQueue) Cancel(ctx context.Context, playerID, ticketID, idempotencyKey string, revision uint64, now time.Time) (domain.QueueTicket, error) { + return CancelQueueTicket(ctx, q.DB, playerID, ticketID, idempotencyKey, revision, now) +} +func (q PostgresQueue) Get(ctx context.Context, playerID, ticketID string, now time.Time) (domain.QueueTicket, error) { + return GetQueueTicket(ctx, q.DB, playerID, ticketID, now) +} + func GetQueueTicket(ctx context.Context, db *sql.DB, playerID, ticketID string, now time.Time) (domain.QueueTicket, error) { if db == nil || playerID == "" || ticketID == "" || now.IsZero() { return domain.QueueTicket{}, fmt.Errorf("invalid queue recovery arguments")