feat: add runnable casual matcher role

server/cmd/matcher/main.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+	"github.com/cosmic-clash/cosmic-clash/server/matcher"
+	"github.com/cosmic-clash/cosmic-clash/server/migrations"
+	"github.com/cosmic-clash/cosmic-clash/server/store"
+	_ "github.com/jackc/pgx/v5/stdlib"
+)
+
+func main() {
+	dsn := flag.String("dsn", os.Getenv("COSMIC_CLASH_POSTGRES_DSN"), "PostgreSQL connection string")
+	migrationDir := flag.String("migrations", "migrations", "directory containing numbered SQL migrations")
+	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")
+	flag.Parse()
+	if *dsn == "" {
+		fatalf("--dsn or COSMIC_CLASH_POSTGRES_DSN is required")
+	}
+	if *playlist != string(domain.Casual) {
+		fatalf("unsupported playlist %q; only casual is currently enabled", *playlist)
+	}
+	db, err := sql.Open("pgx", *dsn)
+	if err != nil {
+		fatalf("open PostgreSQL: %v", err)
+	}
+	defer db.Close()
+	startupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	if err := db.PingContext(startupCtx); err != nil {
+		fatalf("ping PostgreSQL: %v", err)
+	}
+	if err := migrations.Apply(startupCtx, db, *migrationDir); err != nil {
+		fatalf("apply migrations: %v", err)
+	}
+	now := func() time.Time { return time.Now().UTC() }
+	worker := matcher.Worker{
+		Source: func(ctx context.Context, at time.Time, limit int) ([]domain.Candidate, error) {
+			return store.ListQueuedCandidates(ctx, db, at, limit)
+		},
+		Creator: matcher.ProposalCreatorFunc(func(ctx context.Context, proposal domain.Proposal, ticketIDs map[string]string, at time.Time) error {
+			return store.CreateProposal(ctx, db, proposal, ticketIDs, at)
+		}),
+		Playlist: domain.Casual, Size: *size, Now: now,
+		NextID: func() string { return fmt.Sprintf("proposal-%d", time.Now().UnixNano()) },
+		Prepare: func(id string, playlist domain.Playlist, formation domain.MatchFormation, at time.Time) (domain.PreparedProposal, error) {
+			return domain.PrepareProposal(id, playlist, formation, nil, domain.RankedArena{}, at)
+		},
+	}
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	if err := worker.Run(ctx, *interval); err != nil && ctx.Err() == nil {
+		fatalf("matcher stopped: %v", err)
+	}
+}
+
+func fatalf(format string, args ...any) {
+	log.Printf("matcher: "+format, args...)
+	os.Exit(1)
+}

server/matcher/worker.go

@@ -34,6 +34,26 @@ type Worker struct {
 	Prepare  PrepareFunc
 }
 
+// Run polls until cancellation. A failed attempt is returned so a supervisor
+// can restart the role rather than silently dropping durable claim failures.
+func (w Worker) Run(ctx context.Context, interval time.Duration) error {
+	if interval <= 0 {
+		return fmt.Errorf("matcher interval must be positive")
+	}
+	for {
+		if _, err := w.RunOnce(ctx); err != nil {
+			return err
+		}
+		timer := time.NewTimer(interval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return nil
+		case <-timer.C:
+		}
+	}
+}
+
 // RunOnce performs one bounded matchmaking attempt. The source may be Redis
 // backed, but the creator must be the durable transaction that claims tickets;
 // a stale cache therefore fails safely and can be retried on the next pass.

server/store/queue_sql.go

@@ -37,6 +37,41 @@ WHERE ticket_id = $1 AND player_id = $2 AND revision = $3
 RETURNING ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at, revision`
 )
 
+const QueueCandidateProjectionSQL = `SELECT ticket_id, player_id, playlist, client_build,
+    protocol_version, enqueued_at
+FROM queue_tickets
+WHERE state = 'QUEUED' AND expires_at > $1
+ORDER BY enqueued_at, ticket_id
+LIMIT $2`
+
+// ListQueuedCandidates is an authoritative, expiry-filtered source for the
+// matcher projection. It deliberately does not claim rows; CreateProposal is
+// the transaction that performs the competing claim with SKIP LOCKED fences.
+func ListQueuedCandidates(ctx context.Context, db *sql.DB, now time.Time, limit int) ([]domain.Candidate, error) {
+	if db == nil || now.IsZero() || limit < 1 || limit > 1000 {
+		return nil, fmt.Errorf("invalid queued candidate arguments")
+	}
+	rows, err := db.QueryContext(ctx, QueueCandidateProjectionSQL, now, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var candidates []domain.Candidate
+	for rows.Next() {
+		var candidate domain.Candidate
+		var playlist string
+		if err := rows.Scan(&candidate.TicketID, &candidate.PlayerID, &playlist, &candidate.ClientBuild, &candidate.ProtocolVersion, &candidate.EnqueuedAt); err != nil {
+			return nil, err
+		}
+		candidate.Playlist = domain.Playlist(playlist)
+		candidates = append(candidates, candidate)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return candidates, nil
+}
+
 func CreateQueueTicket(ctx context.Context, db *sql.DB, ticketID, playerID, idempotencyKey string, spec domain.QueueSpec, now time.Time) (domain.QueueTicket, error) {
 	if db == nil || ticketID == "" || playerID == "" || len(idempotencyKey) < 16 || len(idempotencyKey) > 128 || (spec.Playlist != domain.Casual && spec.Playlist != domain.Ranked) || spec.ClientBuild == "" || len(spec.ClientBuild) > 128 || spec.ProtocolVersion < 1 || now.IsZero() {
 		return domain.QueueTicket{}, fmt.Errorf("invalid queue transaction arguments")