feat: wire API queue projection to Redis

server/api/service.go

@@ -34,6 +34,13 @@ type QueueBackend interface {
 	Get(context.Context, string, string, time.Time) (domain.QueueTicket, error)
 }
 
+// CandidateIndex is a transient projection of durable queue ownership. Index
+// failures must never change the result of an already successful mutation.
+type CandidateIndex interface {
+	Upsert(context.Context, domain.Candidate) error
+	Remove(context.Context, string) error
+}
+
 type SessionBackend interface {
 	Authenticate(context.Context, string, string, time.Time) (domain.Session, error)
 }
@@ -77,6 +84,7 @@ type Service struct {
 	Candidate       CandidateProvider
 	CandidateV2     CandidateProviderV2
 	QueueBackend    QueueBackend
+	CandidateIndex  CandidateIndex
 	Probe           ProbeProvider
 	Assignment      AssignmentProvider
 	Now             func() time.Time
@@ -220,6 +228,7 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) {
 			writeDomainError(w, err)
 			return
 		}
+		s.projectCandidate(r.Context(), ticket)
 		s.publishTicketEvent(ticket, now)
 		writeJSON(w, http.StatusCreated, toQueueResponse(ticket))
 		return
@@ -249,10 +258,23 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) {
 		writeDomainError(w, err)
 		return
 	}
+	s.projectCandidate(r.Context(), ticket)
 	s.publishTicketEvent(ticket, now)
 	writeJSON(w, http.StatusCreated, toQueueResponse(ticket))
 }
 
+func (s *Service) projectCandidate(ctx context.Context, ticket domain.QueueTicket) {
+	if s.CandidateIndex != nil {
+		_ = s.CandidateIndex.Upsert(ctx, ticket.Candidate)
+	}
+}
+
+func (s *Service) removeCandidate(ctx context.Context, ticketID string) {
+	if s.CandidateIndex != nil {
+		_ = s.CandidateIndex.Remove(ctx, ticketID)
+	}
+}
+
 func (s *Service) contractQueueCreate(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		s.queueCreate(w, r)
@@ -391,6 +413,11 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) {
 		writeDomainError(w, err)
 		return
 	}
+	if ticket.State == domain.Cancelled {
+		s.removeCandidate(r.Context(), ticket.TicketID)
+	} else {
+		s.projectCandidate(r.Context(), ticket)
+	}
 	s.publishTicketEvent(ticket, now)
 	if r.Header.Get("X-Contract-Delete") == "1" {
 		w.WriteHeader(http.StatusNoContent)

server/api/service_test.go

@@ -19,6 +19,23 @@ import (
 
 type queueBackendSpy struct{ createCalls, heartbeatCalls, cancelCalls, getCalls int }
 
+type candidateIndexSpy struct {
+	upsertCalls, removeCalls int
+	upsertErr, removeErr     error
+	last                     domain.Candidate
+}
+
+func (i *candidateIndexSpy) Upsert(_ context.Context, candidate domain.Candidate) error {
+	i.upsertCalls++
+	i.last = candidate
+	return i.upsertErr
+}
+
+func (i *candidateIndexSpy) Remove(_ context.Context, _ string) error {
+	i.removeCalls++
+	return i.removeErr
+}
+
 type sessionBackendSpy struct{ calls int }
 
 type proposalBackendSpy struct {
@@ -612,6 +629,49 @@ func TestQueueAPIUsesInjectedPersistentBackendWithoutCandidateProvider(t *testin
 	}
 }
 
+func TestQueueAPIProjectsSuccessfulMutationsWithoutMakingRedisRequired(t *testing.T) {
+	now := time.Unix(1000, 0).UTC()
+	sessions := domain.NewSessionStore()
+	session, token, _ := sessions.Issue("player-1", time.Hour, now)
+	backend := &queueBackendSpy{}
+	index := &candidateIndexSpy{upsertErr: errors.New("redis unavailable"), removeErr: errors.New("redis unavailable")}
+	service := &Service{Sessions: sessions, QueueBackend: backend, CandidateIndex: index, Now: func() time.Time { return now }}
+	server := httptest.NewServer(service.Handler())
+	defer server.Close()
+	auth := "Bearer " + session.SessionID + ":" + token
+	request := func(method, path, key, revision string) *http.Response {
+		req, _ := http.NewRequest(method, server.URL+path, strings.NewReader(`{"ticket_id":"ticket-1","playlist":"ranked","client_build":"build-1","protocol_version":1}`))
+		req.Header.Set("Authorization", auth)
+		if key != "" {
+			req.Header.Set("Idempotency-Key", key)
+		}
+		if revision != "" {
+			req.Header.Set("If-Match-Revision", revision)
+		}
+		response, err := http.DefaultClient.Do(req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return response
+	}
+	response := request(http.MethodPost, "/v1/queue", "create-key-123456", "")
+	if response.StatusCode != http.StatusCreated {
+		t.Fatalf("create status=%d", response.StatusCode)
+	}
+	response.Body.Close()
+	if index.upsertCalls != 1 {
+		t.Fatalf("upsert calls=%d", index.upsertCalls)
+	}
+	response = request(http.MethodPost, "/v1/queue/ticket-1/cancel", "cancel-key-123456", "0")
+	if response.StatusCode != http.StatusOK {
+		t.Fatalf("cancel status=%d", response.StatusCode)
+	}
+	response.Body.Close()
+	if index.removeCalls != 1 {
+		t.Fatalf("remove calls=%d", index.removeCalls)
+	}
+}
+
 func TestQueueAPIDelegatesAllMutationsAndRecoveryToBackend(t *testing.T) {
 	now := time.Unix(1000, 0).UTC()
 	sessions := domain.NewSessionStore()