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") } }