//go:build integration package supervisor import ( "context" "database/sql" "encoding/base64" "fmt" "net/http" "net/http/httptest" "os" "path/filepath" "strings" "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" ) type recordingRegistrar struct { delegate api.ServerRegistrar err error } func (r *recordingRegistrar) RegisterServer(ctx context.Context, binding domain.WorkloadBinding, protocol int, assignmentReady bool, idempotencyKey string, now time.Time) error { r.err = r.delegate.RegisterServer(ctx, binding, protocol, assignmentReady, idempotencyKey, now) return r.err } 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: base64.RawURLEncoding.EncodeToString([]byte(fmt.Sprintf(`{"authorisation":{"match_id":"supervisor-live-match","server_id":"supervisor-live-server","player_id":%q,"expires_at":%q},"signature":"sig"}`, player, now.Add(time.Hour).Format(time.RFC3339)))), 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() rosterPath := filepath.Join(t.TempDir(), "join-roster.json") registrar := &recordingRegistrar{delegate: api.ServerRegistrarFromStore(db)} service := &api.Service{ServerRegistrar: registrar, WorkloadVerify: api.WorkloadVerifierFromSignedToken(secret, db), Roster: func(ctx context.Context, binding domain.WorkloadBinding, at time.Time) ([][]byte, error) { return store.GetAssignmentRoster(ctx, db, binding.MatchID, binding.ServerID, at) }, 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, RosterPath: rosterPath, }) if err != nil { t.Fatal(err) } if err := supervisor.Start(ctx); err != nil { var matchState, matchServerID, allocationID, ticketState string _ = db.QueryRowContext(ctx, `SELECT state, server_id, allocation_id FROM matches WHERE match_id = 'supervisor-live-match'`).Scan(&matchState, &matchServerID, &allocationID) _ = db.QueryRowContext(ctx, `SELECT state FROM queue_tickets WHERE ticket_id = 'supervisor-live-ticket-0'`).Scan(&ticketState) t.Fatalf("start supervisor: %v (registration error=%v; match state=%q server=%q allocation=%q ticket=%q)", err, registrar.err, matchState, matchServerID, allocationID, ticketState) } if err := supervisor.Wait(); err != nil { t.Fatal(err) } roster, err := os.ReadFile(rosterPath) if err != nil || !strings.Contains(string(roster), "supervisor-live-a") || !strings.Contains(string(roster), "supervisor-live-b") { t.Fatalf("materialized live roster=%q err=%v", roster, 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(®istrationCount); err != nil || registrationCount != 2 { t.Fatalf("registration idempotency rows=%d err=%v", registrationCount, err) } }