CosmicClash/server/allocator/allocator_integration_test.go

//go:build integration

package allocator

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"testing"
	"time"

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

func TestRealAllocatorWorkerReconcilesAgonesAllocationAndBindsMatch(t *testing.T) {
	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.Fatal(err)
	}
	defer db.Close()
	ctx := context.Background()
	if err := db.PingContext(ctx); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(ctx, `DROP SCHEMA public CASCADE; CREATE SCHEMA public`); err != nil {
		t.Fatal(err)
	}
	if err := migrations.Apply(ctx, db, filepath.Join("..", "migrations")); err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC().Truncate(time.Microsecond)
	for index, player := range []string{"allocator-worker-a", "allocator-worker-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("allocator-worker-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 ('allocator-worker-match', 'casual', 'ALLOCATING', 'EU', 1)`); err != nil {
		t.Fatal(err)
	}
	for index, player := range []string{"allocator-worker-a", "allocator-worker-b"} {
		if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('allocator-worker-match', $1, $2, $3, $4)`, player, fmt.Sprintf("allocator-worker-ticket-%d", index), index, index); err != nil {
			t.Fatal(err)
		}
	}

	var allocationCalls int
	provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method == http.MethodGet {
			_, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"agones-ready-1","labels":{"cosmic-clash.io/region":"EU","cosmic-clash.io/build":"build-1","cosmic-clash.io/protocol":"1","cosmic-clash.io/transport":"enet"}},"status":{"state":"Ready"}}]}`))
			return
		}
		if r.Method != http.MethodPost {
			t.Fatalf("provider method = %s", r.Method)
		}
		allocationCalls++
		var body map[string]any
		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
			t.Fatal(err)
		}
		if body["kind"] != "GameServerAllocation" {
			t.Fatalf("provider body kind = %v", body["kind"])
		}
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"status":{"state":"Allocated","gameServerName":"agones-ready-1","address":"127.0.0.1","ports":[{"name":"default","port":7777}]}}`))
	}))
	defer provider.Close()

	agonesClient := agones.Client{BaseURL: provider.URL, Namespace: "games", HTTP: provider.Client()}
	ready, err := agonesClient.ListReadyServers(ctx)
	if err != nil || len(ready) != 1 {
		t.Fatalf("ready projection = %+v, err=%v", ready, err)
	}
	if err := store.RegisterReadyServer(ctx, db, ready[0], now); err != nil {
		t.Fatal(err)
	}
	worker := Worker{
		Claims: store.AllocatingMatchClaims{DB: db, Transport: "enet"},
		Service: Service{Provider: agonesClient, Durable: store.AllocationRegistry{DB: db}, Now: func() time.Time { return now }},
		Now: func() time.Time { return now },
	}
	processed, err := worker.RunOnce(ctx)
	if err != nil || !processed || allocationCalls != 1 {
		t.Fatalf("worker processed=%t err=%v provider calls=%d", processed, err, allocationCalls)
	}
	var serverID, matchState, ticketState string
	if err := db.QueryRowContext(ctx, `SELECT server_id, state FROM matches WHERE match_id = 'allocator-worker-match'`).Scan(&serverID, &matchState); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRowContext(ctx, `SELECT state FROM queue_tickets WHERE ticket_id = 'allocator-worker-ticket-0'`).Scan(&ticketState); err != nil {
		t.Fatal(err)
	}
	if serverID != "agones-ready-1" || matchState != "ALLOCATING" || ticketState != "ALLOCATING" {
		t.Fatalf("durable lifecycle server=%q match=%q ticket=%q", serverID, matchState, ticketState)
	}
	var recorded int
	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM allocations WHERE allocation_id = 'allocation-allocator-worker-match' AND server_id = 'agones-ready-1' AND state = 'ALLOCATED'`).Scan(&recorded); err != nil || recorded != 1 {
		t.Fatalf("recorded allocations=%d err=%v", recorded, err)
	}
}

// The root blocker: the worker bound the provider allocation and stopped.
// Service.PublishRoster and store.SaveVerifiedAssignmentRoster both existed but
// had no non-test callers, so nothing in production ever wrote the assignments
// table. The allocated supervisor fetches a non-empty roster before launching
// the game child, so every real allocation died at that fetch and no match
// could reach ASSIGNMENT_READY or accept a player.
//
// This drives the real worker and asserts against the durable tables. It never
// seeds the assignments table, which is exactly how the existing tests missed
// the missing hand-off.
func TestRealAllocatorWorkerPublishesSignedAssignmentRoster(t *testing.T) {
	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.Fatal(err)
	}
	defer db.Close()
	ctx := context.Background()
	if _, err := db.ExecContext(ctx, `DROP SCHEMA public CASCADE; CREATE SCHEMA public`); err != nil {
		t.Fatal(err)
	}
	if err := migrations.Apply(ctx, db, filepath.Join("..", "migrations")); err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC().Truncate(time.Microsecond)
	players := []string{"roster-worker-a", "roster-worker-b"}
	for index, player := range players {
		if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $2)`, player, "steam-"+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("roster-worker-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 ('roster-worker-match', 'casual', 'ALLOCATING', 'EU', 1)`); err != nil {
		t.Fatal(err)
	}
	for index, player := range players {
		if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('roster-worker-match', $1, $2, $3, $4)`, player, fmt.Sprintf("roster-worker-ticket-%d", index), index*3, index); err != nil {
			t.Fatal(err)
		}
	}

	provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method == http.MethodGet {
			_, _ = w.Write([]byte(`{"items":[{"metadata":{"name":"roster-ready-1","labels":{"cosmic-clash.io/region":"EU","cosmic-clash.io/build":"build-1","cosmic-clash.io/protocol":"1","cosmic-clash.io/transport":"enet"}},"status":{"state":"Ready"}}]}`))
			return
		}
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"status":{"state":"Allocated","gameServerName":"roster-ready-1","address":"127.0.0.1","ports":[{"name":"default","port":7777}]}}`))
	}))
	defer provider.Close()

	agonesClient := agones.Client{BaseURL: provider.URL, Namespace: "games", HTTP: provider.Client()}
	ready, err := agonesClient.ListReadyServers(ctx)
	if err != nil || len(ready) != 1 {
		t.Fatalf("ready projection = %+v err=%v", ready, err)
	}
	if err := store.RegisterReadyServer(ctx, db, ready[0], now); err != nil {
		t.Fatal(err)
	}

	// Two keys, signing with the newer: proves the rotation set is threaded
	// through signing and the persistence boundary's re-verification.
	keys := JoinSigningKeys{
		ActiveKeyID: "key-new",
		Keys:        map[string][]byte{"key-old": []byte("retired-key"), "key-new": []byte("active-key")},
	}
	worker := Worker{
		Claims:  store.AllocatingMatchClaims{DB: db, Transport: "enet"},
		Service: Service{Provider: agonesClient, Durable: store.AllocationRegistry{DB: db}, Roster: store.PostgresRosterStore{DB: db}, Now: func() time.Time { return now }},
		Now:     func() time.Time { return now },
		Roster:  store.AssignmentRosters{DB: db},
		Keys:    keys,
	}
	processed, err := worker.RunOnce(ctx)
	if err != nil || !processed {
		t.Fatalf("worker processed=%t err=%v", processed, err)
	}

	// One assignment row per participant, which is precisely what the
	// ASSIGNMENT_READY transition and the supervisor's roster fetch require.
	var assignments int
	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM assignments WHERE match_id = 'roster-worker-match'`).Scan(&assignments); err != nil {
		t.Fatal(err)
	}
	if assignments != len(players) {
		t.Fatalf("assignments = %d, want %d; the allocator did not publish the roster", assignments, len(players))
	}

	// The supervisor's own read path must return a usable roster.
	roster, err := store.GetAssignmentRoster(ctx, db, "roster-worker-match", "roster-ready-1", now)
	if err != nil {
		t.Fatalf("supervisor roster fetch: %v", err)
	}
	if len(roster) != len(players) {
		t.Fatalf("supervisor roster has %d entries, want %d", len(roster), len(players))
	}
	verify := domain.VerifyJoinAuthorisationHMAC(keys.Keys)
	seenSlots := map[int]bool{}
	for _, encoded := range roster {
		var signed domain.SignedJoinAuthorisation
		if err := json.Unmarshal(encoded, &signed); err != nil {
			t.Fatalf("decode roster entry: %v", err)
		}
		if signed.Authorisation.KeyID != "key-new" {
			t.Fatalf("entry signed with %q, want the active key", signed.Authorisation.KeyID)
		}
		if !verify(domain.JoinAuthorisationBytes(signed.Authorisation), signed.Signature) {
			t.Fatalf("roster entry for %s does not verify", signed.Authorisation.PlayerID)
		}
		if signed.Authorisation.MatchID != "roster-worker-match" || signed.Authorisation.ServerID != "roster-ready-1" {
			t.Fatalf("roster entry bound to the wrong match/server: %+v", signed.Authorisation)
		}
		seenSlots[signed.Authorisation.Slot] = true
	}
	if len(seenSlots) != len(players) {
		t.Fatalf("roster slots collided: %v", seenSlots)
	}

	// Republishing must be idempotent: a worker that crashed after binding but
	// before publishing retries this same path.
	allocation, recorded, err := store.AllocatingMatchClaims{DB: db, Transport: "enet"}.FindProviderAllocation(ctx, domain.AllocationRequest{
		AllocationID: "allocation-roster-worker-match", MatchID: "roster-worker-match",
		Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet",
	})
	if err != nil || !recorded {
		t.Fatalf("recover allocation: recorded=%t err=%v", recorded, err)
	}
	if allocation.Endpoint == "" {
		t.Fatal("the recovered allocation lost its endpoint, so a crashed worker could never republish")
	}
	if err := worker.publishAssignmentRoster(ctx, allocation); err != nil {
		t.Fatalf("republish: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM assignments WHERE match_id = 'roster-worker-match'`).Scan(&assignments); err != nil {
		t.Fatal(err)
	}
	if assignments != len(players) {
		t.Fatalf("republish duplicated assignments: %d", assignments)
	}
}