From 60fe2caf8fead10faa3c84efe726e9ec95283e7a Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Mon, 31 Aug 2026 23:12:20 +0100 Subject: [PATCH] feat: publish matchmaking state events --- multiplayer-todo.md | 2 +- server/api/events.go | 18 +++++++++++++++ server/api/service.go | 13 +++++++++-- server/api/service_test.go | 46 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 76 insertions(+), 3 deletions(-) diff --git a/multiplayer-todo.md b/multiplayer-todo.md index 51c6873e..7d9696e2 100644 --- a/multiplayer-todo.md +++ b/multiplayer-todo.md @@ -1226,7 +1226,7 @@ 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; the Go API now provides targeted authenticated revisioned event publication and the Godot client consumes state/proposal events | `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.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 publishes targeted authenticated revisioned queue/proposal/assignment events and the Godot client consumes state/proposal events | `test_matchmaking_state.gd`, `test_control_plane_client.gd`, `test_matchmaking_ui.gd`, `TestProposalRecoveryIsParticipantScopedAndExpiresAtReadBoundary`, `TestAuthenticatedWebSocketDeliversOnlyTargetedRevisionedEvents` and `TestStateChangingAPIActionsPublishTargetedEvents` 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/vocabulary validation and REST-resync-safe slow-client failure; Godot `ControlPlaneClient` can connect to the stream, consume validated events, automatically reconnect with bounded backoff, and trigger REST recovery on projection gaps and stream return | `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, targeted event delivery, exactly-once slow-subscriber closure and invalid-event rejection; durable outbox fan-out 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 | diff --git a/server/api/events.go b/server/api/events.go index 21e68db0..61a06918 100644 --- a/server/api/events.go +++ b/server/api/events.go @@ -14,6 +14,8 @@ import ( "strings" "sync" "time" + + "github.com/cosmic-clash/cosmic-clash/server/domain" ) const ( @@ -205,6 +207,22 @@ func (s *Service) PublishControlPlaneEvent(event ControlPlaneEvent) error { return s.getEventHub().publish(event) } +func (s *Service) publishTicketEvent(ticket domain.QueueTicket, now time.Time) { + _ = s.PublishControlPlaneEvent(ControlPlaneEvent{ + Event: "state_changed", Revision: ticket.Revision, ResourceID: ticket.TicketID, + OccurredAt: now, State: string(ticket.State), PlayerID: ticket.PlayerID, + }) +} + +func (s *Service) publishProposalEvent(proposal domain.Proposal, now time.Time) { + for _, participant := range proposal.Participants { + _ = s.PublishControlPlaneEvent(ControlPlaneEvent{ + Event: "proposal_changed", Revision: proposal.Revision, ResourceID: proposal.ProposalID, + OccurredAt: now, State: string(proposal.State), PlayerID: participant.PlayerID, + }) + } +} + func isWebSocketUpgrade(r *http.Request) bool { return strings.EqualFold(r.Header.Get("Upgrade"), "websocket") && headerContainsToken(r.Header.Values("Connection"), "upgrade") } diff --git a/server/api/service.go b/server/api/service.go index 8bc401f4..e2d1732f 100644 --- a/server/api/service.go +++ b/server/api/service.go @@ -197,6 +197,7 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) { writeDomainError(w, err) return } + s.publishTicketEvent(ticket, now) writeJSON(w, http.StatusCreated, toQueueResponse(ticket)) return } @@ -225,6 +226,7 @@ func (s *Service) queueCreate(w http.ResponseWriter, r *http.Request) { writeDomainError(w, err) return } + s.publishTicketEvent(ticket, now) writeJSON(w, http.StatusCreated, toQueueResponse(ticket)) } @@ -366,6 +368,7 @@ func (s *Service) queueMutation(w http.ResponseWriter, r *http.Request) { writeDomainError(w, err) return } + s.publishTicketEvent(ticket, now) if r.Header.Get("X-Contract-Delete") == "1" { w.WriteHeader(http.StatusNoContent) return @@ -404,7 +407,10 @@ func (s *Service) proposalMutation(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusNotFound, "not_found") return } - proposal.Expire(s.now()) + now := s.now() + if proposal.Expire(now) { + s.publishProposalEvent(*proposal, now) + } writeJSON(w, http.StatusOK, toProposalResponse(*proposal)) return } @@ -429,11 +435,13 @@ func (s *Service) proposalMutation(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusNotFound, "not_found") return } - updated, err := proposal.Respond(playerID, key, parts[1] == "accept", revision, s.now()) + now := s.now() + updated, err := proposal.Respond(playerID, key, parts[1] == "accept", revision, now) if err != nil { writeDomainError(w, err) return } + s.publishProposalEvent(updated, now) writeJSON(w, http.StatusOK, toProposalResponse(updated)) } @@ -465,6 +473,7 @@ func (s *Service) assignment(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusServiceUnavailable, "assignment_unavailable") return } + _ = s.PublishControlPlaneEvent(ControlPlaneEvent{Event: "assignment_changed", Revision: 0, ResourceID: view.MatchID, OccurredAt: now, MatchID: view.MatchID, ServerID: view.ServerID, PlayerID: view.PlayerID}) writeJSON(w, http.StatusOK, view) } diff --git a/server/api/service_test.go b/server/api/service_test.go index 53192cbc..4a7a2cbd 100644 --- a/server/api/service_test.go +++ b/server/api/service_test.go @@ -295,6 +295,52 @@ func TestEventHubRejectsEventsOutsideTheV1Vocabulary(t *testing.T) { } } +func TestStateChangingAPIActionsPublishTargetedEvents(t *testing.T) { + now := time.Unix(1000, 0).UTC() + backend := &queueBackendSpy{} + service := &Service{SessionBackend: &sessionBackendSpy{}, QueueBackend: backend, Now: func() time.Time { return now }, Proposals: make(map[string]*domain.Proposal)} + subscriber := service.getEventHub().subscribe("player-1") + defer service.getEventHub().unsubscribe(subscriber) + + create := httptest.NewRequest(http.MethodPost, "/v1/queue", strings.NewReader(`{"ticket_id":"ticket-1234567890123456","playlist":"casual","client_build":"build-1","protocol_version":1}`)) + create.Header.Set("Authorization", "Bearer session-1:token-1") + create.Header.Set("Idempotency-Key", "create-event-key-123456") + createRecorder := httptest.NewRecorder() + service.queueCreate(createRecorder, create) + if createRecorder.Code != http.StatusCreated { + t.Fatalf("create status = %d", createRecorder.Code) + } + var queueEvent ControlPlaneEvent + if err := json.Unmarshal(<-subscriber.queue, &queueEvent); err != nil { + t.Fatal(err) + } + if queueEvent.Event != "state_changed" || queueEvent.ResourceID != "ticket-1234567890123456" || queueEvent.PlayerID != "" { + t.Fatalf("queue event = %+v", queueEvent) + } + + proposal, err := domain.NewProposal("proposal-1234567890123456", domain.Casual, []string{"player-1", "player-2"}, now) + if err != nil { + t.Fatal(err) + } + service.Proposals[proposal.ProposalID] = &proposal + respond := httptest.NewRequest(http.MethodPost, "/v1/proposals/"+proposal.ProposalID+"/accept", nil) + respond.Header.Set("Authorization", "Bearer session-1:token-1") + respond.Header.Set("Idempotency-Key", "proposal-event-key-123456") + respond.Header.Set("If-Match-Revision", "0") + respondRecorder := httptest.NewRecorder() + service.proposalMutation(respondRecorder, respond) + if respondRecorder.Code != http.StatusOK { + t.Fatalf("proposal status = %d", respondRecorder.Code) + } + var proposalEvent ControlPlaneEvent + if err := json.Unmarshal(<-subscriber.queue, &proposalEvent); err != nil { + t.Fatal(err) + } + if proposalEvent.Event != "proposal_changed" || proposalEvent.ResourceID != proposal.ProposalID || proposalEvent.State != "OPEN" { + t.Fatalf("proposal event = %+v", proposalEvent) + } +} + 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())