feat(multiplayer): sweep initial connect outcomes

server/cmd/maintenance/main.go

@@ -23,6 +23,7 @@ func main() {
 	batch := flag.Int("batch", 100, "maximum player rollovers per pass")
 	stalledAllocationDeadline := flag.Duration("stalled-allocation-deadline", 2*time.Minute, "reclaim a match stuck in ALLOCATING/PROCESS_READY/ASSIGNMENT_READY (server crashed or was reclaimed before registering) after this long, requeuing every participant without penalty")
 	stalledAllocationBatch := flag.Int("stalled-allocation-batch", 100, "maximum stalled matches reclaimed per pass")
+	initialConnectBatch := flag.Int("initial-connect-batch", 100, "maximum pre-live matches evaluated per pass")
 	flag.Parse()
 	if *dsn == "" {
 		fatalf("--dsn or COSMIC_CLASH_POSTGRES_DSN is required")
@@ -33,6 +34,9 @@ func main() {
 	if *stalledAllocationDeadline <= 0 || *stalledAllocationBatch < 1 || *stalledAllocationBatch > 1000 {
 		fatalf("invalid stalled-allocation deadline or batch")
 	}
+	if *initialConnectBatch < 1 || *initialConnectBatch > 1000 {
+		fatalf("invalid initial-connect batch")
+	}
 	db, err := sql.Open("pgx", *dsn)
 	if err != nil {
 		fatalf("open PostgreSQL: %v", err)
@@ -64,6 +68,13 @@ func main() {
 		if reclaimed > 0 {
 			log.Printf("reclaimed %d stalled allocations, requeuing their participants", reclaimed)
 		}
+		reconciled, err := store.ReconcileInitialConnect(ctx, db, now, *initialConnectBatch)
+		if err != nil {
+			fatalf("initial-connect maintenance: %v", err)
+		}
+		if reconciled > 0 {
+			log.Printf("reconciled %d initial-connect outcomes", reconciled)
+		}
 		timer := time.NewTimer(*interval)
 		select {
 		case <-ctx.Done():

server/store/initial_connect_maintenance.go

@@ -0,0 +1,108 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/domain"
+)
+
+const initialConnectCandidatesSQL = `SELECT match_id, playlist, created_at
+FROM matches
+WHERE state IN ('ASSIGNMENT_READY', 'ASSIGNED', 'CONNECTING')
+ORDER BY created_at, match_id
+LIMIT $1`
+
+const initialConnectHistorySQL = `SELECT starts_at
+FROM penalties
+WHERE player_id = $1 AND kind = 'INITIAL_CONNECT_NO_SHOW'
+ORDER BY starts_at`
+
+// ReconcileInitialConnect evaluates a bounded set of matches and applies only
+// terminal or bot-start decisions. WAIT is intentionally non-mutating. A
+// concurrent allocator/server transition is harmless: ApplyInitialConnectPlan
+// locks and revalidates the match before changing anything.
+func ReconcileInitialConnect(ctx context.Context, db *sql.DB, now time.Time, limit int) (int, error) {
+	if db == nil || now.IsZero() || limit < 1 || limit > 1000 {
+		return 0, fmt.Errorf("invalid initial-connect maintenance arguments")
+	}
+	rows, err := db.QueryContext(ctx, initialConnectCandidatesSQL, limit)
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	count := 0
+	for rows.Next() {
+		var matchID, playlist string
+		var readyAt time.Time
+		if err := rows.Scan(&matchID, &playlist, &readyAt); err != nil {
+			return count, err
+		}
+		participants, err := loadInitialConnectSnapshot(ctx, db, matchID)
+		if err != nil {
+			return count, err
+		}
+		history, err := loadInitialConnectHistory(ctx, db, participants)
+		if err != nil {
+			return count, err
+		}
+		plan, err := domain.PlanInitialConnect(domain.Playlist(playlist), readyAt, now, participants, history)
+		if err != nil {
+			return count, fmt.Errorf("plan initial connect %s: %w", matchID, err)
+		}
+		if plan.Action == domain.InitialConnectWait {
+			continue
+		}
+		if err := ApplyInitialConnectPlan(ctx, db, matchID, "initial-connect:"+matchID, plan, now); err != nil {
+			return count, err
+		}
+		count++
+	}
+	return count, rows.Err()
+}
+
+func loadInitialConnectSnapshot(ctx context.Context, db *sql.DB, matchID string) ([]domain.ConnectParticipant, error) {
+	rows, err := db.QueryContext(ctx, `SELECT player_id, team, connected_at
+FROM match_participants WHERE match_id = $1 AND participation_active ORDER BY player_id`, matchID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var participants []domain.ConnectParticipant
+	for rows.Next() {
+		var playerID string
+		var team int
+		var connectedAt sql.NullTime
+		if err := rows.Scan(&playerID, &team, &connectedAt); err != nil {
+			return nil, err
+		}
+		participants = append(participants, domain.ConnectParticipant{PlayerID: playerID, Team: team, Connected: connectedAt.Valid})
+	}
+	return participants, rows.Err()
+}
+
+func loadInitialConnectHistory(ctx context.Context, db *sql.DB, participants []domain.ConnectParticipant) (map[string][]time.Time, error) {
+	history := make(map[string][]time.Time)
+	for _, participant := range participants {
+		rows, err := db.QueryContext(ctx, initialConnectHistorySQL, participant.PlayerID)
+		if err != nil {
+			return nil, err
+		}
+		for rows.Next() {
+			var started time.Time
+			if err := rows.Scan(&started); err != nil {
+				rows.Close()
+				return nil, err
+			}
+			history[participant.PlayerID] = append(history[participant.PlayerID], started)
+		}
+		if err := rows.Err(); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		rows.Close()
+	}
+	return history, nil
+}

server/store/initial_connect_sql_test.go

@@ -9,6 +9,9 @@ import (
 )
 
 func TestInitialConnectSQLPreservesAtomicNoShowReconciliation(t *testing.T) {
+	if !contains(initialConnectCandidatesSQL, "ASSIGNMENT_READY") || !contains(initialConnectCandidatesSQL, "LIMIT $1") {
+		t.Fatal("initial-connect sweep is not bounded to pre-live matches")
+	}
 	for query, fragments := range map[string][]string{
 		initialConnectIdempotencyInsertSQL: {"ON CONFLICT", "payload_digest"},
 		initialConnectMatchLockSQL:         {"FOR UPDATE", "match_id = $1"},