CosmicClash/server/store/result_sql.go

package store

import (
	"bytes"
	"context"
	"database/sql"
	"fmt"
	"time"

	"github.com/cosmic-clash/cosmic-clash/server/domain"
)

// ResultReceiptInsertSQL intentionally uses DO NOTHING. The adapter must
// select the existing receipt afterward and compare its digest; an identical
// retry is acknowledged, while a different payload is a conflict with no
// update side effect.
const ResultReceiptInsertSQL = `INSERT INTO result_receipts
    (result_id, match_id, result_nonce, payload_digest, integrity_state, received_at)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT DO NOTHING`

const ResultReceiptSelectSQL = `SELECT result_id, match_id, result_nonce, payload_digest,
    integrity_state, received_at, committed_at
FROM result_receipts
WHERE match_id = $1
FOR UPDATE`

// ResultCommitLockSQL establishes the match lock before participant/rating
// locks. Rating rows are then locked in lexical player-ID order by the
// adapter, ensuring every concurrent result computes from one snapshot.
const ResultCommitLockSQL = `SELECT match_id, playlist, state, revision
FROM matches
WHERE match_id = $1 AND server_id = $2
FOR UPDATE`

const ResultMatchPendingSQL = `UPDATE matches
SET state = 'RESULT_PENDING', revision = revision + 1
WHERE match_id = $1 AND state = 'LIVE'`

const ResultTicketsPendingSQL = `UPDATE queue_tickets q
SET state = 'RESULT_PENDING', revision = revision + 1
FROM match_participants mp
WHERE mp.match_id = $1 AND q.ticket_id = mp.ticket_id AND q.player_id = mp.player_id
  AND mp.participation_active AND q.state = 'LIVE'`

const ResultMatchCompleteSQL = `UPDATE matches
SET state = 'COMPLETED', revision = revision + 1, completed_at = $2
WHERE match_id = $1 AND state = 'RESULT_PENDING'`

const ResultReceiptCommitSQL = `UPDATE result_receipts
SET committed_at = COALESCE(committed_at, $2)
WHERE match_id = $1`

const ResultTicketsCompleteSQL = `UPDATE queue_tickets q
SET state = 'COMPLETED', revision = revision + 1
FROM match_participants mp
WHERE mp.match_id = $1 AND q.ticket_id = mp.ticket_id AND q.player_id = mp.player_id
  AND mp.participation_active AND q.state = 'RESULT_PENDING'`

const ResultParticipantCountSQL = `SELECT count(*) FROM match_participants WHERE match_id = $1 AND participation_active`

const ResultOutboxSQL = `INSERT INTO outbox
    (event_id, aggregate_type, aggregate_id, revision, event_type, payload)
VALUES ($1, 'match', $2, $3, 'match_completed', $4)`

const RatingLockSQL = `SELECT player_id, rating, deviation, volatility, ranked_games, revision
FROM ratings
WHERE player_id = ANY($1)
ORDER BY player_id
FOR UPDATE`

const MatchParticipantRatingsSQL = `SELECT mp.player_id, mp.team, mp.abandoned_at, r.rating, r.deviation,
    r.volatility, r.ranked_games, r.updated_at
FROM match_participants mp
JOIN ratings r ON r.player_id = mp.player_id
WHERE mp.match_id = $1
  AND mp.participation_active
ORDER BY mp.player_id`

const RatingValuesSQL = `SELECT player_id, rating, deviation, volatility, ranked_games, updated_at
FROM ratings
WHERE player_id = ANY($1)
ORDER BY player_id`

const RatingUpdateSQL = `UPDATE ratings
SET rating = $2, deviation = $3, volatility = $4,
    ranked_games = ranked_games + $5, updated_at = $6, revision = revision + 1
WHERE player_id = $1`

type PostgresResults struct{ DB *sql.DB }

func (r PostgresResults) SubmitResult(ctx context.Context, resultID string, result domain.MatchResult, binding domain.WorkloadBinding, payload []byte, now time.Time) error {
	if r.DB == nil || resultID == "" || binding.ServerID == "" || binding.MatchID != result.MatchID || binding.ServerID != result.ServerID || len(payload) == 0 || now.IsZero() {
		return fmt.Errorf("invalid result submission")
	}
	validator, err := domain.NewResultStore(binding)
	if err != nil {
		return err
	}
	if _, _, err := validator.Submit(resultID, result, binding, now); err != nil {
		return err
	}
	receipt := domain.ResultReceipt{ResultID: resultID, MatchID: result.MatchID, ResultNonce: result.ResultNonce, PayloadDigest: domain.ResultDigest(result), IntegrityState: result.IntegrityState, ReceivedAt: now}
	return CompleteResultWithResult(ctx, r.DB, receipt, binding.ServerID, resultID, payload, result, now)
}

// CompleteResult is the durable receipt/reconciliation boundary. The caller
// must have already authenticated the workload and computed the receipt
// digest. Duplicate identical receipts continue the same completion path;
// conflicting payloads fail without mutating the existing receipt.
func CompleteResult(ctx context.Context, db *sql.DB, receipt domain.ResultReceipt, serverID, eventID string, payload []byte, now time.Time) error {
	return completeResult(ctx, db, receipt, serverID, eventID, payload, now, nil)
}

func CompleteResultWithResult(ctx context.Context, db *sql.DB, receipt domain.ResultReceipt, serverID, eventID string, payload []byte, result domain.MatchResult, now time.Time) error {
	if result.MatchID != receipt.MatchID || result.ServerID != serverID || result.ResultNonce != receipt.ResultNonce || result.IntegrityState != receipt.IntegrityState || domain.ResultDigest(result) != receipt.PayloadDigest {
		return fmt.Errorf("result does not match receipt")
	}
	return completeResult(ctx, db, receipt, serverID, eventID, payload, now, &result)
}

func completeResult(ctx context.Context, db *sql.DB, receipt domain.ResultReceipt, serverID, eventID string, payload []byte, now time.Time, result *domain.MatchResult) error {
	if db == nil || receipt.ResultID == "" || receipt.MatchID == "" || len(receipt.ResultNonce) < 16 || len(receipt.ResultNonce) > 128 || receipt.ReceivedAt.IsZero() || now.IsZero() || serverID == "" || eventID == "" || len(payload) == 0 || (receipt.IntegrityState != domain.IntegrityCertified && receipt.IntegrityState != domain.IntegritySuppressed && receipt.IntegrityState != domain.IntegrityReview) {
		return fmt.Errorf("invalid result transaction arguments")
	}
	return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
		insertResult, err := tx.ExecContext(ctx, ResultReceiptInsertSQL, receipt.ResultID, receipt.MatchID, receipt.ResultNonce, receipt.PayloadDigest[:], string(receipt.IntegrityState), receipt.ReceivedAt)
		if err != nil {
			return err
		}
		inserted, err := insertResult.RowsAffected()
		if err != nil {
			return err
		}
		if inserted == 0 {
			var priorID, priorMatch, priorNonce, priorIntegrity string
			var priorDigest []byte
			var receivedAt, committedAt time.Time
			if err := tx.QueryRowContext(ctx, ResultReceiptSelectSQL, receipt.MatchID).Scan(&priorID, &priorMatch, &priorNonce, &priorDigest, &priorIntegrity, &receivedAt, &committedAt); err != nil {
				return fmt.Errorf("result receipt conflict: %w", err)
			}
			if priorID != receipt.ResultID || priorMatch != receipt.MatchID || priorNonce != receipt.ResultNonce || priorIntegrity != string(receipt.IntegrityState) || !bytes.Equal(priorDigest, receipt.PayloadDigest[:]) {
				return fmt.Errorf("%w: durable receipt differs", domain.ErrResultConflict)
			}
		}
		var lockedMatch, playlist, state string
		var revision uint64
		if err := tx.QueryRowContext(ctx, ResultCommitLockSQL, receipt.MatchID, serverID).Scan(&lockedMatch, &playlist, &state, &revision); err != nil {
			return err
		}
		if state == "COMPLETED" {
			_, err := tx.ExecContext(ctx, ResultReceiptCommitSQL, receipt.MatchID, now)
			return err
		}
		if state == string(domain.Live) {
			updated, err := tx.ExecContext(ctx, ResultMatchPendingSQL, receipt.MatchID)
			if err != nil {
				return err
			}
			if changed, err := updated.RowsAffected(); err != nil || changed != 1 {
				return fmt.Errorf("result-pending transition lost race")
			}
			state = string(domain.ResultPending)
			revision++
		}
		if state != "RESULT_PENDING" {
			return fmt.Errorf("match is not result-pending: %s", state)
		}
		if _, err := tx.ExecContext(ctx, ResultTicketsPendingSQL, receipt.MatchID); err != nil {
			return err
		}
		if result != nil && domain.RatingEligible(receipt) {
			if err := applyResultRatings(ctx, tx, receipt.MatchID, domain.Playlist(playlist), *result, now); err != nil {
				return err
			}
		}
		updated, err := tx.ExecContext(ctx, ResultMatchCompleteSQL, receipt.MatchID, now)
		if err != nil {
			return err
		}
		changed, err := updated.RowsAffected()
		if err != nil {
			return err
		}
		if changed != 1 {
			return fmt.Errorf("result completion lost race")
		}
		completedTickets, err := tx.ExecContext(ctx, ResultTicketsCompleteSQL, receipt.MatchID)
		if err != nil {
			return err
		}
		var participants int64
		if err := tx.QueryRowContext(ctx, ResultParticipantCountSQL, receipt.MatchID).Scan(&participants); err != nil {
			return err
		}
		completed, err := completedTickets.RowsAffected()
		if err != nil {
			return err
		}
		if completed != participants {
			return fmt.Errorf("result ticket completion mismatch: completed=%d participants=%d", completed, participants)
		}
		if _, err := tx.ExecContext(ctx, ResultReceiptCommitSQL, receipt.MatchID, now); err != nil {
			return err
		}
		_, err = tx.ExecContext(ctx, ResultOutboxSQL, eventID, receipt.MatchID, revision+1, payload)
		return err
	})
}

type participantRating struct {
	playerID    string
	team        int
	rating      domain.Rating
	rankedGames int
	abandoned   bool
}

func applyResultRatings(ctx context.Context, tx *sql.Tx, matchID string, playlist domain.Playlist, result domain.MatchResult, now time.Time) error {
	rows, err := tx.QueryContext(ctx, MatchParticipantRatingsSQL, matchID)
	if err != nil {
		return err
	}
	defer rows.Close()
	var players []participantRating
	for rows.Next() {
		var player participantRating
		var abandonedAt sql.NullTime
		if err := rows.Scan(&player.playerID, &player.team, &abandonedAt, &player.rating.Value, &player.rating.RD, &player.rating.Volatility, &player.rankedGames, &player.rating.LastRatedAt); err != nil {
			return err
		}
		player.abandoned = abandonedAt.Valid
		players = append(players, player)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if err := rows.Close(); err != nil {
		return err
	}
	if len(players) == 0 {
		return nil
	}
	var participantCount int
	if err := tx.QueryRowContext(ctx, ResultParticipantCountSQL, matchID).Scan(&participantCount); err != nil {
		return err
	}
	if participantCount != len(players) {
		return fmt.Errorf("result rating roster is incomplete")
	}
	ids := make([]string, len(players))
	for i := range players {
		ids[i] = players[i].playerID
	}
	// Lock all rating rows in lexical order before computing updates. This
	// matches the lock order used by every result transaction and prevents
	// cross-match deadlocks.
	locked, err := tx.QueryContext(ctx, RatingLockSQL, ids)
	if err != nil {
		return err
	}
	for locked.Next() {
		var ignored string
		var rating domain.Rating
		var games int
		var revision uint64
		if err := locked.Scan(&ignored, &rating.Value, &rating.RD, &rating.Volatility, &games, &revision); err != nil {
			locked.Close()
			return err
		}
	}
	if err := locked.Err(); err != nil {
		locked.Close()
		return err
	}
	if err := locked.Close(); err != nil {
		return err
	}
	// Re-read after acquiring the locks so the calculations use the values
	// protected by those locks rather than a pre-lock snapshot.
	values, err := tx.QueryContext(ctx, RatingValuesSQL, ids)
	if err != nil {
		return err
	}
	ratings := make(map[string]domain.Rating, len(players))
	for values.Next() {
		var playerID string
		var rating domain.Rating
		var rankedGames int
		if err := values.Scan(&playerID, &rating.Value, &rating.RD, &rating.Volatility, &rankedGames, &rating.LastRatedAt); err != nil {
			values.Close()
			return err
		}
		ratings[playerID] = rating
	}
	if err := values.Err(); err != nil {
		values.Close()
		return err
	}
	if err := values.Close(); err != nil {
		return err
	}
	outcome := domain.MatchOutcome{Team0Score: result.Team0Score, Team1Score: result.Team1Score, Abandoners: make(map[string]bool)}
	for _, player := range players {
		if player.abandoned {
			outcome.Abandoners[player.playerID] = true
		}
	}
	for _, player := range players {
		current, ok := ratings[player.playerID]
		if !ok {
			return fmt.Errorf("rating row disappeared for player %s", player.playerID)
		}
		opponents := make([]domain.Opponent, 0, len(players)-1)
		for _, opponent := range players {
			if opponent.team != player.team {
				score, err := domain.ScoreForPlayer(outcome, player.playerID, player.team)
				if err != nil {
					return err
				}
				opponents = append(opponents, domain.Opponent{PlayerID: opponent.playerID, Rating: ratings[opponent.playerID], Score: score})
			}
		}
		var weighted []domain.Opponent
		if playlist == domain.Ranked {
			weighted, err = domain.RankedOpponents(opponents)
		} else if playlist == domain.Casual {
			weighted, err = domain.CasualOpponents(opponents)
		} else {
			return fmt.Errorf("unsupported result playlist")
		}
		if err != nil {
			return err
		}
		updated, err := domain.UpdateRating(current, weighted, now)
		if err != nil {
			return err
		}
		rankedIncrement := 0
		if playlist == domain.Ranked {
			rankedIncrement = 1
		}
		if _, err := tx.ExecContext(ctx, RatingUpdateSQL, player.playerID, updated.Value, updated.RD, updated.Volatility, rankedIncrement, now); err != nil {
			return err
		}
	}
	return nil
}