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() type candidate struct { matchID string playlist string readyAt time.Time } var candidates []candidate for rows.Next() { var matchID, playlist string var readyAt time.Time if err := rows.Scan(&matchID, &playlist, &readyAt); err != nil { return 0, err } candidates = append(candidates, candidate{matchID: matchID, playlist: playlist, readyAt: readyAt}) } if err := rows.Err(); err != nil { return 0, err } if err := rows.Close(); err != nil { return 0, err } count := 0 for _, candidate := range candidates { participants, err := loadInitialConnectSnapshot(ctx, db, candidate.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(candidate.playlist), candidate.readyAt, now, participants, history) if err != nil { return count, fmt.Errorf("plan initial connect %s: %w", candidate.matchID, err) } if plan.Action == domain.InitialConnectWait { continue } if err := ApplyInitialConnectPlan(ctx, db, candidate.matchID, "initial-connect:"+candidate.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 }