//go:build integration package api import ( "bufio" "context" "database/sql" "encoding/json" "io" "net" "net/http/httptest" "os" "path/filepath" "strings" "testing" "time" "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 TestResultOutboxFanoutReachesAnAuthenticatedWebSocket(t *testing.T) { db := openIntegrationPostgres(t) now := time.Now().UTC().Truncate(time.Microsecond) ctx := context.Background() playerID := "result-fanout-player" if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $1)`, playerID); err != nil { t.Fatal(err) } sessions := store.PostgresSessions{DB: db} session, token, err := sessions.Issue(ctx, playerID, time.Hour, now) if err != nil { t.Fatalf("issue session: %v", err) } if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id, revision) VALUES ('result-fanout-match', 'casual', 'COMPLETED', 'EU', 1, 'result-fanout-server', 4)`); 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 ('result-fanout-ticket', $1, 'casual', 'COMPLETED', 'build-1', 1, $2, $3)`, playerID, now, now.Add(time.Hour)); err != nil { t.Fatal(err) } if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('result-fanout-match', $1, 'result-fanout-ticket', 0, 0)`, playerID); err != nil { t.Fatal(err) } payload := []byte(`{"match_id":"result-fanout-match","result_nonce":"fanout-result-nonce","score":{"team_0":1,"team_1":0},"integrity_state":"CERTIFIED"}`) if _, err := db.ExecContext(ctx, `INSERT INTO outbox (event_id, aggregate_type, aggregate_id, revision, event_type, payload, created_at) VALUES ('result-fanout-event', 'match', 'result-fanout-match', 5, 'match_completed', $1, $2)`, payload, now); err != nil { t.Fatal(err) } service := &Service{SessionBackend: sessions} server := httptest.NewServer(service.Handler()) defer server.Close() connection, err := net.Dial("tcp", strings.TrimPrefix(server.URL, "http://")) if err != nil { t.Fatal(err) } defer connection.Close() if _, err := io.WriteString(connection, "GET /v1/events HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nAuthorization: Bearer "+session.SessionID+":"+token+"\r\n\r\n"); err != nil { t.Fatal(err) } reader := bufio.NewReader(connection) status, err := reader.ReadString('\n') if err != nil || !strings.Contains(status, "101 Switching Protocols") { t.Fatalf("websocket handshake status=%q err=%v", status, err) } for { line, readErr := reader.ReadString('\n') if readErr != nil { t.Fatal(readErr) } if line == "\r\n" { break } } if err := deliverResultOutboxEvent(ctx, db, store.OutboxEvent{EventID: "result-fanout-event", EventType: "match_completed", AggregateID: "result-fanout-match", Revision: 5, CreatedAt: now, Payload: payload}, service); err != nil { t.Fatalf("deliver result event: %v", err) } frame, err := readServerWebSocketFrame(reader) if err != nil { t.Fatalf("read result event: %v", err) } var event ControlPlaneEvent if err := json.Unmarshal(frame, &event); err != nil { t.Fatal(err) } if event.Event != "state_changed" || event.Revision != 5 || event.ResourceID != "result-fanout-match" || event.State != "COMPLETED" || event.MatchID != "result-fanout-match" { t.Fatalf("unexpected result fan-out event: %+v", event) } } // This binary is deliberately opt-in, matching store's integration suite: 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() }) migrationDir := os.Getenv("COSMIC_CLASH_MIGRATIONS_DIR") if migrationDir == "" { migrationDir = filepath.Join("..", "migrations") } if err := migrations.Apply(ctx, db, migrationDir); err != nil { t.Fatalf("apply migrations: %v", err) } return db } // seedRealAllocation claims a real ready server and allocation row, exactly // the durable state a signed workload token's allocation_id must resolve // against (see store.AllocationBindingByAllocationID). func seedRealAllocation(t *testing.T, db *sql.DB, allocationID, matchID string, now time.Time) domain.Allocation { t.Helper() ctx := context.Background() serverID := "server-" + allocationID if err := store.RegisterReadyServer(ctx, db, domain.ReadyServer{ServerID: serverID, Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerReady}, now); err != nil { t.Fatalf("register ready server: %v", err) } allocation, err := store.ClaimAllocation(ctx, db, domain.AllocationRequest{AllocationID: allocationID, MatchID: matchID, Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet"}, now) if err != nil { t.Fatalf("claim allocation: %v", err) } return allocation } // TestWorkloadVerifierFromSignedTokenAcceptsARealAllocation proves the full // wired path: a token issued by workload.IssueSignedWorkloadToken naming only // a real allocation_id verifies successfully through // WorkloadVerifierFromSignedToken and returns a binding whose match_id/ // server_id came from the durable allocation record (the token itself never // carries them -- see signed_token.go), matching what serverMutation // actually checks (ServerID, MatchID). This is the "wired, working" // counterpart to cmd/control-plane's // TestServerRoutesRequireWorkloadVerifyToBeWired, which pins the // unconfigured-503 case. func TestWorkloadVerifierFromSignedTokenAcceptsARealAllocation(t *testing.T) { db := openIntegrationPostgres(t) now := time.Now().UTC() allocation := seedRealAllocation(t, db, "alloc-verify-1", "match-verify-1", now) secret := []byte("integration-test-secret") token, err := workload.IssueSignedWorkloadToken(secret, allocation.AllocationID, now, time.Minute) if err != nil { t.Fatalf("issue token: %v", err) } verify := WorkloadVerifierFromSignedToken(secret, db) if verify == nil { t.Fatal("WorkloadVerifierFromSignedToken returned nil with a real secret and database") } binding, err := verify(token, now.Add(30*time.Second)) if err != nil { t.Fatalf("verify: %v", err) } if binding.ServerID != allocation.ServerID || binding.MatchID != allocation.MatchID || binding.AllocationID != allocation.AllocationID { t.Fatalf("unexpected binding: %+v, want server=%s match=%s allocation=%s", binding, allocation.ServerID, allocation.MatchID, allocation.AllocationID) } } // TestWorkloadVerifierFromSignedTokenRejectsAnUnknownAllocation proves the // durable lookup actually runs: a validly-signed, unexpired token whose // allocation was never recorded (e.g. simply fabricated) must still be // rejected. Signature and expiry checks alone are not enough. func TestWorkloadVerifierFromSignedTokenRejectsAnUnknownAllocation(t *testing.T) { db := openIntegrationPostgres(t) now := time.Now().UTC() secret := []byte("integration-test-secret") token, err := workload.IssueSignedWorkloadToken(secret, "alloc-never-recorded", now, time.Minute) if err != nil { t.Fatalf("issue token: %v", err) } verify := WorkloadVerifierFromSignedToken(secret, db) if _, err := verify(token, now.Add(time.Second)); err == nil { t.Fatal("expected rejection for a token naming an allocation that was never recorded") } } // TestWorkloadVerifierFromSignedTokenNeverTrustsCallerSuppliedBinding proves // the binding returned is entirely derived from the durable allocation row, // never from anything embedded in or inferable from the token: two distinct // allocations produce tokens that resolve to their own, and only their own, // match/server pairing. func TestWorkloadVerifierFromSignedTokenNeverTrustsCallerSuppliedBinding(t *testing.T) { db := openIntegrationPostgres(t) now := time.Now().UTC() first := seedRealAllocation(t, db, "alloc-verify-2a", "match-verify-2a", now) second := seedRealAllocation(t, db, "alloc-verify-2b", "match-verify-2b", now) secret := []byte("integration-test-secret") verify := WorkloadVerifierFromSignedToken(secret, db) firstToken, err := workload.IssueSignedWorkloadToken(secret, first.AllocationID, now, time.Minute) if err != nil { t.Fatalf("issue first token: %v", err) } firstBinding, err := verify(firstToken, now.Add(time.Second)) if err != nil { t.Fatalf("verify first: %v", err) } if firstBinding.MatchID != first.MatchID || firstBinding.ServerID != first.ServerID { t.Fatalf("first binding %+v resolved to the wrong allocation", firstBinding) } secondToken, err := workload.IssueSignedWorkloadToken(secret, second.AllocationID, now, time.Minute) if err != nil { t.Fatalf("issue second token: %v", err) } secondBinding, err := verify(secondToken, now.Add(time.Second)) if err != nil { t.Fatalf("verify second: %v", err) } if secondBinding.MatchID != second.MatchID || secondBinding.ServerID != second.ServerID { t.Fatalf("second binding %+v resolved to the wrong allocation", secondBinding) } if secondBinding.MatchID == firstBinding.MatchID || secondBinding.ServerID == firstBinding.ServerID { t.Fatalf("distinct allocations resolved to the same binding: %+v vs %+v", firstBinding, secondBinding) } } // TestWorkloadVerifierFromSignedTokenClosesTheDefaultUnwiredGap proves the // 503-by-default gap pinned by // cmd/control-plane.TestServerRoutesRequireWorkloadVerifyToBeWired is // actually closed once a secret and database are wired: a Service built the // same way newAPIHandler builds one now accepts a validly-issued token for a // real allocation, through the exact Service.WorkloadVerify field the HTTP // handler calls. (serverMutation's deeper match-state transition -- // requiring the match to already be ALLOCATING -- is exercised separately by // the store package's own allocation/match integration tests; this test's // job is only the WorkloadVerify boundary itself.) func TestWorkloadVerifierFromSignedTokenClosesTheDefaultUnwiredGap(t *testing.T) { db := openIntegrationPostgres(t) now := time.Now().UTC() allocation := seedRealAllocation(t, db, "alloc-verify-3", "match-verify-3", now) secret := []byte("integration-test-secret") token, err := workload.IssueSignedWorkloadToken(secret, allocation.AllocationID, now, time.Minute) if err != nil { t.Fatalf("issue token: %v", err) } svc := &Service{ ServerRegistrar: ServerRegistrarFromStore(db), WorkloadVerify: WorkloadVerifierFromSignedToken(secret, db), Now: func() time.Time { return now.Add(time.Second) }, } binding, err := svc.WorkloadVerify(token, now.Add(time.Second)) if err != nil { t.Fatalf("WorkloadVerify rejected a validly-issued token for a real allocation: %v", err) } if binding.ServerID != allocation.ServerID { t.Fatalf("binding.ServerID = %q, want %q", binding.ServerID, allocation.ServerID) } }