feat: atomically complete match results

server/store/result_sql.go

@@ -1,5 +1,15 @@
 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
@@ -18,7 +28,7 @@ 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
+const ResultCommitLockSQL = `SELECT match_id, playlist, state, revision
 FROM matches
 WHERE match_id = $1 AND server_id = $2
 FOR UPDATE`
@@ -27,6 +37,10 @@ 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)`
@@ -36,3 +50,62 @@ 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
+	})
+}

server/store/result_sql_test.go

@@ -8,6 +8,7 @@ func TestResultSQLPreservesReceiptConflictAndAtomicCommitBoundaries(t *testing.T
 		ResultReceiptSelectSQL: {"FOR UPDATE", "committed_at"},
 		ResultCommitLockSQL:    {"server_id = $2", "FOR UPDATE"},
 		ResultMatchCompleteSQL: {"state = 'RESULT_PENDING'", "revision = revision + 1"},
+		ResultReceiptCommitSQL: {"COALESCE(committed_at", "committed_at"},
 		ResultOutboxSQL:        {"match_completed", "aggregate_id", "revision"},
 		RatingLockSQL:          {"ORDER BY player_id", "FOR UPDATE"},
 	}