mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-11 17:43:42 +00:00
112 lines
4.3 KiB
Go
112 lines
4.3 KiB
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 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 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`
|
|
|
|
// 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 {
|
|
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)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
inserted, err := result.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("conflicting result receipt")
|
|
}
|
|
}
|
|
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 != "RESULT_PENDING" {
|
|
return fmt.Errorf("match is not result-pending: %s", state)
|
|
}
|
|
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")
|
|
}
|
|
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
|
|
})
|
|
}
|