test(multiplayer): verify allocated supervisor registration

server/supervisor/supervisor_integration_test.go

@@ -0,0 +1,133 @@
+//go:build integration
+
+package supervisor
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/api"
+	"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/cosmic-clash/cosmic-clash/server/workload"
+	_ "github.com/jackc/pgx/v5/stdlib"
+)
+
+func TestRealSupervisorRegistersAllocatedServerThroughControlPlane(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 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.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{"supervisor-live-a", "supervisor-live-b"}
+	for index, player := range players {
+		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("supervisor-live-ticket-%d", index), player, now, now.Add(time.Hour)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version) VALUES ('supervisor-live-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 ('supervisor-live-match', $1, $2, $3, $4)`, player, fmt.Sprintf("supervisor-live-ticket-%d", index), index, index); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := store.RegisterReadyServer(ctx, db, domain.ReadyServer{ServerID: "supervisor-live-server", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady}, now); err != nil {
+		t.Fatal(err)
+	}
+	claim, found, err := store.ClaimAllocatingMatch(ctx, db, "enet", now)
+	if err != nil || !found {
+		t.Fatalf("claim allocating match found=%t err=%v", found, err)
+	}
+	request := claim.Request
+	allocation, err := store.ClaimAllocation(ctx, db, request, now)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := store.BindAllocatedMatch(ctx, db, allocation); err != nil {
+		t.Fatal(err)
+	}
+	for index, player := range players {
+		if err := store.SaveAssignment(ctx, db, store.DurableAssignment{
+			MatchID: "supervisor-live-match", PlayerID: player, AllocationID: request.AllocationID, ServerID: "supervisor-live-server", Slot: index,
+			Region: "EU", ClientBuild: "build-1", ProtocolVersion: 1, Transport: "enet", Endpoint: "127.0.0.1:7777",
+			JoinAuthorisation: "join-" + player, ManifestDigest: []byte{0, 1, 2, 3}, ExpiresAt: now.Add(time.Hour), Revision: 1,
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	secret := []byte("supervisor-live-workload-secret")
+	token, err := workload.IssueSignedWorkloadToken(secret, request.AllocationID, now, time.Hour)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sdk := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/gameserver":
+			_, _ = fmt.Fprintf(w, `{"object_meta":{"annotations":{"cosmic-clash.io/match-id":"supervisor-live-match","cosmic-clash.io/workload-token":%q}},"status":{"address":"127.0.0.1","ports":[{"name":"game","port":7777}]}}`, token)
+		case "/ready-probe", "/ready":
+			w.WriteHeader(http.StatusOK)
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	defer sdk.Close()
+	service := &api.Service{ServerRegistrar: api.ServerRegistrarFromStore(db), WorkloadVerify: api.WorkloadVerifierFromSignedToken(secret, db), Now: func() time.Time { return now }}
+	control := httptest.NewServer(service.Handler())
+	defer control.Close()
+	supervisor, err := New(Config{
+		Command: []string{"/bin/sh", "-c", "true"}, SDKBaseURL: sdk.URL, ReadyURL: sdk.URL + "/ready-probe", ControlPlaneURL: control.URL,
+		ServerID: "supervisor-live-server", ProtocolVersion: 1, ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+		ReadyTimeout: time.Second, PollInterval: time.Millisecond, AssignmentReadyAttempts: 1,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := supervisor.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if err := supervisor.Wait(); err != nil {
+		t.Fatal(err)
+	}
+	var matchState, ticketState string
+	if err := db.QueryRowContext(ctx, `SELECT state FROM matches WHERE match_id = 'supervisor-live-match'`).Scan(&matchState); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.QueryRowContext(ctx, `SELECT state FROM queue_tickets WHERE ticket_id = 'supervisor-live-ticket-0'`).Scan(&ticketState); err != nil {
+		t.Fatal(err)
+	}
+	if matchState != "ASSIGNMENT_READY" || ticketState != "ASSIGNMENT_READY" {
+		t.Fatalf("registration lifecycle match=%q ticket=%q", matchState, ticketState)
+	}
+	var registrationCount int
+	if err := db.QueryRowContext(ctx, `SELECT count(*) FROM idempotency_keys WHERE scope = 'server.register' AND idempotency_key LIKE 'supervisor-register-supervisor-live-server-supervisor-live-match-%'`).Scan(&registrationCount); err != nil || registrationCount != 2 {
+		t.Fatalf("registration idempotency rows=%d err=%v", registrationCount, err)
+	}
+}