From 79a6092b28dafe233b8b78960b2214916f53eedf Mon Sep 17 00:00:00 2001 From: Josh Creek <8179928+jcreek@users.noreply.github.com> Date: Tue, 1 Sep 2026 15:21:49 +0100 Subject: [PATCH] feat(multiplayer): wire production proposal outbox delivery --- multiplayer-next.md | 2 +- server/api/outbox.go | 85 ++++++++++++++++++++++++++++++++ server/api/outbox_test.go | 51 +++++++++++++++++++ server/cmd/control-plane/main.go | 12 +++-- server/cmd/testkit-api/main.go | 43 +--------------- server/store/outbox.go | 21 +++++++- server/store/outbox_test.go | 5 +- 7 files changed, 170 insertions(+), 49 deletions(-) create mode 100644 server/api/outbox.go create mode 100644 server/api/outbox_test.go diff --git a/multiplayer-next.md b/multiplayer-next.md index c20ff55a..da4dacd7 100644 --- a/multiplayer-next.md +++ b/multiplayer-next.md @@ -1234,7 +1234,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 publishes targeted authenticated revisioned queue/proposal/assignment events and the Godot client consumes state/proposal/assignment events, fetching the authoritative assignment after assignment readiness; the API can now use the durable participant-scoped proposal provider for both recovery and accept/decline mutations | `test_matchmaking_state.gd`, `test_control_plane_client.gd`, `test_matchmaking_ui.gd`, `TestProposalRecoveryIsParticipantScopedAndExpiresAtReadBoundary`, `TestAuthenticatedWebSocketDeliversOnlyTargetedRevisionedEvents`, `TestStateChangingAPIActionsPublishTargetedEvents`, `TestProposalRecoveryUsesDurableBackendAndRemainsParticipantScoped` and store proposal SQL tests 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 using Godot 4.7's handshake-header API, consume validated events, automatically reconnect with bounded backoff, and trigger REST recovery on projection gaps and stream return, including a deferred recovery when an event arrives during an in-flight HTTP mutation; `OutboxDispatcher` now provides ordered at-least-once delivery after durable commit; assignment recovery events carry the durable assignment revision without leaking it into REST JSON; prediction startup now distinguishes the sequence-0 warm-up acknowledgement from genuine missing/evicted history, preventing a false hard-snap during a live match; proposal creation now writes one durable targeted `proposal_changed` outbox event per proposal aggregate revision, and the test-only API harness dispatches it to authenticated participants | `server/domain/sync.go`, `server/api/events.go`, `server/api/service.go`, `server/store/outbox.go`, `server/store/proposal_sql.go`, `server/cmd/testkit-api`, `service_test.go`, `outbox_test.go`, `matchmaking_state.gd`, `control_plane_client.gd`, `local_prediction_history.gd`, `net_ship_predictor.gd` and tests 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, invalid-event rejection, delivery-before-ack failure ordering, assignment event/response revision separation, prediction warm-up/hard-resync separation and deferred proposal recovery; Godot 4.7.1 headless project parse and 143-test unit harness pass with compatibility rendering, and a two-process ENet authoritative match smoke passes spawn, movement, prediction, 0% snapshot loss and 0 hard snaps; `cmd/control-plane` provides a signal-bound API role, `cmd/matcher` now supports casual and explicitly enabled ranked roles with durable identity lookup, and `cmd/maintenance` provides bounded season maintenance. **Live multi-process control-plane/game verification now exists**: `scripts/verify_control_plane_integration.sh` runs a real `postgres:17-alpine`, the real `api.Service` (via the new test-only `server/cmd/testkit-api`, wired identically to `cmd/control-plane` except for a fake Steam login — see §8.7), and a real headless Godot client (`control_plane_smoke.gd`) round-tripping login → `fetch_ranked_profile` (expect 404, §8.22) → `queue_create` → heartbeat → `cancel_queue` over an actual network connection — the first time this boundary was exercised end to end rather than against a mock on either side. **The two-player proposal extension now also passes**: `scripts/verify_control_plane_proposal_integration.sh` starts isolated PostgreSQL, the real matcher and testkit API, launches two headless Godot clients, seeds deterministic RTT projections, observes the real WebSocket `OPEN` proposal, accepts from both clients, and verifies both durable tickets plus the proposal reach `ACCEPTED` (including authoritative recovery when the concurrent revision advances). Allocator, production outbox wiring and Redis fan-out live 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 using Godot 4.7's handshake-header API, consume validated events, automatically reconnect with bounded backoff, and trigger REST recovery on projection gaps and stream return, including a deferred recovery when an event arrives during an in-flight HTTP mutation; `OutboxDispatcher` now provides ordered at-least-once delivery after durable commit; assignment recovery events carry the durable assignment revision without leaking it into REST JSON; prediction startup now distinguishes the sequence-0 warm-up acknowledgement from genuine missing/evicted history, preventing a false hard-snap during a live match; proposal creation now writes one durable targeted `proposal_changed` outbox event per proposal aggregate revision, and both production `cmd/control-plane` and the test-only API harness dispatch only that event type to authenticated participants, leaving result events for their separate consumer | `server/domain/sync.go`, `server/api/events.go`, `server/api/outbox.go`, `server/api/service.go`, `server/store/outbox.go`, `server/store/proposal_sql.go`, `server/cmd/control-plane`, `server/cmd/testkit-api`, `service_test.go`, `outbox_test.go`, `matchmaking_state.gd`, `control_plane_client.gd`, `local_prediction_history.gd`, `net_ship_predictor.gd` and tests 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, invalid-event rejection, delivery-before-ack failure ordering, assignment event/response revision separation, prediction warm-up/hard-resync separation, deferred proposal recovery and proposal outbox filtering/delivery failures; Godot 4.7.1 headless project parse and 150-test unit harness pass with compatibility rendering, and a two-process ENet authoritative match smoke passes spawn, movement, prediction, 0% snapshot loss and 0 hard snaps; `cmd/control-plane` provides a signal-bound API role, `cmd/matcher` now supports casual and explicitly enabled ranked roles with durable identity lookup, and `cmd/maintenance` provides bounded season maintenance. **Live multi-process control-plane/game verification now exists**: `scripts/verify_control_plane_integration.sh` runs a real `postgres:17-alpine`, the real `api.Service` (via the new test-only `server/cmd/testkit-api`, wired identically to `cmd/control-plane` except for a fake Steam login — see §8.7), and a real headless Godot client (`control_plane_smoke.gd`) round-tripping login → `fetch_ranked_profile` (expect 404, §8.22) → `queue_create` → heartbeat → `cancel_queue` over an actual network connection — the first time this boundary was exercised end to end rather than against a mock on either side. **The two-player proposal extension now also passes**: `scripts/verify_control_plane_proposal_integration.sh` starts isolated PostgreSQL, the real matcher and testkit API, launches two headless Godot clients, seeds deterministic RTT projections, observes the real WebSocket `OPEN` proposal, accepts from both clients, and verifies both durable tickets plus the proposal reach `ACCEPTED` (including authoritative recovery when the concurrent revision advances). Allocator and Redis fan-out live 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 including the hosted endpoint; Godot `AssignmentState`/`ControlPlaneClient.fetch_assignment()` bind the response to the authenticated player, recheck expiry, validate and retain the endpoint, and `connect_to_assignment()` now starts only the validated ENet/Steam transport after assignment readiness; the opaque join authorisation is carried in the MatchNet hello payload rather than the endpoint URL; allocated Godot servers now fail closed unless an operator-mounted JSON roster of control-plane signed envelopes and an HMAC-SHA256 key are present, and MatchNet verifies the canonical Go claim bytes, checks exact token membership plus match/server/protocol/slot/expiry claims before admitting a peer, rejects concurrent reuse of an active token, tracks server-owned reconnect generations across the 60-second reclaim window, fences expired reclaims and invalidates old peers; migration 0002 and the Go store adapter now persist/recover the complete player-scoped assignment projection with conflict-safe identical replay; `AssignmentProviderFromStore` wires that durable projection into the API injection point; `SaveAssignments` publishes a complete signed roster atomically instead of allowing partial player visibility; `SaveVerifiedAssignmentRoster` rechecks signed claims before deriving player rows | `server/api/service.go`, `store_adapters.go`, `service_test.go`, `assignment_state.gd`, `control_plane_client.gd`, `match_net.gd`, `server_boot.gd`, `server_config.gd`, `test_assignment_state.gd`, `test_control_plane_client.gd`, `test_match_net.gd`, `server/contracts/v1/openapi.json`, `server/migrations/0002_assignments.sql` and `server/store/assignment_sql.go` cover participant/identity/expiry/shape/transport/endpoint boundaries, player-scoped schema keys, expiry-filtered reads, assignment upsert conflict handling, atomic batch validation, signed-claim binding, strict endpoint splitting, allowlisted claim rejection, canonical HMAC interoperability, forged-signature rejection, duplicate active-token rejection, reclaim generation/expiry behavior and 146-test Godot compatibility coverage; the three-process disconnect/reclaim smoke passes ship retention, replacement drive and old-peer invalidation; SDR relay-ticket installation, live allocated-token process integration and live Godot/PostgreSQL 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 | | 8.43 `[D:8.39,8.40,8.41]` | **IN PROGRESS.** Matchmaking client now distinguishes expired queue recovery, session expiry, missing records and retryable control-plane outages; a 401 clears the in-memory token, emits `session_expired` and disables retry until a new session is configured; terminal messages remain visible and active searches are not falsely failed on transient errors | `MatchmakingState` and `ControlPlaneClient` tests cover explicit expiry and the existing terminal/retry-safe state paths; decline, version mismatch, regional outage retry UI, failed reconnect, duplicate-action recovery and live Godot verification remain | diff --git a/server/api/outbox.go b/server/api/outbox.go new file mode 100644 index 00000000..357e02fc --- /dev/null +++ b/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 +} diff --git a/server/api/outbox_test.go b/server/api/outbox_test.go new file mode 100644 index 00000000..d9b03a4d --- /dev/null +++ b/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") + } + }) + } +} diff --git a/server/cmd/control-plane/main.go b/server/cmd/control-plane/main.go index b78549bc..33e363e2 100644 --- a/server/cmd/control-plane/main.go +++ b/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. diff --git a/server/cmd/testkit-api/main.go b/server/cmd/testkit-api/main.go index 42487511..ecd510fe 100644 --- a/server/cmd/testkit-api/main.go +++ b/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. diff --git a/server/store/outbox.go b/server/store/outbox.go index 1bf70500..9dd208c9 100644 --- a/server/store/outbox.go +++ b/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 } diff --git a/server/store/outbox_test.go b/server/store/outbox_test.go index 03aa77c3..8e6a6135 100644 --- a/server/store/outbox_test.go +++ b/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) {