feat: apply certified ratings during result completion

server/store/result_sql.go

@@ -51,6 +51,23 @@ WHERE player_id = ANY($1)
 ORDER BY player_id
 FOR UPDATE`
 
+const MatchParticipantRatingsSQL = `SELECT mp.player_id, mp.team, 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
+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 {
@@ -65,7 +82,7 @@ func (r PostgresResults) SubmitResult(ctx context.Context, resultID string, resu
 		return err
 	}
 	receipt := domain.ResultReceipt{ResultID: resultID, MatchID: result.MatchID, ResultNonce: result.ResultNonce, PayloadDigest: domain.ResultDigest(result), IntegrityState: result.IntegrityState, ReceivedAt: now}
-	return CompleteResult(ctx, r.DB, receipt, binding.ServerID, resultID, payload, now)
+	return CompleteResultWithResult(ctx, r.DB, receipt, binding.ServerID, resultID, payload, result, now)
 }
 
 // CompleteResult is the durable receipt/reconciliation boundary. The caller
@@ -73,15 +90,23 @@ func (r PostgresResults) SubmitResult(ctx context.Context, resultID string, resu
 // 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 {
+	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 receipt.ResultID == "" || receipt.MatchID == "" || serverID == "" || eventID == "" || len(payload) == 0 {
 		return fmt.Errorf("invalid result transaction arguments")
 	}
 	return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
-		result, err := tx.ExecContext(ctx, ResultReceiptInsertSQL, receipt.ResultID, receipt.MatchID, receipt.ResultNonce, receipt.PayloadDigest[:], string(receipt.IntegrityState), receipt.ReceivedAt)
+		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 := result.RowsAffected()
+		inserted, err := insertResult.RowsAffected()
 		if err != nil {
 			return err
 		}
@@ -108,6 +133,11 @@ func CompleteResult(ctx context.Context, db *sql.DB, receipt domain.ResultReceip
 		if state != "RESULT_PENDING" {
 			return fmt.Errorf("match is not result-pending: %s", state)
 		}
+		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
@@ -126,3 +156,124 @@ func CompleteResult(ctx context.Context, db *sql.DB, receipt domain.ResultReceip
 		return err
 	})
 }
+
+type participantRating struct {
+	playerID    string
+	team        int
+	rating      domain.Rating
+	rankedGames int
+}
+
+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
+		if err := rows.Scan(&player.playerID, &player.team, &player.rating.Value, &player.rating.RD, &player.rating.Volatility, &player.rankedGames, &player.rating.LastRatedAt); err != nil {
+			return err
+		}
+		players = append(players, player)
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	if len(players) == 0 {
+		return nil
+	}
+	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}
+	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
+}

server/store/result_sql_test.go

@@ -11,6 +11,9 @@ func TestResultSQLPreservesReceiptConflictAndAtomicCommitBoundaries(t *testing.T
 		ResultReceiptCommitSQL: {"COALESCE(committed_at", "committed_at"},
 		ResultOutboxSQL:        {"match_completed", "aggregate_id", "revision"},
 		RatingLockSQL:          {"ORDER BY player_id", "FOR UPDATE"},
+		MatchParticipantRatingsSQL: {"match_participants", "JOIN ratings", "ORDER BY mp.player_id"},
+		RatingValuesSQL:            {"player_id = ANY($1)", "ORDER BY player_id"},
+		RatingUpdateSQL:            {"ranked_games = ranked_games + $5", "revision = revision + 1"},
 	}
 	for query, fragments := range checks {
 		for _, fragment := range fragments {