CosmicClash/server/store/live_abandonment_sql.go

package store

import (
	"context"
	"database/sql"
	"encoding/json"
	"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 := json.Marshal(map[string]any{
			"event": "state_changed", "revision": revision, "resource_id": matchID,
			"occurred_at": now, "state": domain.Live, "match_id": matchID,
			"player_ids": targets, "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
}