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 }