CosmicClash/server/api/workload_verifier_integration_test.go

//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)
	}
}