feat: publish matchmaking state events

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

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

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())