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 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`

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")
	}
	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)
}

// 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
	})
}