CosmicClash/server/store/redis_candidates_test.go

package store

import (
	"context"
	"fmt"
	"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{
		{Playlist: domain.Casual, TicketID: "ticket-b", PlayerID: "player-b", EnqueuedAt: now.Add(time.Second)},
		{Playlist: domain.Casual, TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: now},
	}
	if err := index.Rebuild(context.Background(), domain.Casual, candidates); err != nil {
		t.Fatal(err)
	}
	got, err := index.Snapshot(context.Background(), domain.Casual, now, 1000)
	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(), domain.Casual, "ticket-a"); err != nil {
		t.Fatal(err)
	}
	got, err = index.Snapshot(context.Background(), domain.Casual, now.Add(2*time.Second), 1000)
	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:casual: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(), domain.Casual, 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{Playlist: domain.Casual, TicketID: "ticket-a", PlayerID: "player-a", EnqueuedAt: time.Unix(1000, 0)}
	if err := index.Rebuild(context.Background(), domain.Casual, []domain.Candidate{candidate, candidate}); err == nil {
		t.Fatal("duplicate candidate accepted")
	}
	if err := index.Upsert(context.Background(), domain.Candidate{Playlist: domain.Casual, TicketID: "", PlayerID: "player-a", EnqueuedAt: candidate.EnqueuedAt}); err == nil {
		t.Fatal("invalid candidate accepted")
	}
}

// Both playlists used to share one hash and sorted set. Two failures followed:
// the matcher truncated a mixed snapshot to its candidate limit before
// filtering by playlist, so a large casual prefix could leave the ranked
// worker with zero candidates indefinitely; and Rebuild replaced the shared
// keys, so a casual repair erased ranked projections and vice versa.
func TestRedisCandidateIndexIsolatesPlaylists(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: "isolation", TTL: time.Minute}
	ctx := context.Background()
	now := time.Unix(1000, 0).UTC()

	// A large casual backlog enqueued strictly before the ranked tickets. With
	// shared keys this prefix is exactly what starved the ranked worker.
	casual := make([]domain.Candidate, 0, 300)
	for i := 0; i < 300; i++ {
		casual = append(casual, domain.Candidate{
			Playlist: domain.Casual, TicketID: fmt.Sprintf("casual-%03d", i),
			PlayerID: fmt.Sprintf("casual-player-%03d", i), EnqueuedAt: now.Add(time.Duration(i) * time.Millisecond),
		})
	}
	ranked := []domain.Candidate{
		{Playlist: domain.Ranked, TicketID: "ranked-a", PlayerID: "ranked-player-a", EnqueuedAt: now.Add(time.Second)},
		{Playlist: domain.Ranked, TicketID: "ranked-b", PlayerID: "ranked-player-b", EnqueuedAt: now.Add(2 * time.Second)},
	}
	if err := index.Rebuild(ctx, domain.Casual, casual); err != nil {
		t.Fatalf("casual rebuild: %v", err)
	}
	if err := index.Rebuild(ctx, domain.Ranked, ranked); err != nil {
		t.Fatalf("ranked rebuild: %v", err)
	}

	// The casual rebuild must not have erased the ranked projection.
	at := now.Add(time.Hour)
	gotRanked, err := index.Snapshot(ctx, domain.Ranked, at, 50)
	if err != nil {
		t.Fatalf("ranked snapshot: %v", err)
	}
	if len(gotRanked) != 2 {
		t.Fatalf("ranked worker saw %d candidates behind a 300-deep casual backlog, want 2", len(gotRanked))
	}
	for _, candidate := range gotRanked {
		if candidate.Playlist != domain.Ranked {
			t.Fatalf("ranked snapshot leaked a %q candidate: %s", candidate.Playlist, candidate.TicketID)
		}
	}

	// A ranked repair must likewise leave casual alone.
	if err := index.Rebuild(ctx, domain.Ranked, ranked[:1]); err != nil {
		t.Fatalf("ranked re-repair: %v", err)
	}
	gotCasual, err := index.Snapshot(ctx, domain.Casual, at, 1000)
	if err != nil {
		t.Fatalf("casual snapshot: %v", err)
	}
	if len(gotCasual) != 300 {
		t.Fatalf("ranked rebuild erased casual projection: %d remain", len(gotCasual))
	}
}

// The limit must be applied by Redis, not after transfer: the old unbounded
// ZRANGEBYSCORE plus HMGET decoded the entire queue on every one-second poll.
func TestRedisCandidateIndexSnapshotIsBoundedByRedis(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: "bounded", TTL: time.Minute}
	ctx := context.Background()
	now := time.Unix(1000, 0).UTC()

	candidates := make([]domain.Candidate, 0, 500)
	for i := 0; i < 500; i++ {
		candidates = append(candidates, domain.Candidate{
			Playlist: domain.Casual, TicketID: fmt.Sprintf("bulk-%03d", i),
			PlayerID: fmt.Sprintf("bulk-player-%03d", i), EnqueuedAt: now.Add(time.Duration(i) * time.Millisecond),
		})
	}
	if err := index.Rebuild(ctx, domain.Casual, candidates); err != nil {
		t.Fatal(err)
	}
	got, err := index.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 10)
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 10 {
		t.Fatalf("snapshot returned %d candidates for a limit of 10", len(got))
	}
	// Oldest-first ordering must survive the bound.
	if got[0].TicketID != "bulk-000" || got[9].TicketID != "bulk-009" {
		t.Fatalf("bounded snapshot lost enqueue ordering: %s..%s", got[0].TicketID, got[9].TicketID)
	}
	if _, err := index.Snapshot(ctx, domain.Casual, now.Add(time.Hour), 0); err == nil {
		t.Fatal("unbounded snapshot accepted")
	}
}