feat: add authenticated matchmaking event stream

2026-09-11 08:23:45 +00:00 · 2026-08-31 23:05:04 +01:00
parent d77563147c
commit d258f17852
4 changed files with 378 additions and 2 deletions
@@ -1226,8 +1226,8 @@ the local/CI/community transport, not a silent production fallback.

 | # | Task | Acceptance |
 |---|---|---|
-| 8.39 `[D:8.3,8.14,8.17]` | **IN PROGRESS.** `MatchmakingState` now projects queue → proposal → allocation/process-ready/assignment-ready/connect/live plus terminal failure states; autoload `ControlPlaneClient` provides authenticated queue create/recovery/heartbeat/cancel and proposal response requests with idempotency/revision headers; `matchmaking.tscn`/`matchmaking.gd` expose the state and authoritative actions from the main menu; authenticated proposal recovery now reconciles missed proposal events and expires them at read time | `test_matchmaking_state.gd`, `test_control_plane_client.gd`, `test_matchmaking_ui.gd` and `TestProposalRecoveryIsParticipantScopedAndExpiresAtReadBoundary` reject stale/gapped/conflicting updates, validate endpoint/token/payload normalization, preserve idempotent duplicates, and guarantee visible phase/terminal copy; server-pushed allocation events, wait/latency explanations and Godot runtime verification remain |
-| 8.40 `[D:8.3,8.14]` | **IN PROGRESS.** Pure Go revisioned replica reducer rejects gaps for REST resync, makes duplicate/out-of-order events idempotent, and resumes from the authoritative snapshot revision; authenticated queue-ticket recovery now has an owner-checked REST read; Godot client projection persists non-secret ticket/proposal state, forces authoritative recovery after restart, and can replay a lost queue-create response with the original ticket/idempotency key | `server/domain/sync.go`, `server/api/service.go`, `service_test.go`, `matchmaking_state.gd` and `control_plane_client.gd` cover gap, snapshot, replay, same-revision conflict, owner-only ticket recovery, expired-ticket terminal handling, malformed restart snapshots and API-level duplicate-create replay/conflict; authenticated WebSocket transport and live Godot verification remain |
+| 8.39 `[D:8.3,8.14,8.17]` | **IN PROGRESS.** `MatchmakingState` now projects queue → proposal → allocation/process-ready/assignment-ready/connect/live plus terminal failure states; autoload `ControlPlaneClient` provides authenticated queue create/recovery/heartbeat/cancel and proposal response requests with idempotency/revision headers; `matchmaking.tscn`/`matchmaking.gd` expose the state and authoritative actions from the main menu; authenticated proposal recovery now reconciles missed proposal events and expires them at read time; the Go API now provides targeted authenticated revisioned event publication | `test_matchmaking_state.gd`, `test_control_plane_client.gd`, `test_matchmaking_ui.gd`, `TestProposalRecoveryIsParticipantScopedAndExpiresAtReadBoundary` and `TestAuthenticatedWebSocketDeliversOnlyTargetedRevisionedEvents` reject stale/gapped/conflicting updates, validate endpoint/token/payload normalization, preserve idempotent duplicates, and guarantee visible phase/terminal copy; wiring durable allocation events into the outbox, wait/latency explanations and Godot runtime verification remain |
+| 8.40 `[D:8.3,8.14]` | **IN PROGRESS.** Pure Go revisioned replica reducer rejects gaps for REST resync, makes duplicate/out-of-order events idempotent, and resumes from the authoritative snapshot revision; authenticated queue-ticket recovery now has an owner-checked REST read; Godot client projection persists non-secret ticket/proposal state, forces authoritative recovery after restart, and can replay a lost queue-create response with the original ticket/idempotency key; the Go API now exposes an authenticated `/v1/events` WebSocket with bounded per-player queues, strict upgrade validation and REST-resync-safe slow-client failure | `server/domain/sync.go`, `server/api/events.go`, `server/api/service.go`, `service_test.go`, `matchmaking_state.gd` and `control_plane_client.gd` cover gap, snapshot, replay, same-revision conflict, owner-only ticket recovery, expired-ticket terminal handling, malformed restart snapshots, API-level duplicate-create replay/conflict, authenticated handshake/key rejection and targeted event delivery; Godot WebSocket client wiring, durable outbox fan-out, reconnect/resync orchestration and live Godot verification remain |
 | 8.41 `[D:7.8,8.9,8.31,8.40]` | **IN PROGRESS.** Authenticated `GET /v1/assignments/{matchId}` now exposes only a validated, player-scoped assignment view; Godot `AssignmentState`/`ControlPlaneClient.fetch_assignment()` bind the response to the authenticated player, recheck expiry, preserve explicit transport/slot/join authorisation and do not connect before assignment-ready | `server/api/service.go`, `service_test.go`, `assignment_state.gd` and `test_assignment_state.gd` cover participant/identity/expiry/shape/transport boundaries and assignment recovery; signed manifest-to-player persistence, SDR relay-ticket installation, `hello` join-authorisation wiring, fencing integration and live Godot verification remain |
 | 8.42 `[D:8.22,8.23,8.24,8.40]` | **IN PROGRESS.** `RankedProfileState` and `ControlPlaneClient.fetch_ranked_profile()` expose the backend-authoritative rating/RD/volatility/games/tier/provisional/season view; matchmaking UI displays provisional/tier status without client-side rating math | `test_control_plane_client.gd` validates profile shape, numeric safety and provisional display; ranked profile fetch/display, committed revision after reconnect, abandon status and season countdown remain dependent on live auth/backend events and Godot runtime verification |
 | 8.43 `[D:8.39,8.40,8.41]` | **IN PROGRESS.** Matchmaking client now distinguishes expired queue recovery, session expiry, missing records and retryable control-plane outages; a 401 clears the in-memory token, emits `session_expired` and disables retry until a new session is configured; terminal messages remain visible and active searches are not falsely failed on transient errors | `MatchmakingState` and `ControlPlaneClient` tests cover explicit expiry and the existing terminal/retry-safe state paths; decline, version mismatch, regional outage retry UI, failed reconnect, duplicate-action recovery and live Godot verification remain |
@@ -0,0 +1,276 @@
+package api
+
+import (
+	"bufio"
+	"crypto/sha1"
+	"encoding/base64"
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+)
+
+const (
+	webSocketGUID      = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
+	maxWebSocketFrame  = 64 << 10
+	eventQueueCapacity = 32
+)
+
+// ControlPlaneEvent is the server-to-client envelope defined by the v1
+// WebSocket contract. PlayerID is routing metadata and is never serialized.
+type ControlPlaneEvent struct {
+	Event      string    `json:"event"`
+	Revision   uint64    `json:"revision"`
+	ResourceID string    `json:"resource_id"`
+	OccurredAt time.Time `json:"occurred_at"`
+	State      string    `json:"state,omitempty"`
+	Code       string    `json:"code,omitempty"`
+	MatchID    string    `json:"match_id,omitempty"`
+	ServerID   string    `json:"server_id,omitempty"`
+	PlayerID   string    `json:"-"`
+}
+
+type eventSubscriber struct {
+	playerID string
+	queue    chan []byte
+}
+
+type eventHub struct {
+	mu          sync.Mutex
+	subscribers map[*eventSubscriber]struct{}
+}
+
+func newEventHub() *eventHub {
+	return &eventHub{subscribers: make(map[*eventSubscriber]struct{})}
+}
+
+func (h *eventHub) subscribe(playerID string) *eventSubscriber {
+	subscriber := &eventSubscriber{playerID: playerID, queue: make(chan []byte, eventQueueCapacity)}
+	h.mu.Lock()
+	h.subscribers[subscriber] = struct{}{}
+	h.mu.Unlock()
+	return subscriber
+}
+
+func (h *eventHub) unsubscribe(subscriber *eventSubscriber) {
+	h.mu.Lock()
+	delete(h.subscribers, subscriber)
+	close(subscriber.queue)
+	h.mu.Unlock()
+}
+
+func (h *eventHub) publish(event ControlPlaneEvent) error {
+	if event.PlayerID == "" || event.Event == "" || event.ResourceID == "" || event.OccurredAt.IsZero() {
+		return errors.New("invalid control-plane event")
+	}
+	payload, err := json.Marshal(event)
+	if err != nil {
+		return err
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for subscriber := range h.subscribers {
+		if subscriber.playerID != event.PlayerID {
+			continue
+		}
+		select {
+		case subscriber.queue <- payload:
+		default:
+			// A slow client must not block state publication for other clients.
+			// Closing its queue makes the connection fail closed and recover via
+			// REST resync rather than silently dropping an unbounded history.
+			delete(h.subscribers, subscriber)
+			close(subscriber.queue)
+		}
+	}
+	return nil
+}
+
+func (s *Service) controlPlaneEvent(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
+		return
+	}
+	if !isWebSocketUpgrade(r) || !validWebSocketKey(r.Header.Get("Sec-WebSocket-Key")) {
+		writeError(w, http.StatusBadRequest, "invalid_websocket_upgrade")
+		return
+	}
+	playerID, ok := s.authenticate(w, r)
+	if !ok {
+		return
+	}
+	hijacker, ok := w.(http.Hijacker)
+	if !ok {
+		writeError(w, http.StatusNotImplemented, "websocket_unavailable")
+		return
+	}
+	connection, buffered, err := hijacker.Hijack()
+	if err != nil {
+		return
+	}
+	defer connection.Close()
+	accept := websocketAccept(r.Header.Get("Sec-WebSocket-Key"))
+	if _, err := fmt.Fprintf(buffered, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: %s\r\n\r\n", accept); err != nil {
+		return
+	}
+	if err := buffered.Flush(); err != nil {
+		return
+	}
+	subscriber := s.getEventHub().subscribe(playerID)
+	defer s.getEventHub().unsubscribe(subscriber)
+
+	var writeMu sync.Mutex
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		readWebSocketFrames(connection, &writeMu)
+	}()
+	for {
+		select {
+		case payload, open := <-subscriber.queue:
+			if !open {
+				return
+			}
+			writeMu.Lock()
+			err := writeWebSocketFrame(connection, 0x1, payload)
+			writeMu.Unlock()
+			if err != nil {
+				return
+			}
+		case <-done:
+			return
+		}
+	}
+}
+
+func (s *Service) getEventHub() *eventHub {
+	s.eventsMu.Lock()
+	defer s.eventsMu.Unlock()
+	if s.events == nil {
+		s.events = newEventHub()
+	}
+	return s.events
+}
+
+// PublishControlPlaneEvent routes an already-authorized event to the matching
+// authenticated player connection. Durable callers should publish from their
+// outbox after commit; this in-memory hub is deliberately non-authoritative.
+func (s *Service) PublishControlPlaneEvent(event ControlPlaneEvent) error {
+	return s.getEventHub().publish(event)
+}
+
+func isWebSocketUpgrade(r *http.Request) bool {
+	return strings.EqualFold(r.Header.Get("Upgrade"), "websocket") && headerContainsToken(r.Header.Values("Connection"), "upgrade")
+}
+
+func headerContainsToken(values []string, wanted string) bool {
+	for _, value := range values {
+		for _, token := range strings.Split(value, ",") {
+			if strings.EqualFold(strings.TrimSpace(token), wanted) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func websocketAccept(key string) string {
+	digest := sha1.Sum([]byte(key + webSocketGUID))
+	return base64.StdEncoding.EncodeToString(digest[:])
+}
+
+func validWebSocketKey(key string) bool {
+	decoded, err := base64.StdEncoding.DecodeString(key)
+	return err == nil && len(decoded) == 16
+}
+
+func readWebSocketFrames(connection net.Conn, writeMu *sync.Mutex) {
+	reader := bufio.NewReader(connection)
+	for {
+		opcode, _, err := readWebSocketFrame(reader)
+		if err != nil || opcode == 0x8 {
+			return
+		}
+		if opcode == 0x9 {
+			writeMu.Lock()
+			_ = writeWebSocketFrame(connection, 0xA, nil)
+			writeMu.Unlock()
+		}
+	}
+}
+
+func readWebSocketFrame(reader *bufio.Reader) (byte, []byte, error) {
+	first, err := reader.ReadByte()
+	if err != nil {
+		return 0, nil, err
+	}
+	second, err := reader.ReadByte()
+	if err != nil {
+		return 0, nil, err
+	}
+	if first&0x70 != 0 || first&0x80 == 0 {
+		return 0, nil, errors.New("unsupported websocket frame")
+	}
+	if second&0x80 == 0 {
+		return 0, nil, errors.New("unmasked websocket frame")
+	}
+	length := int64(second & 0x7f)
+	if length == 126 {
+		var extended uint16
+		if err := binary.Read(reader, binary.BigEndian, &extended); err != nil {
+			return 0, nil, err
+		}
+		length = int64(extended)
+	} else if length == 127 {
+		var extended uint64
+		if err := binary.Read(reader, binary.BigEndian, &extended); err != nil || extended > maxWebSocketFrame {
+			return 0, nil, errors.New("websocket frame too large")
+		}
+		length = int64(extended)
+	}
+	if length > maxWebSocketFrame {
+		return 0, nil, errors.New("websocket frame too large")
+	}
+	var mask [4]byte
+	if _, err := io.ReadFull(reader, mask[:]); err != nil {
+		return 0, nil, err
+	}
+	payload := make([]byte, length)
+	if _, err := io.ReadFull(reader, payload); err != nil {
+		return 0, nil, err
+	}
+	for i := range payload {
+		payload[i] ^= mask[i%4]
+	}
+	return first & 0x0f, payload, nil
+}
+
+func writeWebSocketFrame(connection net.Conn, opcode byte, payload []byte) error {
+	if len(payload) > maxWebSocketFrame {
+		return errors.New("websocket frame too large")
+	}
+	header := []byte{0x80 | opcode}
+	switch {
+	case len(payload) < 126:
+		header = append(header, byte(len(payload)))
+	case len(payload) <= 65535:
+		header = append(header, 126, 0, 0)
+		binary.BigEndian.PutUint16(header[len(header)-2:], uint16(len(payload)))
+	default:
+		header = append(header, 127)
+		var extended [8]byte
+		binary.BigEndian.PutUint64(extended[:], uint64(len(payload)))
+		header = append(header, extended[:]...)
+	}
+	if _, err := connection.Write(header); err != nil {
+		return err
+	}
+	_, err := connection.Write(payload)
+	return err
+}
@@ -72,6 +72,8 @@ type Service struct {
 	RankedProfiles map[string]domain.RankedProfile
 	TierPolicy     domain.TierPolicy
 	proposalMu     sync.Mutex
+	eventsMu       sync.Mutex
+	events         *eventHub
 }

 func (s *Service) Handler() http.Handler {
@@ -84,6 +86,7 @@ func (s *Service) Handler() http.Handler {
 	mux.HandleFunc("/v1/assignments/", s.assignment)
 	mux.HandleFunc("/v1/profile/ranked", s.rankedProfile)
 	mux.HandleFunc("/v1/probes/", s.probe)
+	mux.HandleFunc("/v1/events", s.controlPlaneEvent)
 	// The public contract is served below /api/v1. Keep the original /v1
 	// routes for the Godot client while exposing the documented names.
 	mux.HandleFunc("/api/v1/session/steam", s.steamSession)
@@ -92,6 +95,7 @@ func (s *Service) Handler() http.Handler {
 	mux.HandleFunc("/api/v1/queue/tickets/", s.contractQueueMutation)
 	mux.HandleFunc("/api/v1/proposals/", s.contractProposalMutation)
 	mux.HandleFunc("/api/v1/assignments/", s.contractAssignment)
+	mux.HandleFunc("/api/v1/events", s.controlPlaneEvent)
 	return mux
 }

@@ -1,8 +1,13 @@
 package api

 import (
+	"bufio"
 	"context"
+	"encoding/binary"
 	"encoding/json"
+	"errors"
+	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -161,6 +166,97 @@ func TestDocumentedContractRoutesAdaptToServiceAPI(t *testing.T) {
 	}
 }

+func TestAuthenticatedWebSocketDeliversOnlyTargetedRevisionedEvents(t *testing.T) {
+	service := &Service{SessionBackend: &sessionBackendSpy{}}
+	server := httptest.NewServer(service.Handler())
+	defer server.Close()
+	invalid, err := http.NewRequest(http.MethodGet, server.URL+"/v1/events", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	invalid.Header.Set("Upgrade", "websocket")
+	invalid.Header.Set("Connection", "Upgrade")
+	invalid.Header.Set("Sec-WebSocket-Key", "not-a-websocket-key")
+	invalid.Header.Set("Authorization", "Bearer session-1:token-1")
+	invalidResponse, err := server.Client().Do(invalid)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = invalidResponse.Body.Close()
+	if invalidResponse.StatusCode != http.StatusBadRequest {
+		t.Fatalf("invalid handshake status = %d", invalidResponse.StatusCode)
+	}
+	connection, err := net.Dial("tcp", strings.TrimPrefix(server.URL, "http://"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer connection.Close()
+	_, 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-1:token-1\r\n\r\n")
+	if err != nil {
+		t.Fatal(err)
+	}
+	reader := bufio.NewReader(connection)
+	status, err := reader.ReadString('\n')
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(status, "101 Switching Protocols") {
+		t.Fatalf("handshake status = %q", status)
+	}
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			t.Fatal(err)
+		}
+		if line == "\r\n" {
+			break
+		}
+	}
+	time.Sleep(10 * time.Millisecond)
+	if err := service.PublishControlPlaneEvent(ControlPlaneEvent{Event: "state_changed", Revision: 1, ResourceID: "ticket-1234567890123456", OccurredAt: time.Unix(1000, 0).UTC(), State: "QUEUED", PlayerID: "player-1"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := service.PublishControlPlaneEvent(ControlPlaneEvent{Event: "state_changed", Revision: 2, ResourceID: "ticket-1234567890123456", OccurredAt: time.Unix(1001, 0).UTC(), State: "PROPOSED", PlayerID: "player-2"}); err != nil {
+		t.Fatal(err)
+	}
+	first, err := readServerWebSocketFrame(reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var event ControlPlaneEvent
+	if err := json.Unmarshal(first, &event); err != nil {
+		t.Fatal(err)
+	}
+	if event.PlayerID != "" || event.Revision != 1 || event.ResourceID != "ticket-1234567890123456" || event.State != "QUEUED" {
+		t.Fatalf("event = %+v", event)
+	}
+}
+
+func readServerWebSocketFrame(reader *bufio.Reader) ([]byte, error) {
+	first, err := reader.ReadByte()
+	if err != nil {
+		return nil, err
+	}
+	second, err := reader.ReadByte()
+	if err != nil {
+		return nil, err
+	}
+	if first&0x0f != 0x1 || second&0x80 != 0 {
+		return nil, errors.New("unexpected server websocket frame")
+	}
+	length := int(second & 0x7f)
+	if length == 126 {
+		var extended uint16
+		if err := binary.Read(reader, binary.BigEndian, &extended); err != nil {
+			return nil, err
+		}
+		length = int(extended)
+	}
+	payload := make([]byte, length)
+	_, err = io.ReadFull(reader, payload)
+	return payload, err
+}
+
 func TestQueueAPIRejectsUnauthenticatedUnknownAndOversizedInput(t *testing.T) {
 	service := &Service{Sessions: domain.NewSessionStore(), Queue: domain.NewQueue(), Candidate: func(string, string) (domain.Candidate, error) { return domain.Candidate{}, nil }}
 	server := httptest.NewServer(service.Handler())