feat: add rebuildable Redis candidate index

2026-09-11 08:23:45 +00:00 · 2026-09-01 09:01:32 +01:00
parent 414df530f0
commit eadaa7b54e
6 changed files with 259 additions and 3 deletions
@@ -0,0 +1,161 @@
+package store
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+	"github.com/redis/go-redis/v9"
+)
+
+// RedisCandidateIndex is a rebuildable acceleration index. It never decides
+// ownership or claims a match; callers must source candidates from the
+// durable queue projection before rebuilding it.
+type RedisCandidateIndex struct {
+	Client *redis.Client
+	Prefix string
+	TTL    time.Duration
+}
+
+func (r RedisCandidateIndex) keys() (string, string) {
+	prefix := r.Prefix
+	if prefix == "" {
+		prefix = "cosmic-clash"
+	}
+	return prefix + ":queue:candidates:data", prefix + ":queue:candidates:order"
+}
+
+func (r RedisCandidateIndex) validate() error {
+	if r.Client == nil || r.TTL <= 0 {
+		return fmt.Errorf("invalid Redis candidate index")
+	}
+	return nil
+}
+
+func validateRedisCandidate(candidate domain.Candidate) error {
+	if candidate.TicketID == "" || candidate.PlayerID == "" || candidate.EnqueuedAt.IsZero() {
+		return fmt.Errorf("invalid candidate")
+	}
+	return nil
+}
+
+// Upsert stores the candidate payload and its deterministic enqueue ordering.
+// Both keys receive a TTL so a Redis restart or abandoned index cannot become
+// a permanent source of stale presence.
+func (r RedisCandidateIndex) Upsert(ctx context.Context, candidate domain.Candidate) error {
+	if err := r.validate(); err != nil {
+		return err
+	}
+	if err := validateRedisCandidate(candidate); err != nil {
+		return err
+	}
+	payload, err := json.Marshal(candidate)
+	if err != nil {
+		return err
+	}
+	dataKey, orderKey := r.keys()
+	pipe := r.Client.TxPipeline()
+	pipe.HSet(ctx, dataKey, candidate.TicketID, payload)
+	pipe.ZAdd(ctx, orderKey, redis.Z{Score: float64(candidate.EnqueuedAt.UnixNano()), Member: candidate.TicketID})
+	pipe.Expire(ctx, dataKey, r.TTL)
+	pipe.Expire(ctx, orderKey, r.TTL)
+	_, err = pipe.Exec(ctx)
+	return err
+}
+
+func (r RedisCandidateIndex) Remove(ctx context.Context, ticketID string) error {
+	if err := r.validate(); err != nil {
+		return err
+	}
+	if ticketID == "" {
+		return fmt.Errorf("ticket ID is required")
+	}
+	dataKey, orderKey := r.keys()
+	pipe := r.Client.TxPipeline()
+	pipe.HDel(ctx, dataKey, ticketID)
+	pipe.ZRem(ctx, orderKey, ticketID)
+	_, err := pipe.Exec(ctx)
+	return err
+}
+
+// Snapshot reads only candidates whose enqueue timestamp is not in the
+// future. Missing payloads are ignored; the durable rebuild path repairs such
+// partial cache state without allowing it to affect ownership.
+func (r RedisCandidateIndex) Snapshot(ctx context.Context, now time.Time) ([]domain.Candidate, error) {
+	if err := r.validate(); err != nil {
+		return nil, err
+	}
+	if now.IsZero() {
+		return nil, fmt.Errorf("authoritative time is required")
+	}
+	dataKey, orderKey := r.keys()
+	tickets, err := r.Client.ZRangeByScore(ctx, orderKey, &redis.ZRangeBy{
+		Min: "-inf", Max: fmt.Sprint(now.UnixNano()),
+	}).Result()
+	if err != nil {
+		return nil, err
+	}
+	if len(tickets) == 0 {
+		return []domain.Candidate{}, nil
+	}
+	payloads, err := r.Client.HMGet(ctx, dataKey, tickets...).Result()
+	if err != nil {
+		return nil, err
+	}
+	result := make([]domain.Candidate, 0, len(payloads))
+	for _, raw := range payloads {
+		text, ok := raw.(string)
+		if !ok {
+			continue
+		}
+		var candidate domain.Candidate
+		if err := json.Unmarshal([]byte(text), &candidate); err != nil {
+			continue
+		}
+		if err := validateRedisCandidate(candidate); err != nil || candidate.EnqueuedAt.After(now) {
+			continue
+		}
+		result = append(result, candidate)
+	}
+	return result, nil
+}
+
+// Rebuild atomically replaces both Redis keys from the authoritative queue
+// projection. It is the required path after Redis restart/failover or cache
+// loss, and rejects duplicate ticket IDs before touching Redis.
+func (r RedisCandidateIndex) Rebuild(ctx context.Context, candidates []domain.Candidate) error {
+	if err := r.validate(); err != nil {
+		return err
+	}
+	seen := make(map[string]struct{}, len(candidates))
+	values := make([]interface{}, 0, len(candidates)*2)
+	scores := make([]redis.Z, 0, len(candidates))
+	for _, candidate := range candidates {
+		if err := validateRedisCandidate(candidate); err != nil {
+			return err
+		}
+		if _, exists := seen[candidate.TicketID]; exists {
+			return fmt.Errorf("duplicate candidate in rebuild")
+		}
+		seen[candidate.TicketID] = struct{}{}
+		payload, err := json.Marshal(candidate)
+		if err != nil {
+			return err
+		}
+		values = append(values, candidate.TicketID, payload)
+		scores = append(scores, redis.Z{Score: float64(candidate.EnqueuedAt.UnixNano()), Member: candidate.TicketID})
+	}
+	dataKey, orderKey := r.keys()
+	pipe := r.Client.TxPipeline()
+	pipe.Del(ctx, dataKey, orderKey)
+	if len(values) > 0 {
+		pipe.HSet(ctx, dataKey, values...)
+		pipe.ZAdd(ctx, orderKey, scores...)
+	}
+	pipe.Expire(ctx, dataKey, r.TTL)
+	pipe.Expire(ctx, orderKey, r.TTL)
+	_, err := pipe.Exec(ctx)
+	return err
+}
@@ -0,0 +1,72 @@
+package store
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+	"github.com/redis/go-redis/v9"
+)
+
+func TestRedisCandidateIndexRebuildSnapshotAndRemove(t *testing.T) {
+	mini, err := miniredis.Run()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer mini.Close()
+	client := redis.NewClient(&redis.Options{Addr: mini.Addr()})
+	defer client.Close()
+	index := RedisCandidateIndex{Client: client, Prefix: "integration", TTL: time.Minute}
+	now := time.Unix(1000, 0).UTC()
+	candidates := []domain.Candidate{
+		{TicketID: "ticket-b", PlayerID: "player-b", EnqueuedAt: now.Add(time.Second)},
+		{TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: now},
+	}
+	if err := index.Rebuild(context.Background(), candidates); err != nil {
+		t.Fatal(err)
+	}
+	got, err := index.Snapshot(context.Background(), now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 1 || got[0].TicketID != "ticket-a" {
+		t.Fatalf("snapshot before future candidate = %+v", got)
+	}
+	if err := index.Remove(context.Background(), "ticket-a"); err != nil {
+		t.Fatal(err)
+	}
+	got, err = index.Snapshot(context.Background(), now.Add(2*time.Second))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 1 || got[0].TicketID != "ticket-b" {
+		t.Fatalf("snapshot after remove = %+v", got)
+	}
+	if ttl, err := client.TTL(context.Background(), "integration:queue:candidates:data").Result(); err != nil || ttl <= 0 {
+		t.Fatalf("candidate data TTL = %v, err = %v", ttl, err)
+	}
+}
+
+func TestRedisCandidateIndexRejectsInvalidAndDuplicateRebuilds(t *testing.T) {
+	index := RedisCandidateIndex{TTL: time.Minute}
+	if err := index.Rebuild(context.Background(), nil); err == nil {
+		t.Fatal("nil Redis client accepted")
+	}
+	mini, err := miniredis.Run()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer mini.Close()
+	client := redis.NewClient(&redis.Options{Addr: mini.Addr()})
+	defer client.Close()
+	index.Client = client
+	candidate := domain.Candidate{TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: time.Unix(1000, 0)}
+	if err := index.Rebuild(context.Background(), []domain.Candidate{candidate, candidate}); err == nil {
+		t.Fatal("duplicate candidate accepted")
+	}
+	if err := index.Upsert(context.Background(), domain.Candidate{TicketID: "", PlayerID: "player-a", EnqueuedAt: candidate.EnqueuedAt}); err == nil {
+		t.Fatal("invalid candidate accepted")
+	}
+}