package store import ( "context" "database/sql" "fmt" "time" "github.com/cosmic-clash/cosmic-clash/server/domain" ) const liveAbandonmentCandidatesSQL = `SELECT m.match_id FROM matches m WHERE m.playlist = 'ranked' AND m.state = 'LIVE' AND EXISTS ( SELECT 1 FROM match_participants mp WHERE mp.match_id = m.match_id AND mp.participation_active AND mp.abandoned_at IS NULL AND mp.disconnected_at IS NOT NULL AND mp.disconnected_at < $1 ) ORDER BY m.match_id LIMIT $2` const liveAbandonmentMatchLockSQL = `SELECT playlist, state FROM matches WHERE match_id = $1 FOR UPDATE` const liveAbandonmentParticipantsSQL = `SELECT player_id, disconnected_at FROM match_participants WHERE match_id = $1 AND participation_active AND abandoned_at IS NULL AND disconnected_at IS NOT NULL ORDER BY player_id FOR UPDATE` const liveAbandonmentHistorySQL = `SELECT starts_at FROM penalties WHERE player_id = $1 AND kind IN ('INITIAL_CONNECT_NO_SHOW', 'MATCH_ABANDONED') ORDER BY starts_at` const liveAbandonmentParticipantSQL = `UPDATE match_participants SET abandoned_at = $3 WHERE match_id = $1 AND player_id = $2 AND participation_active AND abandoned_at IS NULL AND disconnected_at IS NOT NULL RETURNING player_id` const liveAbandonmentPenaltySQL = `INSERT INTO penalties (penalty_id, player_id, match_id, playlist, kind, starts_at, ends_at) VALUES ($1, $2, $3, 'ranked', 'MATCH_ABANDONED', $4, $5) ON CONFLICT (penalty_id) DO NOTHING` const liveAbandonmentRevisionSQL = `UPDATE matches SET revision = revision + 1 WHERE match_id = $1 AND state = 'LIVE' RETURNING revision` const liveAbandonmentTargetsSQL = `SELECT player_id FROM match_participants WHERE match_id = $1 AND participation_active ORDER BY player_id` const liveAbandonmentOutboxSQL = `INSERT INTO outbox (event_id, aggregate_type, aggregate_id, revision, event_type, payload) VALUES ($1, 'match', $2, $3, 'state_changed', $4)` // ReconcileLiveAbandonments applies a bounded, durable reconnect-grace sweep. // It does not deactivate participants or alter LIVE tickets: an abandonment // must remain in the authoritative result roster so rating correctly scores a // loss if the match later completes. func ReconcileLiveAbandonments(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 live-abandonment maintenance arguments") } rows, err := db.QueryContext(ctx, liveAbandonmentCandidatesSQL, now.Add(-domain.RankedReconnectGrace), limit) if err != nil { return 0, err } defer rows.Close() var matchIDs []string for rows.Next() { var matchID string if err := rows.Scan(&matchID); err != nil { return 0, err } matchIDs = append(matchIDs, matchID) } if err := rows.Err(); err != nil { return 0, err } // Do not hold the candidate cursor while opening serializable per-match // transactions. A deliberately small production pool (including size one) // would otherwise wait on its own still-open read connection. if err := rows.Close(); err != nil { return 0, err } count := 0 for _, matchID := range matchIDs { changed, err := ApplyLiveAbandonments(ctx, db, matchID, now) if err != nil { return count, err } if changed > 0 { count++ } } return count, nil } // ApplyLiveAbandonments is independently serializable so concurrent // maintenance replicas or a result submission cannot double-penalise a // player. It returns the number of participants newly abandoned. func ApplyLiveAbandonments(ctx context.Context, db *sql.DB, matchID string, now time.Time) (int, error) { if db == nil || matchID == "" || now.IsZero() { return 0, fmt.Errorf("invalid live-abandonment transaction arguments") } changed := 0 err := RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error { var playlist, state string if err := tx.QueryRowContext(ctx, liveAbandonmentMatchLockSQL, matchID).Scan(&playlist, &state); err != nil { return err } if playlist != string(domain.Ranked) || state != string(domain.Live) { return nil } participants, err := loadLiveReconnectParticipants(ctx, tx, matchID) if err != nil { return err } history, err := loadLiveAbandonmentHistory(ctx, tx, participants) if err != nil { return err } planned, err := domain.PlanRankedAbandonments(now, participants, history) if err != nil { return err } if len(planned) == 0 { return nil } for _, abandonment := range planned { var playerID string if err := tx.QueryRowContext(ctx, liveAbandonmentParticipantSQL, matchID, abandonment.PlayerID, abandonment.AbandonedAt).Scan(&playerID); err != nil { if err == sql.ErrNoRows { return fmt.Errorf("%w: reconnect participant changed", domain.ErrConflict) } return err } penaltyID := "live-abandon:" + matchID + ":" + abandonment.PlayerID if _, err := tx.ExecContext(ctx, liveAbandonmentPenaltySQL, penaltyID, abandonment.PlayerID, matchID, abandonment.AbandonedAt, abandonment.AbandonedAt.Add(abandonment.Cooldown)); err != nil { return err } } var revision uint64 if err := tx.QueryRowContext(ctx, liveAbandonmentRevisionSQL, matchID).Scan(&revision); err != nil { return err } targets, err := loadLiveAbandonmentTargets(ctx, tx, matchID) if err != nil { return err } if len(targets) == 0 { return fmt.Errorf("%w: live match has no active event targets", domain.ErrConflict) } payload, err := MarshalOutboxEnvelope(OutboxEnvelope{ Event: "state_changed", ResourceID: matchID, Revision: int64(revision), OccurredAt: now, State: string(domain.Live), MatchID: matchID, PlayerIDs: targets, Extra: map[string]any{"abandoned_player_ids": abandonmentIDs(planned)}, }) if err != nil { return err } if _, err := tx.ExecContext(ctx, liveAbandonmentOutboxSQL, fmt.Sprintf("live-abandon:%s:%d", matchID, revision), matchID, revision, payload); err != nil { return err } changed = len(planned) return nil }) return changed, err } func loadLiveReconnectParticipants(ctx context.Context, tx *sql.Tx, matchID string) ([]domain.ReconnectParticipant, error) { rows, err := tx.QueryContext(ctx, liveAbandonmentParticipantsSQL, matchID) if err != nil { return nil, err } defer rows.Close() participants := make([]domain.ReconnectParticipant, 0) for rows.Next() { var participant domain.ReconnectParticipant if err := rows.Scan(&participant.PlayerID, &participant.DisconnectedAt); err != nil { return nil, err } participants = append(participants, participant) } return participants, rows.Err() } func loadLiveAbandonmentHistory(ctx context.Context, tx *sql.Tx, participants []domain.ReconnectParticipant) (map[string][]time.Time, error) { history := make(map[string][]time.Time, len(participants)) for _, participant := range participants { rows, err := tx.QueryContext(ctx, liveAbandonmentHistorySQL, 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 } if err := rows.Close(); err != nil { return nil, err } } return history, nil } func loadLiveAbandonmentTargets(ctx context.Context, tx *sql.Tx, matchID string) ([]string, error) { rows, err := tx.QueryContext(ctx, liveAbandonmentTargetsSQL, matchID) if err != nil { return nil, err } defer rows.Close() var players []string for rows.Next() { var playerID string if err := rows.Scan(&playerID); err != nil { return nil, err } players = append(players, playerID) } return players, rows.Err() } func abandonmentIDs(abandonments []domain.Abandonment) []string { ids := make([]string, len(abandonments)) for i := range abandonments { ids[i] = abandonments[i].PlayerID } return ids }