CosmicClash/server/supervisor/supervisor_integration_test.go

//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 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{"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(&registrationCount); err != nil || registrationCount != 2 {
		t.Fatalf("registration idempotency rows=%d err=%v", registrationCount, err)
	}
}