feat(multiplayer): publish allocation state events

multiplayer-next.md

@@ -1416,3 +1416,5 @@ The no-show policy now has an explicit domain translation layer (`PlanInitialCon
 The durable no-show boundary is now implemented by `ApplyInitialConnectPlan`: it locks the match and roster, validates that the plan covers every active participant, records deterministic no-show cooldown penalties, fails no-show tickets, requeues innocent tickets on cancellation or advances connected tickets to `LIVE` for eligible casual bot start, and emits a replayable state-change outbox event under the same serializable transaction. Idempotency keys reject conflicting retries. Focused store tests, race tests, and vet pass; the real PostgreSQL integration remains an environment-dependent gate.
 
 The maintenance command now invokes a bounded `ReconcileInitialConnect` sweep for `ASSIGNMENT_READY`/`ASSIGNED`/`CONNECTING` matches, carrying ranked no-show history into the domain ladder and skipping non-actionable WAIT plans. This closes the local control-plane trigger for task 8.35; actual allocated-server bot spawning, shutdown signaling, and live Agones integration remain separate gates.
+
+Allocation registration now writes a participant-targeted, revisioned `state_changed` outbox event for both `PROCESS_READY` and `ASSIGNMENT_READY` transitions. The production and test API binaries run a type-scoped dispatcher with delivery-before-ack semantics, so allocation lifecycle events survive WebSocket outages without competing with proposal or result consumers. Store/API adversarial tests cover event-type isolation, target validation, and revision mismatches; live allocator/Agones delivery remains an integration gate.

server/api/outbox.go

@@ -64,6 +64,32 @@ func RunResultOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service
 	}
 }
 
+// RunStateOutboxDispatcher delivers committed allocation/no-show lifecycle
+// transitions to each participant without acknowledging proposal or result
+// events owned by the other dispatchers.
+func RunStateOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service) {
+	if db == nil || service == nil {
+		return
+	}
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+	dispatcher := store.NewOutboxDispatcher(db, func(deliveryCtx context.Context, event store.OutboxEvent) error {
+		return deliverStateOutboxEvent(deliveryCtx, event, service)
+	})
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			events, err := store.ReadUnpublishedStateOutbox(ctx, db, 100)
+			if err != nil {
+				continue
+			}
+			_ = dispatchOutboxEvents(ctx, dispatcher, events)
+		}
+	}
+}
+
 func dispatchOutboxEvents(ctx context.Context, dispatcher *store.OutboxDispatcher, events []store.OutboxEvent) error {
 	if len(events) == 0 {
 		return nil
@@ -136,3 +162,33 @@ func deliverResultOutboxEvent(ctx context.Context, db *sql.DB, event store.Outbo
 	}
 	return nil
 }
+
+func deliverStateOutboxEvent(_ context.Context, event store.OutboxEvent, service *Service) error {
+	if event.EventType != "state_changed" || event.AggregateID == "" || event.Revision == 0 || len(event.Payload) == 0 {
+		return fmt.Errorf("invalid state outbox event")
+	}
+	var envelope struct {
+		Event      string    `json:"event"`
+		Revision   uint64    `json:"revision"`
+		ResourceID string    `json:"resource_id"`
+		OccurredAt time.Time `json:"occurred_at"`
+		State      string    `json:"state"`
+		MatchID    string    `json:"match_id"`
+		PlayerIDs  []string  `json:"player_ids"`
+	}
+	if err := json.Unmarshal(event.Payload, &envelope); err != nil {
+		return fmt.Errorf("decode state outbox event: %w", err)
+	}
+	if envelope.Event != "state_changed" || envelope.ResourceID != event.AggregateID || envelope.Revision != event.Revision || envelope.State == "" || len(envelope.PlayerIDs) == 0 {
+		return fmt.Errorf("invalid state outbox payload")
+	}
+	for _, playerID := range envelope.PlayerIDs {
+		if playerID == "" {
+			return fmt.Errorf("state outbox event has empty participant")
+		}
+		if err := service.PublishControlPlaneEvent(ControlPlaneEvent{Event: "state_changed", Revision: envelope.Revision, ResourceID: envelope.ResourceID, OccurredAt: envelope.OccurredAt, State: envelope.State, MatchID: envelope.MatchID, PlayerID: playerID}); err != nil {
+			return err
+		}
+	}
+	return nil
+}

server/api/outbox_test.go

@@ -61,3 +61,22 @@ func TestDeliverResultOutboxEventRejectsMalformedRows(t *testing.T) {
 		}
 	}
 }
+
+func TestDeliverStateOutboxEventValidatesRevisionAndTargets(t *testing.T) {
+	service := &Service{}
+	first := service.getEventHub().subscribe("player-a")
+	defer service.getEventHub().unsubscribe(first)
+	payload := []byte(`{"event":"state_changed","revision":4,"resource_id":"match-1","occurred_at":"1970-01-01T00:16:40Z","state":"ASSIGNMENT_READY","match_id":"match-1","player_ids":["player-a"]}`)
+	if err := deliverStateOutboxEvent(context.Background(), store.OutboxEvent{EventType: "state_changed", AggregateID: "match-1", Revision: 4, Payload: payload}, service); err != nil {
+		t.Fatalf("valid state event rejected: %v", err)
+	}
+	select {
+	case <-first.queue:
+	case <-time.After(time.Second):
+		t.Fatal("participant did not receive state event")
+	}
+	bad := []byte(`{"event":"state_changed","revision":3,"resource_id":"match-1","state":"LIVE","player_ids":["player-a"]}`)
+	if err := deliverStateOutboxEvent(context.Background(), store.OutboxEvent{EventType: "state_changed", AggregateID: "match-1", Revision: 4, Payload: bad}, service); err == nil {
+		t.Fatal("revision-mismatched state event accepted")
+	}
+}

server/cmd/control-plane/main.go

@@ -70,6 +70,7 @@ func main() {
 	defer stop()
 	go api.RunProposalOutboxDispatcher(ctx, db, service)
 	go api.RunResultOutboxDispatcher(ctx, db, service)
+	go api.RunStateOutboxDispatcher(ctx, db, service)
 	select {
 	case err := <-serveErr:
 		if err != nil && err != http.ErrServerClosed {

server/cmd/testkit-api/main.go

@@ -86,6 +86,7 @@ func main() {
 	defer stop()
 	go api.RunProposalOutboxDispatcher(ctx, db, service)
 	go api.RunResultOutboxDispatcher(ctx, db, service)
+	go api.RunStateOutboxDispatcher(ctx, db, service)
 	select {
 	case err := <-serveErr:
 		if err != nil && err != http.ErrServerClosed {

server/store/allocation_match_sql.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"time"
 
@@ -45,7 +46,7 @@ const BindAllocatedMatchParticipantsSQL = `WITH bound AS (
           SELECT 1 FROM allocations
           WHERE allocation_id = $2 AND match_id = $1 AND server_id = $3 AND state = 'ALLOCATED'
       )
-    RETURNING match_id
+    RETURNING match_id, revision
 ), participants AS (
     SELECT mp.ticket_id, mp.player_id
     FROM match_participants mp
@@ -77,7 +78,13 @@ const AdvanceServerRegistrationSQL = `WITH matched AS (
     WHERE q.ticket_id = mp.ticket_id AND q.player_id = mp.player_id AND q.state = $3
     RETURNING q.ticket_id
 )
-SELECT (SELECT count(*) FROM matched), (SELECT count(*) FROM match_participants WHERE match_id = $1), (SELECT count(*) FROM advanced)`
+SELECT (SELECT count(*) FROM matched), (SELECT count(*) FROM match_participants WHERE match_id = $1), (SELECT count(*) FROM advanced), COALESCE((SELECT revision FROM matched), -1)`
+
+const serverRegistrationParticipantIDsSQL = `SELECT player_id FROM match_participants WHERE match_id = $1 ORDER BY player_id`
+
+const serverRegistrationOutboxSQL = `INSERT INTO outbox
+    (event_id, aggregate_type, aggregate_id, revision, event_type, payload)
+VALUES ($1, 'match', $2, $3, 'state_changed', $4)`
 
 const ServerRegistrationIdempotencyScope = "server.register"
 
@@ -120,13 +127,42 @@ func AdvanceServerRegistration(ctx context.Context, db *sql.DB, binding domain.W
 			return nil
 		}
 		var matched, participants, advanced int
-		if err := tx.QueryRowContext(ctx, AdvanceServerRegistrationSQL, binding.MatchID, binding.ServerID, from, to, binding.AllocationID, now, protocol).Scan(&matched, &participants, &advanced); err != nil {
+		var revision int64
+		if err := tx.QueryRowContext(ctx, AdvanceServerRegistrationSQL, binding.MatchID, binding.ServerID, from, to, binding.AllocationID, now, protocol).Scan(&matched, &participants, &advanced, &revision); err != nil {
 			return err
 		}
 		if matched != 1 || participants == 0 || advanced != participants {
 			return domain.ErrConflict
 		}
-		return nil
+		rows, err := tx.QueryContext(ctx, serverRegistrationParticipantIDsSQL, binding.MatchID)
+		if err != nil {
+			return err
+		}
+		playerIDs := make([]string, 0, participants)
+		for rows.Next() {
+			var playerID string
+			if err := rows.Scan(&playerID); err != nil {
+				rows.Close()
+				return err
+			}
+			playerIDs = append(playerIDs, playerID)
+		}
+		if err := rows.Err(); err != nil {
+			rows.Close()
+			return err
+		}
+		if err := rows.Close(); err != nil {
+			return err
+		}
+		payload, err := json.Marshal(map[string]any{
+			"event": "state_changed", "revision": revision, "resource_id": binding.MatchID,
+			"occurred_at": now, "state": string(to), "match_id": binding.MatchID, "player_ids": playerIDs,
+		})
+		if err != nil {
+			return err
+		}
+		_, err = tx.ExecContext(ctx, serverRegistrationOutboxSQL, fmt.Sprintf("match:%s:%d", binding.MatchID, revision), binding.MatchID, revision, payload)
+		return err
 	})
 }
 

server/store/outbox.go

@@ -42,6 +42,13 @@ WHERE published_at IS NULL AND event_type = 'match_completed'
 ORDER BY created_at, event_id
 LIMIT $1`
 
+const OutboxUnpublishedStateSelectSQL = `SELECT event_id, aggregate_type, aggregate_id, revision,
+    event_type, payload, created_at, published_at
+FROM outbox
+WHERE published_at IS NULL AND event_type = 'state_changed'
+ORDER BY created_at, event_id
+LIMIT $1`
+
 const MatchParticipantIDsSQL = `SELECT player_id
 FROM match_participants
 WHERE match_id = $1
@@ -125,6 +132,12 @@ func ReadUnpublishedResultOutbox(ctx context.Context, db *sql.DB, limit int) ([]
 	return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedResultSelectSQL)
 }
 
+// ReadUnpublishedStateOutbox returns lifecycle state events, leaving proposal
+// and result rows to their dedicated consumers.
+func ReadUnpublishedStateOutbox(ctx context.Context, db *sql.DB, limit int) ([]OutboxEvent, error) {
+	return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedStateSelectSQL)
+}
+
 func ReadMatchParticipantIDs(ctx context.Context, db *sql.DB, matchID string) ([]string, error) {
 	if db == nil || matchID == "" {
 		return nil, fmt.Errorf("invalid match participant read arguments")

server/store/outbox_test.go

@@ -13,6 +13,7 @@ func TestOutboxSQLPreservesReplayableOrderedReadAndPublishAck(t *testing.T) {
 		OutboxUnpublishedSelectSQL:         {"published_at IS NULL", "ORDER BY created_at, event_id", "LIMIT $1"},
 		OutboxUnpublishedProposalSelectSQL: {"published_at IS NULL", "event_type = 'proposal_changed'", "ORDER BY created_at, event_id", "LIMIT $1"},
 		OutboxUnpublishedResultSelectSQL:   {"published_at IS NULL", "event_type = 'match_completed'", "ORDER BY created_at, event_id", "LIMIT $1"},
+		OutboxUnpublishedStateSelectSQL:    {"published_at IS NULL", "event_type = 'state_changed'", "ORDER BY created_at, event_id", "LIMIT $1"},
 		MatchParticipantIDsSQL:             {"SELECT player_id", "match_participants", "match_id = $1", "ORDER BY player_id"},
 		OutboxMarkPublishedSQL:             {"published_at = $2", "event_id = $1", "published_at IS NULL"},
 	} {