CosmicClash/server/store/postgres_integration_test.go

//go:build integration

package store

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"fmt"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/cosmic-clash/cosmic-clash/server/domain"
	"github.com/cosmic-clash/cosmic-clash/server/migrations"
	_ "github.com/jackc/pgx/v5/stdlib"
)

// This binary is deliberately opt-in. It requires a disposable PostgreSQL
// instance supplied by scripts/run_postgres_integration.sh.
func openIntegrationPostgres(t *testing.T) *sql.DB {
	t.Helper()
	dsn := os.Getenv("COSMIC_CLASH_POSTGRES_DSN")
	if dsn == "" {
		t.Skip("COSMIC_CLASH_POSTGRES_DSN is not set")
	}
	db, err := sql.Open("pgx", dsn)
	if err != nil {
		t.Fatalf("open PostgreSQL: %v", err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := db.PingContext(ctx); err != nil {
		db.Close()
		t.Fatalf("ping PostgreSQL: %v", err)
	}
	t.Cleanup(func() { db.Close() })
	return db
}

func applyIntegrationMigrations(t *testing.T, db *sql.DB) {
	t.Helper()
	if _, err := db.ExecContext(context.Background(), `DROP TABLE IF EXISTS schema_migrations, assignments, audit_events, outbox, result_receipts, ranked_season_rollovers, penalties, seasons, ratings, match_participants, matches, proposal_participants, proposals, allocations, game_servers, queue_tickets, idempotency_keys, sessions, identities CASCADE`); err != nil {
		t.Fatalf("reset PostgreSQL schema: %v", err)
	}
	if err := migrations.Apply(context.Background(), db, filepath.Join("..", "migrations")); err != nil {
		t.Fatalf("apply migrations: %v", err)
	}
}

func TestPostgreSQLAllocatorClaimReplayAndCapacityFence(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)
	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	servers := []domain.ReadyServer{
		{ServerID: "allocator-server-b", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady},
		{ServerID: "allocator-server-a", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady},
	}
	for _, server := range servers {
		if err := RegisterReadyServer(ctx, db, server, now); err != nil {
			t.Fatal(err)
		}
	}
	request := domain.AllocationRequest{AllocationID: "allocation-integration-1", MatchID: "match-integration-1", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}
	allocation, err := ClaimAllocation(ctx, db, request, now)
	if err != nil {
		t.Fatalf("claim: %v", err)
	}
	if allocation.ServerID != "allocator-server-a" || allocation.State != domain.ServerAllocated {
		t.Fatalf("allocation=%+v", allocation)
	}
	if err := RegisterReadyServer(ctx, db, servers[1], now.Add(time.Second)); err != nil {
		t.Fatalf("stale Ready projection: %v", err)
	}
	var lifecycle string
	if err := db.QueryRowContext(ctx, `SELECT state FROM game_servers WHERE server_id = 'allocator-server-a'`).Scan(&lifecycle); err != nil || lifecycle != "ALLOCATED" {
		t.Fatalf("stale Ready projection reopened allocation state=%q err=%v", lifecycle, err)
	}
	replay, err := ClaimAllocation(ctx, db, request, now.Add(time.Second))
	if err != nil || replay.ServerID != allocation.ServerID || !replay.AllocatedAt.Equal(allocation.AllocatedAt) {
		t.Fatalf("replay=%+v err=%v", replay, err)
	}
	conflict := request
	conflict.MatchID = "match-integration-other"
	if _, err := ClaimAllocation(ctx, db, conflict, now); err != domain.ErrConflict {
		t.Fatalf("conflicting replay err=%v", err)
	}
	second := request
	second.AllocationID = "allocation-integration-2"
	second.MatchID = "match-integration-2"
	if _, err := ClaimAllocation(ctx, db, second, now); err != nil {
		t.Fatalf("second claim: %v", err)
	}
	third := second
	third.AllocationID = "allocation-integration-3"
	third.MatchID = "match-integration-3"
	if _, err := ClaimAllocation(ctx, db, third, now); err != domain.ErrNoCapacity {
		t.Fatalf("capacity err=%v", err)
	}
}

func TestPostgreSQLAcceptedProposalPromotesOneAtomicAllocatingMatch(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)
	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	for _, player := range []string{"promote-a", "promote-b"} {
		if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
			t.Fatal(err)
		}
	}
	for index, player := range []string{"promote-a", "promote-b"} {
		if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ($1, $2, 'casual', 'PROPOSED', 'build-1', 1, $3, $4)`, fmt.Sprintf("promote-ticket-%d", index), player, now, now.Add(time.Minute)); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO proposals (proposal_id, playlist, state, expires_at, revision) VALUES ('promote-proposal', 'casual', 'ACCEPTED', $1, 2)`, now.Add(time.Minute)); err != nil {
		t.Fatal(err)
	}
	for index, player := range []string{"promote-a", "promote-b"} {
		if _, err := db.ExecContext(ctx, `INSERT INTO proposal_participants (proposal_id, player_id, ticket_id, response) VALUES ('promote-proposal', $1, $2, 'ACCEPTED')`, player, fmt.Sprintf("promote-ticket-%d", index)); err != nil {
			t.Fatal(err)
		}
	}
	plan := AcceptedMatchPlan{MatchID: "promote-match", ProposalID: "promote-proposal", Region: "EU", Protocol: 1, Players: []MatchPlayer{{PlayerID: "promote-a", Team: 0, Slot: 0}, {PlayerID: "promote-b", Team: 1, Slot: 3}}}
	if err := CreateMatchFromAcceptedProposal(ctx, db, plan, now); err != nil {
		t.Fatalf("promote accepted proposal: %v", err)
	}
	var state string
	if err := db.QueryRowContext(ctx, `SELECT state FROM matches WHERE match_id = 'promote-match'`).Scan(&state); err != nil || state != "ALLOCATING" {
		t.Fatalf("match state=%q err=%v", state, err)
	}
	var acceptedTickets, participantCount int
	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM queue_tickets WHERE ticket_id LIKE 'promote-ticket-%' AND state = 'ACCEPTED'`).Scan(&acceptedTickets); err != nil || acceptedTickets != 2 {
		t.Fatalf("accepted tickets=%d err=%v", acceptedTickets, err)
	}
	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM match_participants WHERE match_id = 'promote-match'`).Scan(&participantCount); err != nil || participantCount != 2 {
		t.Fatalf("participants=%d err=%v", participantCount, err)
	}
	if err := CreateMatchFromAcceptedProposal(ctx, db, plan, now.Add(time.Second)); err != nil {
		t.Fatalf("identical match promotion replay: %v", err)
	}
	conflict := plan
	conflict.Players = append([]MatchPlayer(nil), plan.Players...)
	conflict.Players[1].Slot = 4
	if err := CreateMatchFromAcceptedProposal(ctx, db, conflict, now.Add(2*time.Second)); err == nil {
		t.Fatal("conflicting match promotion replay was accepted")
	}
}

func TestPostgreSQLAllocationMatchClaimLeaseAndBindFence(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)
	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	for index, player := range []string{"allocation-match-a", "allocation-match-b"} {
		if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
			t.Fatal(err)
		}
		if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ($1, $2, 'casual', 'ACCEPTED', 'build-1', 1, $3, $4)`, fmt.Sprintf("allocation-match-ticket-%d", index), player, now, now.Add(time.Minute)); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version) VALUES ('allocation-match', 'casual', 'ALLOCATING', 'EU', 1)`); err != nil {
		t.Fatal(err)
	}
	for index, player := range []string{"allocation-match-a", "allocation-match-b"} {
		if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('allocation-match', $1, $2, $3, $4)`, player, fmt.Sprintf("allocation-match-ticket-%d", index), index*3, index); err != nil {
			t.Fatal(err)
		}
	}
	claim, found, err := ClaimAllocatingMatch(ctx, db, "enet", now)
	if err != nil || !found || claim.Request != (domain.AllocationRequest{AllocationID: "allocation-allocation-match", MatchID: "allocation-match", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}) {
		t.Fatalf("claim=%+v found=%t err=%v", claim, found, err)
	}
	if err := ReleaseAllocatedMatchClaim(ctx, db, claim.Request.MatchID, "different-allocation"); err != domain.ErrConflict {
		t.Fatalf("wrong-claim release err=%v", err)
	}
	if err := ReleaseAllocatedMatchClaim(ctx, db, claim.Request.MatchID, claim.Request.AllocationID); err != nil {
		t.Fatalf("release claim: %v", err)
	}
	reclaimed, found, err := ClaimAllocatingMatch(ctx, db, "enet", now.Add(time.Second))
	if err != nil || !found || reclaimed.Request.AllocationID != claim.Request.AllocationID {
		t.Fatalf("reclaimed=%+v found=%t err=%v", reclaimed, found, err)
	}
	server := domain.ReadyServer{ServerID: "allocation-server", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady}
	if err := RegisterReadyServer(ctx, db, server, now); err != nil {
		t.Fatalf("register allocation server: %v", err)
	}
	allocation, err := ClaimAllocation(ctx, db, reclaimed.Request, now.Add(time.Second))
	if err != nil {
		t.Fatalf("record provider allocation: %v", err)
	}
	if err := BindAllocatedMatch(ctx, db, allocation); err != nil {
		t.Fatalf("bind allocation: %v", err)
	}
	var allocatingTickets int
	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM queue_tickets WHERE ticket_id LIKE 'allocation-match-ticket-%' AND state = 'ALLOCATING'`).Scan(&allocatingTickets); err != nil || allocatingTickets != 2 {
		t.Fatalf("allocating tickets=%d err=%v", allocatingTickets, err)
	}
	if _, found, err := ClaimAllocatingMatch(ctx, db, "enet", now.Add(2*time.Second)); err != nil || found {
		t.Fatalf("bound match re-claimed found=%t err=%v", found, err)
	}
}

func TestPostgreSQLQueueAdapterAgainstRealDatabase(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)

	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('integration-player', 'integration-steam')`); err != nil {
		t.Fatal(err)
	}
	spec := domain.QueueSpec{Playlist: domain.Casual, ClientBuild: "integration-build", ProtocolVersion: 1}
	ticket, err := CreateQueueTicket(ctx, db, "integration-ticket", "integration-player", "integration-create-0001", spec, now)
	if err != nil {
		t.Fatalf("create queue ticket: %v", err)
	}
	if ticket.State != domain.Queued || ticket.Revision != 0 {
		t.Fatalf("unexpected ticket: %+v", ticket)
	}
	replay, err := CreateQueueTicket(ctx, db, "integration-ticket", "integration-player", "integration-create-0001", spec, now.Add(time.Second))
	if err != nil {
		t.Fatalf("idempotent queue replay: %v", err)
	}
	if replay.TicketID != ticket.TicketID || !replay.ExpiresAt.Equal(ticket.ExpiresAt) {
		t.Fatalf("replay changed durable result: %+v vs %+v", replay, ticket)
	}
	if _, err := CreateQueueTicket(ctx, db, "integration-ticket-2", "integration-player", "integration-create-0002", spec, now); err == nil {
		t.Fatal("second active player ticket was accepted")
	}
	if _, err := GetQueueTicket(ctx, db, "integration-player", "integration-ticket", now); err != nil {
		t.Fatalf("owner recovery: %v", err)
	}
	if _, err := GetQueueTicket(ctx, db, "other-player", "integration-ticket", now); err == nil {
		t.Fatal("non-owner recovered queue ticket")
	}
}

func TestPostgreSQLQueueHeartbeatAndCancelAreRevisionFenced(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)

	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('heartbeat-player', 'heartbeat-steam')`); err != nil {
		t.Fatal(err)
	}
	spec := domain.QueueSpec{Playlist: domain.Ranked, ClientBuild: "integration-build", ProtocolVersion: 1}
	if _, err := CreateQueueTicket(ctx, db, "heartbeat-ticket", "heartbeat-player", "heartbeat-create-0001", spec, now); err != nil {
		t.Fatal(err)
	}
	heartbeat, err := HeartbeatQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000001", 0, now.Add(5*time.Second))
	if err != nil {
		t.Fatalf("heartbeat: %v", err)
	}
	if heartbeat.Revision != 1 || !heartbeat.ExpiresAt.Equal(now.Add(35*time.Second)) {
		t.Fatalf("unexpected heartbeat result: %+v", heartbeat)
	}
	if _, err := HeartbeatQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000002", 0, now.Add(6*time.Second)); err == nil {
		t.Fatal("stale heartbeat revision was accepted")
	}
	cancelled, err := CancelQueueTicket(ctx, db, "heartbeat-player", "heartbeat-ticket", "heartbeat-op-0000003", 1, now.Add(7*time.Second))
	if err != nil {
		t.Fatalf("cancel: %v", err)
	}
	if cancelled.State != domain.Cancelled || cancelled.Revision != 2 {
		t.Fatalf("unexpected cancellation result: %+v", cancelled)
	}
}

func TestPostgreSQLAssignmentPersistenceIsPlayerScopedAndExpiryBound(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)

	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	for _, player := range []string{"assignment-player", "assignment-other"} {
		if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ('assignment-ticket', 'assignment-player', 'casual', 'ASSIGNED', 'integration-build', 1, $1, $2)`, now, now.Add(time.Minute)); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id) VALUES ('assignment-match', 'casual', 'ASSIGNED', 'EU', 1, 'assignment-server')`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('assignment-match', 'assignment-player', 'assignment-ticket', 0, 0)`); err != nil {
		t.Fatal(err)
	}
	assignment := DurableAssignment{MatchID: "assignment-match", PlayerID: "assignment-player", AllocationID: "allocation-1", ServerID: "assignment-server", Slot: 0, Region: "EU", ClientBuild: "integration-build", ProtocolVersion: 1, Transport: "enet", Endpoint: "127.0.0.1:7777", JoinAuthorisation: "join-token", ManifestDigest: []byte("manifest"), ExpiresAt: now.Add(time.Minute), Revision: 1}
	if err := SaveAssignment(ctx, db, assignment); err != nil {
		t.Fatalf("save assignment: %v", err)
	}
	got, err := GetAssignment(ctx, db, assignment.PlayerID, assignment.MatchID, now)
	if err != nil {
		t.Fatalf("recover assignment: %v", err)
	}
	if got.JoinAuthorisation != assignment.JoinAuthorisation || got.Slot != assignment.Slot {
		t.Fatalf("assignment changed on round trip: %+v", got)
	}
	if _, err := GetAssignment(ctx, db, "assignment-other", assignment.MatchID, now); err == nil {
		t.Fatal("non-owner recovered assignment")
	}
	if _, err := GetAssignment(ctx, db, assignment.PlayerID, assignment.MatchID, now.Add(2*time.Minute)); err == nil {
		t.Fatal("expired assignment was recovered")
	}
}

func TestPostgreSQLProposalClaimAndResponseAreAtomic(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)

	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	for _, player := range []string{"proposal-player-a", "proposal-player-b"} {
		if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, player); err != nil {
			t.Fatal(err)
		}
	}
	for i, player := range []string{"proposal-player-a", "proposal-player-b"} {
		if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ($1, $2, 'casual', 'QUEUED', 'integration-build', 1, $3, $4)`, fmt.Sprintf("proposal-ticket-%d", i), player, now, now.Add(time.Minute)); err != nil {
			t.Fatal(err)
		}
	}
	proposal, err := domain.NewProposal("proposal-integration", domain.Casual, []string{"proposal-player-a", "proposal-player-b"}, now)
	if err != nil {
		t.Fatal(err)
	}
	if err := CreateProposal(ctx, db, proposal, map[string]string{"proposal-player-a": "proposal-ticket-0", "proposal-player-b": "proposal-ticket-1"}, now); err != nil {
		t.Fatalf("create proposal: %v", err)
	}
	var proposed int
	if err := db.QueryRow(`SELECT count(*) FROM queue_tickets WHERE state = 'PROPOSED'`).Scan(&proposed); err != nil || proposed != 2 {
		t.Fatalf("proposed queue tickets = %d, err = %v", proposed, err)
	}
	recovered, err := GetProposal(ctx, db, "proposal-player-a", proposal.ProposalID, now)
	if err != nil {
		t.Fatalf("recover proposal: %v", err)
	}
	if len(recovered.Participants) != 2 || recovered.Revision != 0 {
		t.Fatalf("unexpected recovered proposal: %+v", recovered)
	}
	accepted, err := RespondToProposal(ctx, db, "proposal-player-a", proposal.ProposalID, "proposal-response-a-0001", true, 0, now)
	if err != nil {
		t.Fatalf("first proposal acceptance: %v", err)
	}
	if accepted.Revision != 1 || accepted.State != domain.Open {
		t.Fatalf("unexpected first acceptance: %+v", accepted)
	}
	accepted, err = RespondToProposal(ctx, db, "proposal-player-b", proposal.ProposalID, "proposal-response-b-0001", true, 1, now)
	if err != nil {
		t.Fatalf("second proposal acceptance: %v", err)
	}
	if accepted.State != domain.Accepted || accepted.Revision != 2 {
		t.Fatalf("proposal did not close after unanimous acceptance: %+v", accepted)
	}
}

func TestPostgreSQLProposalCreationRollsBackPartialClaims(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)
	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('rollback-player', 'rollback-steam')`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ('rollback-ticket', 'rollback-player', 'casual', 'QUEUED', 'integration-build', 1, $1, $2)`, now, now.Add(time.Minute)); err != nil {
		t.Fatal(err)
	}
	proposal, err := domain.NewProposal("rollback-proposal", domain.Casual, []string{"rollback-player", "missing-player"}, now)
	if err != nil {
		t.Fatal(err)
	}
	if err := CreateProposal(ctx, db, proposal, map[string]string{"rollback-player": "rollback-ticket"}, now); err == nil {
		t.Fatal("proposal with missing ticket mapping was accepted")
	}
	var proposals, participants, proposed int
	if err := db.QueryRow(`SELECT count(*) FROM proposals WHERE proposal_id = 'rollback-proposal'`).Scan(&proposals); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT count(*) FROM proposal_participants WHERE proposal_id = 'rollback-proposal'`).Scan(&participants); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT count(*) FROM queue_tickets WHERE ticket_id = 'rollback-ticket' AND state = 'PROPOSED'`).Scan(&proposed); err != nil {
		t.Fatal(err)
	}
	if proposals != 0 || participants != 0 || proposed != 0 {
		t.Fatalf("partial proposal claim was not rolled back: proposals=%d participants=%d proposed=%d", proposals, participants, proposed)
	}
}

func TestPostgreSQLResultCompletionAndOutboxAreAtomicAndReplayable(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)
	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id) VALUES ('result-match', 'casual', 'RESULT_PENDING', 'NA', 1, 'result-server')`); err != nil {
		t.Fatal(err)
	}
	payload := []byte(`{"match_id":"result-match","team0_score":2,"team1_score":1}`)
	digest := sha256.Sum256(payload)
	receipt := domain.ResultReceipt{ResultID: "result-receipt", MatchID: "result-match", ResultNonce: "result-nonce-123456", PayloadDigest: digest, IntegrityState: domain.IntegrityCertified, ReceivedAt: now}
	if err := CompleteResult(ctx, db, receipt, "result-server", "result-event", payload, now); err != nil {
		t.Fatalf("complete result: %v", err)
	}
	var state string
	if err := db.QueryRow(`SELECT state FROM matches WHERE match_id = 'result-match'`).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != "COMPLETED" {
		t.Fatalf("result match state = %s", state)
	}
	events, err := ReadUnpublishedOutbox(ctx, db, 10)
	if err != nil || len(events) != 1 || events[0].EventID != "result-event" {
		t.Fatalf("unpublished result events = %+v, err = %v", events, err)
	}
	if err := MarkOutboxPublished(ctx, db, events[0].EventID, now.Add(time.Second)); err != nil {
		t.Fatalf("ack result event: %v", err)
	}
	if remaining, err := ReadUnpublishedOutbox(ctx, db, 10); err != nil || len(remaining) != 0 {
		t.Fatalf("outbox after ack = %+v, err = %v", remaining, err)
	}
	if err := CompleteResult(ctx, db, receipt, "result-server", "result-event-retry", payload, now.Add(time.Second)); err != nil {
		t.Fatalf("identical completed result replay: %v", err)
	}
	conflict := receipt
	conflict.ResultID = "different-result"
	if err := CompleteResult(ctx, db, conflict, "result-server", "different-event", []byte(`{"conflict":true}`), now.Add(2*time.Second)); err == nil {
		t.Fatal("conflicting completed result was accepted")
	}
}

func TestPostgreSQLRankedSeasonRolloverIsExactlyOnce(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)
	now := time.Now().UTC().Truncate(time.Microsecond)
	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ('season-player', 'season-steam')`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO ratings (player_id, rating, deviation, volatility, ranked_games) VALUES ('season-player', 1900, 100, 0.12, 25)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO seasons (season_id, playlist, starts_at, ends_at) VALUES ('season-1', 'ranked', $1, $2)`, now.Add(-12*7*24*time.Hour), now); err != nil {
		t.Fatal(err)
	}
	profile := domain.RankedProfile{Rating: domain.Rating{Value: 1900, RD: 100, Volatility: 0.12}, RankedGames: 25}
	updated, applied, err := ApplyRankedSeasonRollover(ctx, db, "season-player", "season-1", profile, now)
	if err != nil || !applied {
		t.Fatalf("first season rollover = %+v applied=%v err=%v", updated, applied, err)
	}
	if updated.Value != 1800 || updated.RD != 200 {
		t.Fatalf("unexpected rolled rating: %+v", updated)
	}
	var rating float64
	var markers int
	if err := db.QueryRow(`SELECT rating FROM ratings WHERE player_id = 'season-player'`).Scan(&rating); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT count(*) FROM ranked_season_rollovers WHERE player_id = 'season-player' AND season_id = 'season-1'`).Scan(&markers); err != nil {
		t.Fatal(err)
	}
	if rating != 1800 || markers != 1 {
		t.Fatalf("durable rollover state rating=%v markers=%d", rating, markers)
	}
	_, applied, err = ApplyRankedSeasonRollover(ctx, db, "season-player", "season-1", profile, now.Add(time.Second))
	if err != nil || applied {
		t.Fatalf("duplicate season rollover applied=%v err=%v", applied, err)
	}
	if err := db.QueryRow(`SELECT rating FROM ratings WHERE player_id = 'season-player'`).Scan(&rating); err != nil {
		t.Fatal(err)
	}
	if rating != 1800 {
		t.Fatalf("duplicate rollover changed rating to %v", rating)
	}
}

func TestPostgreSQLMigrationsAreForwardExecutable(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)
	var tableCount int
	if err := db.QueryRow(`SELECT count(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'assignments'`).Scan(&tableCount); err != nil {
		t.Fatal(err)
	}
	if tableCount != 1 {
		t.Fatal("assignments migration did not create its table")
	}
}

func TestPostgreSQLMigrationsRollBackAndReapplyCleanly(t *testing.T) {
	db := openIntegrationPostgres(t)
	applyIntegrationMigrations(t, db)
	dir := filepath.Join("..", "migrations")
	tableExists := func(table string) bool {
		var count int
		if err := db.QueryRow(`SELECT count(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1`, table).Scan(&count); err != nil {
			t.Fatal(err)
		}
		return count == 1
	}
	if !tableExists("assignments") || !tableExists("allocations") {
		t.Fatal("expected forward-applied schema before rollback")
	}

	// Roll back every migration one at a time, in reverse, checking each
	// down file actually undoes what its forward file created — not just
	// that Rollback returns nil.
	if err := migrations.Rollback(context.Background(), db, dir, 1); err != nil {
		t.Fatalf("rollback 0006: %v", err)
	}
	var hasAllocationClaimColumn bool
	if err := db.QueryRow(`SELECT count(*) > 0 FROM information_schema.columns WHERE table_name = 'matches' AND column_name = 'allocation_id'`).Scan(&hasAllocationClaimColumn); err != nil {
		t.Fatal(err)
	}
	if hasAllocationClaimColumn {
		t.Fatal("0006 rollback did not drop matches.allocation_id")
	}

	if err := migrations.Rollback(context.Background(), db, dir, 4); err != nil {
		t.Fatalf("rollback remaining down to 0001: %v", err)
	}
	if tableExists("assignments") || tableExists("allocations") || tableExists("game_servers") {
		t.Fatal("rollback left later-migration tables behind")
	}

	if err := migrations.Rollback(context.Background(), db, dir, 1); err != nil {
		t.Fatalf("rollback 0001: %v", err)
	}
	if tableExists("identities") || tableExists("matches") {
		t.Fatal("0001 rollback did not drop its own tables")
	}
	var remaining int
	if err := db.QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&remaining); err != nil {
		t.Fatal(err)
	}
	if remaining != 0 {
		t.Fatalf("expected schema_migrations empty after full rollback, got %d rows", remaining)
	}

	// Reapplying from a fully rolled-back state must reach the same schema,
	// proving down files don't leave orphaned state that trips a forward
	// re-run (e.g. a constraint or index Apply then tries to recreate).
	if err := migrations.Apply(context.Background(), db, dir); err != nil {
		t.Fatalf("reapply after full rollback: %v", err)
	}
	if !tableExists("assignments") || !tableExists("allocations") {
		t.Fatal("reapply after rollback did not recreate the schema")
	}
}