test(multiplayer): verify result websocket fanout

server/api/workload_verifier_integration_test.go

@@ -3,10 +3,16 @@
 package api
 
 import (
+	"bufio"
 	"context"
 	"database/sql"
+	"encoding/json"
+	"io"
+	"net"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -17,6 +23,73 @@ import (
 	_ "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.