feat(multiplayer): dispatch completed result events

server/api/outbox.go

@@ -38,6 +38,32 @@ func RunProposalOutboxDispatcher(ctx context.Context, db *sql.DB, service *Servi
 	}
 }
 
+// RunResultOutboxDispatcher delivers committed match results as targeted
+// COMPLETED state events. It owns only match_completed rows; proposal rows
+// remain with RunProposalOutboxDispatcher.
+func RunResultOutboxDispatcher(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 deliverResultOutboxEvent(deliveryCtx, db, event, service)
+	})
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			events, err := store.ReadUnpublishedResultOutbox(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
@@ -83,3 +109,30 @@ func deliverProposalOutboxEvent(_ context.Context, event store.OutboxEvent, serv
 	}
 	return nil
 }
+
+func deliverResultOutboxEvent(ctx context.Context, db *sql.DB, event store.OutboxEvent, service *Service) error {
+	if event.EventType != "match_completed" || event.AggregateID == "" || event.Revision == 0 || len(event.Payload) == 0 {
+		return fmt.Errorf("invalid result outbox event")
+	}
+	var payload map[string]any
+	if err := json.Unmarshal(event.Payload, &payload); err != nil || payload == nil {
+		return fmt.Errorf("decode result outbox event: %w", err)
+	}
+	players, err := store.ReadMatchParticipantIDs(ctx, db, event.AggregateID)
+	if err != nil {
+		return err
+	}
+	if len(players) == 0 {
+		return fmt.Errorf("result outbox event has no participants")
+	}
+	for _, playerID := range players {
+		if err := service.PublishControlPlaneEvent(ControlPlaneEvent{
+			Event: "state_changed", Revision: event.Revision, ResourceID: event.AggregateID,
+			OccurredAt: event.CreatedAt, State: "COMPLETED", MatchID: event.AggregateID,
+			PlayerID: playerID,
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
+}

server/api/outbox_test.go

@@ -49,3 +49,15 @@ func TestDeliverProposalOutboxEventRejectsMalformedOrUntargetedRows(t *testing.T
 		})
 	}
 }
+
+func TestDeliverResultOutboxEventRejectsMalformedRows(t *testing.T) {
+	for _, event := range []store.OutboxEvent{
+		{EventType: "proposal_changed", AggregateID: "match-1", Revision: 1, Payload: []byte(`{}`)},
+		{EventType: "match_completed", AggregateID: "", Revision: 1, Payload: []byte(`{}`)},
+		{EventType: "match_completed", AggregateID: "match-1", Revision: 1, Payload: []byte(`not-json`)},
+	} {
+		if err := deliverResultOutboxEvent(nil, nil, event, &Service{}); err == nil {
+			t.Fatalf("invalid result event accepted: %+v", event)
+		}
+	}
+}

server/cmd/control-plane/main.go

@@ -69,6 +69,7 @@ func main() {
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 	go api.RunProposalOutboxDispatcher(ctx, db, service)
+	go api.RunResultOutboxDispatcher(ctx, db, service)
 	select {
 	case err := <-serveErr:
 		if err != nil && err != http.ErrServerClosed {

server/cmd/testkit-api/main.go

@@ -82,6 +82,7 @@ func main() {
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
 	go api.RunProposalOutboxDispatcher(ctx, db, service)
+	go api.RunResultOutboxDispatcher(ctx, db, service)
 	select {
 	case err := <-serveErr:
 		if err != nil && err != http.ErrServerClosed {

server/store/outbox.go

@@ -35,6 +35,18 @@ WHERE published_at IS NULL AND event_type = 'proposal_changed'
 ORDER BY created_at, event_id
 LIMIT $1`
 
+const OutboxUnpublishedResultSelectSQL = `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 = 'match_completed'
+ORDER BY created_at, event_id
+LIMIT $1`
+
+const MatchParticipantIDsSQL = `SELECT player_id
+FROM match_participants
+WHERE match_id = $1
+ORDER BY player_id`
+
 const OutboxMarkPublishedSQL = `UPDATE outbox
 SET published_at = $2
 WHERE event_id = $1 AND published_at IS NULL`
@@ -106,6 +118,36 @@ func ReadUnpublishedProposalOutbox(ctx context.Context, db *sql.DB, limit int) (
 	return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedProposalSelectSQL)
 }
 
+// ReadUnpublishedResultOutbox returns only durable match-completion events.
+// Proposal and result consumers acknowledge separate event types so one
+// transient fan-out outage cannot hide rows owned by another consumer.
+func ReadUnpublishedResultOutbox(ctx context.Context, db *sql.DB, limit int) ([]OutboxEvent, error) {
+	return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedResultSelectSQL)
+}
+
+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")
+	}
+	rows, err := db.QueryContext(ctx, MatchParticipantIDsSQL, matchID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var players []string
+	for rows.Next() {
+		var playerID string
+		if err := rows.Scan(&playerID); err != nil {
+			return nil, err
+		}
+		players = append(players, playerID)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return players, nil
+}
+
 func readUnpublishedOutbox(ctx context.Context, db *sql.DB, limit int, query string) ([]OutboxEvent, error) {
 	if db == nil || limit < 1 || limit > 1000 {
 		return nil, fmt.Errorf("invalid outbox read arguments")

server/store/outbox_test.go

@@ -12,6 +12,8 @@ func TestOutboxSQLPreservesReplayableOrderedReadAndPublishAck(t *testing.T) {
 	for query, fragments := range map[string][]string{
 		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"},
+		MatchParticipantIDsSQL:             {"SELECT player_id", "match_participants", "match_id = $1", "ORDER BY player_id"},
 		OutboxMarkPublishedSQL:             {"published_at = $2", "event_id = $1", "published_at IS NULL"},
 	} {
 		for _, fragment := range fragments {
@@ -67,4 +69,7 @@ func TestOutboxAdaptersRejectUnsafeArgumentsWithoutDatabase(t *testing.T) {
 	if err := MarkOutboxPublished(nil, nil, "", time.Unix(1000, 0)); err == nil {
 		t.Fatal("empty event acknowledgement accepted")
 	}
+	if _, err := ReadMatchParticipantIDs(nil, nil, ""); err == nil {
+		t.Fatal("invalid participant read arguments accepted")
+	}
 }