feat: add rebuildable candidate cache

server/store/candidates.go

@@ -0,0 +1,75 @@
+package store
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+// CandidateCache is intentionally rebuildable. A real Redis implementation
+// can satisfy this interface, but no cache operation is an ownership fence.
+type CandidateCache struct {
+	mu         sync.RWMutex
+	candidates map[string]domain.Candidate
+}
+
+func NewCandidateCache() *CandidateCache {
+	return &CandidateCache{candidates: make(map[string]domain.Candidate)}
+}
+
+func (c *CandidateCache) Upsert(candidate domain.Candidate) error {
+	if candidate.TicketID == "" || candidate.PlayerID == "" || candidate.EnqueuedAt.IsZero() {
+		return fmt.Errorf("invalid candidate")
+	}
+	c.mu.Lock()
+	c.candidates[candidate.TicketID] = candidate
+	c.mu.Unlock()
+	return nil
+}
+
+func (c *CandidateCache) Remove(ticketID string) {
+	c.mu.Lock()
+	delete(c.candidates, ticketID)
+	c.mu.Unlock()
+}
+
+func (c *CandidateCache) Snapshot(now time.Time) []domain.Candidate {
+	c.mu.RLock()
+	result := make([]domain.Candidate, 0, len(c.candidates))
+	for _, candidate := range c.candidates {
+		if !candidate.EnqueuedAt.After(now) {
+			result = append(result, candidate)
+		}
+	}
+	c.mu.RUnlock()
+	sort.Slice(result, func(i, j int) bool {
+		if !result[i].EnqueuedAt.Equal(result[j].EnqueuedAt) {
+			return result[i].EnqueuedAt.Before(result[j].EnqueuedAt)
+		}
+		return result[i].TicketID < result[j].TicketID
+	})
+	return result
+}
+
+// Rebuild replaces the cache atomically with the authoritative queue view.
+// Callers should invoke this after Redis restart, failover, or a cache miss;
+// the supplied candidates must already have passed durable queue checks.
+func (c *CandidateCache) Rebuild(candidates []domain.Candidate) error {
+	rebuilt := make(map[string]domain.Candidate, len(candidates))
+	for _, candidate := range candidates {
+		if candidate.TicketID == "" || candidate.PlayerID == "" || candidate.EnqueuedAt.IsZero() {
+			return fmt.Errorf("invalid candidate in rebuild")
+		}
+		if _, exists := rebuilt[candidate.TicketID]; exists {
+			return fmt.Errorf("duplicate candidate in rebuild")
+		}
+		rebuilt[candidate.TicketID] = candidate
+	}
+	c.mu.Lock()
+	c.candidates = rebuilt
+	c.mu.Unlock()
+	return nil
+}

server/store/candidates_test.go

@@ -0,0 +1,39 @@
+package store
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+func TestCandidateCacheRebuildRepairsLossAndKeepsDeterministicOrder(t *testing.T) {
+	now := time.Unix(1000, 0)
+	cache := NewCandidateCache()
+	candidates := []domain.Candidate{{TicketID: "ticket-b", PlayerID: "player-b", EnqueuedAt: now}, {TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: now}}
+	if err := cache.Rebuild(candidates); err != nil {
+		t.Fatal(err)
+	}
+	cache.Remove("ticket-a")
+	if got := cache.Snapshot(now); len(got) != 1 || got[0].TicketID != "ticket-b" {
+		t.Fatalf("stale cache snapshot = %+v", got)
+	}
+	if err := cache.Rebuild(candidates); err != nil {
+		t.Fatal(err)
+	}
+	got := cache.Snapshot(now)
+	if len(got) != 2 || got[0].TicketID != "ticket-a" || got[1].TicketID != "ticket-b" {
+		t.Fatalf("repaired order = %+v", got)
+	}
+}
+
+func TestCandidateCacheRejectsInvalidOrDuplicateDurableProjection(t *testing.T) {
+	cache := NewCandidateCache()
+	if err := cache.Upsert(domain.Candidate{TicketID: "", PlayerID: "p", EnqueuedAt: time.Unix(1000, 0)}); err == nil {
+		t.Fatal("invalid candidate accepted")
+	}
+	candidate := domain.Candidate{TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: time.Unix(1000, 0)}
+	if err := cache.Rebuild([]domain.Candidate{candidate, candidate}); err == nil {
+		t.Fatal("duplicate candidate accepted")
+	}
+}