feat(multiplayer): wire production proposal outbox delivery

server/api/outbox.go

@@ -0,0 +1,85 @@
+package api
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/store"
+)
+
+// RunProposalOutboxDispatcher delivers committed proposal changes to the
+// authenticated WebSocket subscribers. It only reads proposal_changed rows;
+// result and other outbox event types remain owned by their own consumers.
+// Delivery is at-least-once because the row is acknowledged only after every
+// participant publication succeeds.
+func RunProposalOutboxDispatcher(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 deliverProposalOutboxEvent(deliveryCtx, event, service)
+	})
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			events, err := store.ReadUnpublishedProposalOutbox(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
+	}
+	// Use the same delivery-before-ack contract as the general dispatcher,
+	// while keeping the already-filtered batch from being read a second time.
+	for _, event := range events {
+		if event.EventID == "" {
+			return fmt.Errorf("outbox event has no ID")
+		}
+		if err := dispatcher.Deliver(ctx, event); err != nil {
+			return err
+		}
+		if err := dispatcher.Ack(ctx, event.EventID, time.Now().UTC()); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func deliverProposalOutboxEvent(_ context.Context, event store.OutboxEvent, service *Service) error {
+	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"`
+		PlayerIDs  []string  `json:"player_ids"`
+	}
+	if err := json.Unmarshal(event.Payload, &envelope); err != nil {
+		return fmt.Errorf("decode proposal outbox event: %w", err)
+	}
+	if envelope.Event != "proposal_changed" || envelope.ResourceID == "" || len(envelope.PlayerIDs) == 0 {
+		return fmt.Errorf("invalid proposal outbox event")
+	}
+	for _, playerID := range envelope.PlayerIDs {
+		if err := service.PublishControlPlaneEvent(ControlPlaneEvent{
+			Event: envelope.Event, Revision: envelope.Revision, ResourceID: envelope.ResourceID,
+			OccurredAt: envelope.OccurredAt, State: envelope.State, PlayerID: playerID,
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
+}

server/api/outbox_test.go

@@ -0,0 +1,51 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/cosmic-clash/cosmic-clash/server/store"
+)
+
+func TestDeliverProposalOutboxEventPublishesEveryTarget(t *testing.T) {
+	service := &Service{}
+	first := service.getEventHub().subscribe("player-a")
+	second := service.getEventHub().subscribe("player-b")
+	defer service.getEventHub().unsubscribe(first)
+	defer service.getEventHub().unsubscribe(second)
+
+	payload, err := json.Marshal(map[string]any{
+		"event": "proposal_changed", "revision": uint64(0), "resource_id": "proposal-1",
+		"occurred_at": time.Unix(1000, 0).UTC(), "state": "OPEN", "player_ids": []string{"player-a", "player-b"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := deliverProposalOutboxEvent(context.Background(), store.OutboxEvent{EventID: "event-1", Payload: payload}, service); err != nil {
+		t.Fatalf("deliver proposal event: %v", err)
+	}
+	for name, subscriber := range map[string]*eventSubscriber{"player-a": first, "player-b": second} {
+		select {
+		case <-subscriber.queue:
+		case <-time.After(time.Second):
+			t.Fatalf("%s did not receive targeted proposal event", name)
+		}
+	}
+}
+
+func TestDeliverProposalOutboxEventRejectsMalformedOrUntargetedRows(t *testing.T) {
+	service := &Service{}
+	for name, event := range map[string]store.OutboxEvent{
+		"malformed":      {Payload: []byte("{")},
+		"wrong event":    {Payload: []byte(`{"event":"match_completed","resource_id":"match-1","player_ids":["player-a"]}`)},
+		"missing target": {Payload: []byte(`{"event":"proposal_changed","resource_id":"proposal-1","player_ids":[]}`)},
+	} {
+		t.Run(name, func(t *testing.T) {
+			if err := deliverProposalOutboxEvent(context.Background(), event, service); err == nil {
+				t.Fatal("malformed or untargeted event accepted")
+			}
+		})
+	}
+}

server/cmd/control-plane/main.go

@@ -61,11 +61,13 @@ func main() {
 	if *workloadSecret == "" {
 		fmt.Fprintln(os.Stderr, "control-plane: warning: --workload-secret / COSMIC_CLASH_WORKLOAD_SECRET is unset; server registration and result submission will return 503")
 	}
-	server := &http.Server{Addr: *listen, Handler: newAPIHandler(db, *workloadSecret, candidateIndex), ReadHeaderTimeout: 5 * time.Second}
+	service := newAPIService(db, *workloadSecret, candidateIndex)
+	server := &http.Server{Addr: *listen, Handler: service.Handler(), ReadHeaderTimeout: 5 * time.Second}
 	serveErr := make(chan error, 1)
 	go func() { serveErr <- server.ListenAndServe() }()
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
+	go api.RunProposalOutboxDispatcher(ctx, db, service)
 	select {
 	case err := <-serveErr:
 		if err != nil && err != http.ErrServerClosed {
@@ -81,11 +83,15 @@ func main() {
 }
 
 func newAPIHandler(db *sql.DB, workloadSecret string, indexes ...api.CandidateIndex) http.Handler {
+	return newAPIService(db, workloadSecret, indexes...).Handler()
+}
+
+func newAPIService(db *sql.DB, workloadSecret string, indexes ...api.CandidateIndex) *api.Service {
 	var candidateIndex api.CandidateIndex
 	if len(indexes) > 0 {
 		candidateIndex = indexes[0]
 	}
-	return (&api.Service{
+	return &api.Service{
 		SessionBackend:        store.PostgresSessions{DB: db},
 		SessionIssuer:         store.PostgresSessions{DB: db},
 		QueueBackend:          store.PostgresQueue{DB: db},
@@ -100,7 +106,7 @@ func newAPIHandler(db *sql.DB, workloadSecret string, indexes ...api.CandidateIn
 		WorkloadVerify:        api.WorkloadVerifierFromSignedToken([]byte(workloadSecret), db),
 		Now:                   func() time.Time { return time.Now().UTC() },
 		Log:                   logEvent,
-	}).Handler()
+	}
 }
 
 // logEvent writes one credential-safe structured event per line to stderr.

server/cmd/testkit-api/main.go

@@ -16,7 +16,6 @@ import (
 	"crypto/sha256"
 	"database/sql"
 	"encoding/hex"
-	"encoding/json"
 	"flag"
 	"fmt"
 	"net"
@@ -81,7 +80,7 @@ func main() {
 	go func() { serveErr <- server.Serve(listener) }()
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
-	go dispatchProposalOutbox(ctx, db, service)
+	go api.RunProposalOutboxDispatcher(ctx, db, service)
 	select {
 	case err := <-serveErr:
 		if err != nil && err != http.ErrServerClosed {
@@ -94,46 +93,6 @@ func main() {
 	}
 }
 
-func dispatchProposalOutbox(ctx context.Context, db *sql.DB, service *api.Service) {
-	ticker := time.NewTicker(100 * time.Millisecond)
-	defer ticker.Stop()
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-ticker.C:
-			events, err := store.ReadUnpublishedOutbox(ctx, db, 100)
-			if err != nil {
-				continue
-			}
-			for _, event := range events {
-				if event.EventType != "proposal_changed" {
-					continue
-				}
-				var controlEvent api.ControlPlaneEvent
-				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"`
-					PlayerIDs  []string  `json:"player_ids"`
-				}
-				if err := json.Unmarshal(event.Payload, &envelope); err != nil {
-					continue
-				}
-				for _, playerID := range envelope.PlayerIDs {
-					controlEvent = api.ControlPlaneEvent{Event: envelope.Event, Revision: envelope.Revision, ResourceID: envelope.ResourceID, OccurredAt: envelope.OccurredAt, State: envelope.State, PlayerID: playerID}
-					if err := service.PublishControlPlaneEvent(controlEvent); err != nil {
-						continue
-					}
-				}
-				_ = store.MarkOutboxPublished(ctx, db, event.EventID, time.Now().UTC())
-			}
-		}
-	}
-}
-
 // fakeSteamLogin derives a deterministic identity from the ticket string
 // itself (never a real Steam Web API ticket in this binary) and ensures its
 // identities row exists so session issuance's foreign key is satisfied.

server/store/outbox.go

@@ -28,6 +28,13 @@ WHERE published_at IS NULL
 ORDER BY created_at, event_id
 LIMIT $1`
 
+const OutboxUnpublishedProposalSelectSQL = `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 = 'proposal_changed'
+ORDER BY created_at, event_id
+LIMIT $1`
+
 const OutboxMarkPublishedSQL = `UPDATE outbox
 SET published_at = $2
 WHERE event_id = $1 AND published_at IS NULL`
@@ -88,10 +95,22 @@ func (d *OutboxDispatcher) Dispatch(ctx context.Context, limit int, publishedAt
 // ReadUnpublishedOutbox returns a bounded, stable ordered batch. It does not
 // mark rows before delivery: a worker crash therefore leaves events replayable.
 func ReadUnpublishedOutbox(ctx context.Context, db *sql.DB, limit int) ([]OutboxEvent, error) {
+	return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedSelectSQL)
+}
+
+// ReadUnpublishedProposalOutbox returns only WebSocket-routable proposal
+// events. Other outbox consumers (for example result reconciliation) retain
+// ownership of their event types and cannot be acknowledged accidentally by
+// the control-plane WebSocket dispatcher.
+func ReadUnpublishedProposalOutbox(ctx context.Context, db *sql.DB, limit int) ([]OutboxEvent, error) {
+	return readUnpublishedOutbox(ctx, db, limit, OutboxUnpublishedProposalSelectSQL)
+}
+
+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")
 	}
-	rows, err := db.QueryContext(ctx, OutboxUnpublishedSelectSQL, limit)
+	rows, err := db.QueryContext(ctx, query, limit)
 	if err != nil {
 		return nil, err
 	}

server/store/outbox_test.go

@@ -10,8 +10,9 @@ import (
 
 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"},
-		OutboxMarkPublishedSQL:     {"published_at = $2", "event_id = $1", "published_at IS NULL"},
+		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"},
+		OutboxMarkPublishedSQL:             {"published_at = $2", "event_id = $1", "published_at IS NULL"},
 	} {
 		for _, fragment := range fragments {
 			if !contains(query, fragment) {