feat: repair matcher candidates through redis projection

This commit is contained in:
Josh Creek
2026-09-01 10:11:54 +01:00
parent 7a3d520608
commit d882469c79
5 changed files with 82 additions and 3 deletions
+43
View File
@@ -16,6 +16,7 @@ import (
"github.com/cosmic-clash/cosmic-clash/server/migrations"
"github.com/cosmic-clash/cosmic-clash/server/store"
_ "github.com/jackc/pgx/v5/stdlib"
"github.com/redis/go-redis/v9"
)
func main() {
@@ -24,6 +25,9 @@ func main() {
playlist := flag.String("playlist", string(domain.Casual), "playlist to match; ranked requires a provider-enabled role")
size := flag.Int("size", 4, "players per match")
interval := flag.Duration("interval", time.Second, "poll interval")
redisAddr := flag.String("redis-addr", os.Getenv("COSMIC_CLASH_REDIS_ADDR"), "optional Redis candidate projection address")
redisPrefix := flag.String("redis-prefix", envOrDefault("COSMIC_CLASH_REDIS_PREFIX", "cosmic-clash"), "Redis key prefix")
redisTTL := flag.Duration("redis-ttl", 60*time.Second, "TTL for transient candidate projection entries")
flag.Parse()
if *dsn == "" {
fatalf("--dsn or COSMIC_CLASH_POSTGRES_DSN is required")
@@ -31,6 +35,9 @@ func main() {
if *playlist != string(domain.Casual) {
fatalf("unsupported playlist %q; only casual is currently enabled", *playlist)
}
if *redisTTL <= 0 {
fatalf("--redis-ttl must be positive")
}
db, err := sql.Open("pgx", *dsn)
if err != nil {
fatalf("open PostgreSQL: %v", err)
@@ -45,8 +52,37 @@ func main() {
fatalf("apply migrations: %v", err)
}
now := func() time.Time { return time.Now().UTC() }
var redisClient *redis.Client
var projection *store.CandidateProjection
if *redisAddr != "" {
redisClient = redis.NewClient(&redis.Options{Addr: *redisAddr})
defer redisClient.Close()
candidateProjection := store.CandidateProjection{
Index: store.RedisCandidateIndex{Client: redisClient, Prefix: *redisPrefix, TTL: *redisTTL},
Source: func(ctx context.Context, at time.Time) ([]domain.Candidate, error) {
return store.ListQueuedCandidates(ctx, db, domain.Casual, at, 1000)
},
}
projection = &candidateProjection
}
worker := matcher.Worker{
Source: func(ctx context.Context, at time.Time, playlist domain.Playlist, limit int) ([]domain.Candidate, error) {
if projection != nil {
candidates, err := projection.Snapshot(ctx, at)
if err != nil {
return nil, err
}
if len(candidates) > limit {
candidates = candidates[:limit]
}
filtered := make([]domain.Candidate, 0, len(candidates))
for _, candidate := range candidates {
if candidate.Playlist == playlist {
filtered = append(filtered, candidate)
}
}
return filtered, nil
}
return store.ListQueuedCandidates(ctx, db, playlist, at, limit)
},
Creator: matcher.ProposalCreatorFunc(func(ctx context.Context, proposal domain.Proposal, ticketIDs map[string]string, at time.Time) error {
@@ -65,6 +101,13 @@ func main() {
}
}
func envOrDefault(name, fallback string) string {
if value := os.Getenv(name); value != "" {
return value
}
return fallback
}
func fatalf(format string, args ...any) {
log.Printf("matcher: "+format, args...)
os.Exit(1)