diff --git a/server/api/outbox.go b/server/api/outbox.go index 012dabe6..a797e3f0 100644 --- a/server/api/outbox.go +++ b/server/api/outbox.go @@ -35,7 +35,7 @@ func RunProposalOutboxDispatcher(ctx context.Context, db *sql.DB, service *Servi if err != nil { continue } - _ = dispatchOutboxEvents(ctx, dispatcher, events) + _ = dispatchOutboxEvents(ctx, db, dispatcher, events) } } } @@ -61,7 +61,7 @@ func RunResultOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service if err != nil { continue } - _ = dispatchOutboxEvents(ctx, dispatcher, events) + _ = dispatchOutboxEvents(ctx, db, dispatcher, events) } } } @@ -87,29 +87,56 @@ func RunStateOutboxDispatcher(ctx context.Context, db *sql.DB, service *Service) if err != nil { continue } - _ = dispatchOutboxEvents(ctx, dispatcher, events) + _ = dispatchOutboxEvents(ctx, db, dispatcher, events) } } } -func dispatchOutboxEvents(ctx context.Context, dispatcher *store.OutboxDispatcher, events []store.OutboxEvent) error { +func dispatchOutboxEvents(ctx context.Context, db *sql.DB, 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. + // + // A delivery failure does not abort the batch. Returning here meant one + // undeliverable payload -- reads are oldest-first -- was retried ahead of + // every later event of its type on every poll, forever. Instead the failure + // is counted against that row (dead-lettering it once exhausted) and the + // batch continues. + // + // Ordering within one aggregate is still honoured: once an event for a + // match fails, its later events are left for a subsequent poll so a client + // can never observe that match's newer state before its older state. Other + // aggregates are independent and proceed. + blocked := make(map[string]struct{}) + var firstErr error for _, event := range events { if event.EventID == "" { return fmt.Errorf("outbox event has no ID") } + if _, skip := blocked[event.AggregateID]; skip { + continue + } if err := dispatcher.Deliver(ctx, event); err != nil { - return err + blocked[event.AggregateID] = struct{}{} + if firstErr == nil { + firstErr = err + } + if db != nil { + if _, failErr := store.RecordOutboxDeliveryFailure(ctx, db, event.EventID, err, time.Now().UTC()); failErr != nil { + return failErr + } + } + continue } if err := dispatcher.Ack(ctx, event.EventID, time.Now().UTC()); err != nil { + // An ack failure is a database problem, not a payload problem; + // stop rather than counting it against the event. return err } } - return nil + return firstErr } func deliverProposalOutboxEvent(_ context.Context, event store.OutboxEvent, service *Service) error { diff --git a/server/api/outbox_test.go b/server/api/outbox_test.go index a8d00fda..7f0cdfe0 100644 --- a/server/api/outbox_test.go +++ b/server/api/outbox_test.go @@ -3,6 +3,8 @@ package api import ( "context" "encoding/json" + "errors" + "slices" "testing" "time" @@ -103,3 +105,53 @@ func TestDeliverStateOutboxEventRoutesLiveAbandonmentLifecycle(t *testing.T) { } } } + +// One malformed row used to abort the whole batch. Because reads are +// oldest-first and the row was never acknowledged, it was re-read ahead of +// every later event of its type on every 100ms poll -- blocking lifecycle +// delivery for all matches indefinitely, not just its own. +func TestDispatchOutboxEventsIsNotBlockedByOnePoisonRow(t *testing.T) { + delivered := []string{} + acked := []string{} + dispatcher := &store.OutboxDispatcher{ + Read: func(context.Context, int) ([]store.OutboxEvent, error) { return nil, nil }, + Deliver: func(_ context.Context, event store.OutboxEvent) error { + delivered = append(delivered, event.EventID) + if event.AggregateID == "match-poison" { + return errors.New("invalid state outbox payload") + } + return nil + }, + Ack: func(_ context.Context, eventID string, _ time.Time) error { + acked = append(acked, eventID) + return nil + }, + } + + events := []store.OutboxEvent{ + {EventID: "poison-1", AggregateID: "match-poison"}, + {EventID: "healthy-1", AggregateID: "match-healthy"}, + {EventID: "poison-2", AggregateID: "match-poison"}, + {EventID: "healthy-2", AggregateID: "match-other"}, + } + // nil db: the failure counter is exercised against a real PostgreSQL in + // the store integration tests; here we assert only batch progress. + err := dispatchOutboxEvents(context.Background(), nil, dispatcher, events) + if err == nil { + t.Fatal("expected the delivery failure to be reported to the caller") + } + + for _, eventID := range []string{"healthy-1", "healthy-2"} { + if !slices.Contains(acked, eventID) { + t.Fatalf("%s was not acknowledged; a poison row still blocks the batch (acked=%v)", eventID, acked) + } + } + if slices.Contains(acked, "poison-1") { + t.Fatal("a failed delivery must not be acknowledged") + } + // Ordering within the failing aggregate is preserved: poison-2 must wait + // so no client sees that match's newer state before its older state. + if slices.Contains(delivered, "poison-2") { + t.Fatalf("later event of a failed aggregate was delivered out of order: %v", delivered) + } +} diff --git a/server/migrations/0014_outbox_dead_letter.sql b/server/migrations/0014_outbox_dead_letter.sql new file mode 100644 index 00000000..061236fa --- /dev/null +++ b/server/migrations/0014_outbox_dead_letter.sql @@ -0,0 +1,15 @@ +ALTER TABLE outbox + ADD COLUMN delivery_attempts INTEGER NOT NULL DEFAULT 0, + ADD COLUMN last_delivery_error TEXT, + ADD COLUMN dead_lettered_at TIMESTAMPTZ; + +-- The unpublished dispatchers read oldest-first and previously stopped on the +-- first delivery error, so one permanently malformed payload blocked every +-- later event of that type forever. Dead-lettered rows leave the working set +-- via this partial index so a poison row degrades to one lost event instead of +-- a stalled queue. +DROP INDEX IF EXISTS outbox_unpublished_order; + +CREATE INDEX outbox_unpublished_order + ON outbox (created_at, event_id) + WHERE published_at IS NULL AND dead_lettered_at IS NULL; diff --git a/server/migrations/down/0014_outbox_dead_letter.sql b/server/migrations/down/0014_outbox_dead_letter.sql new file mode 100644 index 00000000..b899dcd9 --- /dev/null +++ b/server/migrations/down/0014_outbox_dead_letter.sql @@ -0,0 +1,10 @@ +DROP INDEX IF EXISTS outbox_unpublished_order; + +CREATE INDEX outbox_unpublished_order + ON outbox (created_at, event_id) + WHERE published_at IS NULL; + +ALTER TABLE outbox + DROP COLUMN IF EXISTS delivery_attempts, + DROP COLUMN IF EXISTS last_delivery_error, + DROP COLUMN IF EXISTS dead_lettered_at; diff --git a/server/store/allocation_match_sql.go b/server/store/allocation_match_sql.go index 8a9f9a45..c734ef51 100644 --- a/server/store/allocation_match_sql.go +++ b/server/store/allocation_match_sql.go @@ -5,7 +5,6 @@ import ( "context" "crypto/sha256" "database/sql" - "encoding/json" "fmt" "time" @@ -156,10 +155,7 @@ func AdvanceServerRegistration(ctx context.Context, db *sql.DB, binding domain.W 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, - }) + payload, err := MarshalStateChangedEnvelope(binding.MatchID, revision, string(to), now, playerIDs) if err != nil { return err } @@ -279,10 +275,7 @@ func BindAllocatedMatch(ctx context.Context, db *sql.DB, allocation domain.Alloc if err := rows.Close(); err != nil { return err } - payload, err := json.Marshal(map[string]any{ - "event": "state_changed", "revision": revision, "resource_id": allocation.MatchID, - "occurred_at": allocation.AllocatedAt, "state": string(domain.Allocating), "match_id": allocation.MatchID, "player_ids": playerIDs, - }) + payload, err := MarshalStateChangedEnvelope(allocation.MatchID, revision, string(domain.Allocating), allocation.AllocatedAt, playerIDs) if err != nil { return err } diff --git a/server/store/initial_connect_sql.go b/server/store/initial_connect_sql.go index d63c3c40..527501d1 100644 --- a/server/store/initial_connect_sql.go +++ b/server/store/initial_connect_sql.go @@ -156,7 +156,21 @@ func ApplyInitialConnectPlan(ctx context.Context, db *sql.DB, matchID, idempoten if err := tx.QueryRowContext(ctx, initialConnectMatchUpdateSQL, matchID, string(plan.MatchState)).Scan(&finalRevision); err != nil { return err } - payload, _ := json.Marshal(map[string]any{"match_id": matchID, "state": plan.MatchState, "action": plan.Action}) + // Every participant is told, not just the connected ones: a no-show + // needs to learn their ticket was marked NO_SHOW and a penalty applied. + // Publishing to a player with no live subscriber is a no-op. + recipients := make([]string, 0, len(participants)) + for _, participant := range participants { + recipients = append(recipients, participant.PlayerID) + } + payload, err := MarshalOutboxEnvelope(OutboxEnvelope{ + Event: "state_changed", ResourceID: matchID, Revision: finalRevision, + OccurredAt: now, State: string(plan.MatchState), MatchID: matchID, + PlayerIDs: recipients, Extra: map[string]any{"action": plan.Action}, + }) + if err != nil { + return err + } if _, err := tx.ExecContext(ctx, initialConnectOutboxSQL, "initial-connect:"+matchID+fmt.Sprintf(":%d", finalRevision), matchID, finalRevision, payload); err != nil { return err } diff --git a/server/store/live_abandonment_sql.go b/server/store/live_abandonment_sql.go index 089b2ff4..4135ca00 100644 --- a/server/store/live_abandonment_sql.go +++ b/server/store/live_abandonment_sql.go @@ -3,7 +3,6 @@ package store import ( "context" "database/sql" - "encoding/json" "fmt" "time" @@ -160,10 +159,10 @@ func ApplyLiveAbandonments(ctx context.Context, db *sql.DB, matchID string, now if len(targets) == 0 { return fmt.Errorf("%w: live match has no active event targets", domain.ErrConflict) } - payload, err := json.Marshal(map[string]any{ - "event": "state_changed", "revision": revision, "resource_id": matchID, - "occurred_at": now, "state": domain.Live, "match_id": matchID, - "player_ids": targets, "abandoned_player_ids": abandonmentIDs(planned), + payload, err := MarshalOutboxEnvelope(OutboxEnvelope{ + Event: "state_changed", ResourceID: matchID, Revision: int64(revision), + OccurredAt: now, State: string(domain.Live), MatchID: matchID, PlayerIDs: targets, + Extra: map[string]any{"abandoned_player_ids": abandonmentIDs(planned)}, }) if err != nil { return err diff --git a/server/store/outbox.go b/server/store/outbox.go index 96c16f14..a1ba83bc 100644 --- a/server/store/outbox.go +++ b/server/store/outbox.go @@ -25,28 +25,28 @@ type OutboxEvent struct { const OutboxUnpublishedSelectSQL = `SELECT event_id, aggregate_type, aggregate_id, revision, event_type, payload, created_at, published_at FROM outbox -WHERE published_at IS NULL +WHERE published_at IS NULL AND dead_lettered_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' +WHERE published_at IS NULL AND dead_lettered_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' +WHERE published_at IS NULL AND dead_lettered_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' +WHERE published_at IS NULL AND dead_lettered_at IS NULL AND event_type = 'state_changed' ORDER BY created_at, event_id LIMIT $1` @@ -59,8 +59,65 @@ const OutboxMarkPublishedSQL = `UPDATE outbox SET published_at = $2 WHERE event_id = $1 AND published_at IS NULL` +// MaxOutboxDeliveryAttempts bounds how long one undeliverable row may hold up +// its event type. It must exceed any plausible transient outage of the local +// fan-out adapter, since a healthy event is retried through the same counter. +const MaxOutboxDeliveryAttempts = 20 + +// OutboxRecordFailureSQL increments the attempt counter and dead-letters the +// row once it is exhausted, in one statement so a crash between the two cannot +// leave a row that is retried forever. +const OutboxRecordFailureSQL = `UPDATE outbox +SET delivery_attempts = delivery_attempts + 1, + last_delivery_error = $2, + dead_lettered_at = CASE WHEN delivery_attempts + 1 >= $3 THEN $4 ELSE dead_lettered_at END +WHERE event_id = $1 AND published_at IS NULL AND dead_lettered_at IS NULL +RETURNING dead_lettered_at IS NOT NULL` + +const OutboxDeadLetteredCountSQL = `SELECT count(*) FROM outbox WHERE dead_lettered_at IS NOT NULL` + var ErrOutboxEventNotFound = fmt.Errorf("outbox event not found or already published") +// RecordOutboxDeliveryFailure notes one failed delivery attempt and reports +// whether the row was dead-lettered as a result. Callers should keep going to +// the next event: the whole point is that one poison row must not stall the +// others. +func RecordOutboxDeliveryFailure(ctx context.Context, db *sql.DB, eventID string, cause error, now time.Time) (bool, error) { + if db == nil || eventID == "" || now.IsZero() { + return false, fmt.Errorf("invalid outbox failure arguments") + } + message := "" + if cause != nil { + message = cause.Error() + } + if len(message) > 500 { + message = message[:500] + } + var deadLettered bool + err := db.QueryRowContext(ctx, OutboxRecordFailureSQL, eventID, message, MaxOutboxDeliveryAttempts, now).Scan(&deadLettered) + if err == sql.ErrNoRows { + // Published or already dead-lettered by another replica; nothing owed. + return false, nil + } + if err != nil { + return false, err + } + return deadLettered, nil +} + +// CountDeadLetteredOutboxEvents backs the deletion-lag/poison-row metric. A +// non-zero value means at least one lifecycle event was never delivered. +func CountDeadLetteredOutboxEvents(ctx context.Context, db *sql.DB) (int64, error) { + if db == nil { + return 0, fmt.Errorf("invalid outbox count arguments") + } + var count int64 + if err := db.QueryRowContext(ctx, OutboxDeadLetteredCountSQL).Scan(&count); err != nil { + return 0, err + } + return count, nil +} + type OutboxDelivery func(context.Context, OutboxEvent) error // OutboxDispatcher is the durable-to-transient bridge. Read and Ack are diff --git a/server/store/outbox_envelope.go b/server/store/outbox_envelope.go new file mode 100644 index 00000000..af3b4915 --- /dev/null +++ b/server/store/outbox_envelope.go @@ -0,0 +1,91 @@ +package store + +import ( + "encoding/json" + "fmt" + "time" +) + +// OutboxEnvelope is the payload shape the api package's outbox dispatchers +// decode. Every lifecycle writer used to inline its own map literal, and one of +// them (ApplyInitialConnectPlan) omitted event/resource_id/occurred_at/ +// player_ids entirely. Because delivery rejects a malformed row and the +// dispatcher reads oldest-first, that single row blocked every later +// state_changed event indefinitely. Constructing envelopes through one +// validating builder makes that failure impossible to reintroduce: a writer +// that forgets a required field now fails its own transaction instead of +// silently poisoning the queue. +type OutboxEnvelope struct { + Event string + ResourceID string + // Revision is int64 to match the BIGINT column and, more importantly, so + // the -1 "nothing matched" sentinel some CTEs return surfaces as an error + // here instead of wrapping to a huge uint64 in the payload. + Revision int64 + OccurredAt time.Time + State string + // MatchID is omitted when empty, matching the proposal_changed shape which + // carries no match. + MatchID string + // PlayerIDs is the authoritative recipient list. Delivery rejects an empty + // one, so a writer must resolve participants before building the envelope. + PlayerIDs []string + // Extra carries event-specific keys (for example abandoned_player_ids). It + // may not overwrite a reserved key. + Extra map[string]any +} + +var reservedEnvelopeKeys = map[string]struct{}{ + "event": {}, "resource_id": {}, "revision": {}, "occurred_at": {}, + "state": {}, "match_id": {}, "player_ids": {}, +} + +// MarshalOutboxEnvelope validates and encodes one envelope. The checks mirror +// exactly what api.deliverStateOutboxEvent and api.deliverProposalOutboxEvent +// require, so anything this accepts is deliverable. +func MarshalOutboxEnvelope(envelope OutboxEnvelope) ([]byte, error) { + if envelope.Event == "" || envelope.ResourceID == "" || envelope.State == "" || envelope.OccurredAt.IsZero() { + return nil, fmt.Errorf("invalid outbox envelope: missing event, resource, state or timestamp") + } + if envelope.Revision < 0 { + return nil, fmt.Errorf("invalid outbox envelope: negative revision for %s %s", envelope.Event, envelope.ResourceID) + } + if len(envelope.PlayerIDs) == 0 { + return nil, fmt.Errorf("invalid outbox envelope: no recipients for %s %s", envelope.Event, envelope.ResourceID) + } + seen := make(map[string]struct{}, len(envelope.PlayerIDs)) + for _, playerID := range envelope.PlayerIDs { + if playerID == "" { + return nil, fmt.Errorf("invalid outbox envelope: empty participant") + } + if _, exists := seen[playerID]; exists { + return nil, fmt.Errorf("invalid outbox envelope: duplicate participant %s", playerID) + } + seen[playerID] = struct{}{} + } + payload := map[string]any{ + "event": envelope.Event, "resource_id": envelope.ResourceID, + "revision": envelope.Revision, "occurred_at": envelope.OccurredAt, + "state": envelope.State, "player_ids": envelope.PlayerIDs, + } + if envelope.MatchID != "" { + payload["match_id"] = envelope.MatchID + } + for key, value := range envelope.Extra { + if _, reserved := reservedEnvelopeKeys[key]; reserved { + return nil, fmt.Errorf("invalid outbox envelope: %q is reserved", key) + } + payload[key] = value + } + return json.Marshal(payload) +} + +// MarshalStateChangedEnvelope is the common case: a match lifecycle transition +// fanned out to that match's participants. The resource and match are the same +// aggregate, which is what deliverStateOutboxEvent asserts. +func MarshalStateChangedEnvelope(matchID string, revision int64, state string, occurredAt time.Time, playerIDs []string) ([]byte, error) { + return MarshalOutboxEnvelope(OutboxEnvelope{ + Event: "state_changed", ResourceID: matchID, Revision: revision, + OccurredAt: occurredAt, State: state, MatchID: matchID, PlayerIDs: playerIDs, + }) +} diff --git a/server/store/outbox_envelope_test.go b/server/store/outbox_envelope_test.go new file mode 100644 index 00000000..6556fea6 --- /dev/null +++ b/server/store/outbox_envelope_test.go @@ -0,0 +1,88 @@ +package store + +import ( + "encoding/json" + "testing" + "time" +) + +func TestMarshalStateChangedEnvelopeProducesDeliverableShape(t *testing.T) { + now := time.Date(2026, 9, 5, 12, 0, 0, 0, time.UTC) + payload, err := MarshalStateChangedEnvelope("match-1", 7, "LIVE", now, []string{"player-a", "player-b"}) + if err != nil { + t.Fatalf("marshal: %v", err) + } + // Decode with exactly the struct api.deliverStateOutboxEvent uses, then + // apply exactly its acceptance predicate. This is the regression guard for + // the initial-connect payload that omitted every one of these keys. + 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(payload, &envelope); err != nil { + t.Fatalf("decode: %v", err) + } + if envelope.Event != "state_changed" || envelope.ResourceID != "match-1" || envelope.Revision != 7 || + envelope.State == "" || len(envelope.PlayerIDs) == 0 { + t.Fatalf("envelope would be rejected by the dispatcher: %+v", envelope) + } + if envelope.MatchID != "match-1" || !envelope.OccurredAt.Equal(now) { + t.Fatalf("unexpected envelope: %+v", envelope) + } +} + +func TestMarshalOutboxEnvelopeRejectsUndeliverablePayloads(t *testing.T) { + now := time.Date(2026, 9, 5, 12, 0, 0, 0, time.UTC) + valid := OutboxEnvelope{ + Event: "state_changed", ResourceID: "match-1", Revision: 1, + OccurredAt: now, State: "LIVE", MatchID: "match-1", PlayerIDs: []string{"player-a"}, + } + if _, err := MarshalOutboxEnvelope(valid); err != nil { + t.Fatalf("baseline envelope must be valid: %v", err) + } + + for name, mutate := range map[string]func(*OutboxEnvelope){ + "no event": func(e *OutboxEnvelope) { e.Event = "" }, + "no resource": func(e *OutboxEnvelope) { e.ResourceID = "" }, + "no state": func(e *OutboxEnvelope) { e.State = "" }, + "no timestamp": func(e *OutboxEnvelope) { e.OccurredAt = time.Time{} }, + "no recipients": func(e *OutboxEnvelope) { e.PlayerIDs = nil }, + "empty recipient": func(e *OutboxEnvelope) { e.PlayerIDs = []string{"player-a", ""} }, + "duplicate recipient": func(e *OutboxEnvelope) { e.PlayerIDs = []string{"player-a", "player-a"} }, + // -1 is the "nothing matched" sentinel several CTEs return. Untyped as + // uint64 it would become 18446744073709551615 in the payload. + "sentinel revision": func(e *OutboxEnvelope) { e.Revision = -1 }, + "reserved extra": func(e *OutboxEnvelope) { e.Extra = map[string]any{"state": "CANCELLED"} }, + } { + t.Run(name, func(t *testing.T) { + envelope := valid + mutate(&envelope) + if _, err := MarshalOutboxEnvelope(envelope); err == nil { + t.Fatalf("expected %s to be rejected at construction", name) + } + }) + } +} + +func TestMarshalOutboxEnvelopeOmitsMatchIDForProposals(t *testing.T) { + now := time.Date(2026, 9, 5, 12, 0, 0, 0, time.UTC) + payload, err := MarshalOutboxEnvelope(OutboxEnvelope{ + Event: "proposal_changed", ResourceID: "proposal-1", Revision: 0, + OccurredAt: now, State: "OPEN", PlayerIDs: []string{"player-a"}, + }) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var decoded map[string]any + if err := json.Unmarshal(payload, &decoded); err != nil { + t.Fatalf("decode: %v", err) + } + if _, present := decoded["match_id"]; present { + t.Fatalf("proposal envelope must not carry a match_id: %s", payload) + } +} diff --git a/server/store/postgres_integration_test.go b/server/store/postgres_integration_test.go index 7a8564f4..29d6ec76 100644 --- a/server/store/postgres_integration_test.go +++ b/server/store/postgres_integration_test.go @@ -1792,8 +1792,11 @@ func TestPostgreSQLMigrationsRollBackAndReapplyCleanly(t *testing.T) { // Roll back every migration one at a time, in reverse, checking each // down file actually undoes what its forward file created — not just // that Rollback returns nil. - if err := migrations.Rollback(context.Background(), db, dir, 7); err != nil { - t.Fatalf("rollback 0013 through 0007: %v", err) + // This count is the number of migrations above 0006, so it must grow with + // every new migration; otherwise the later fixed-count rollbacks below + // silently target the wrong files. + if err := migrations.Rollback(context.Background(), db, dir, 8); err != nil { + t.Fatalf("rollback 0014 through 0007: %v", err) } var hasInitialConnectReadyColumn bool if err := db.QueryRow(`SELECT count(*) > 0 FROM information_schema.columns WHERE table_name = 'matches' AND column_name = 'initial_connect_ready_at'`).Scan(&hasInitialConnectReadyColumn); err != nil { diff --git a/server/store/proposal_sql.go b/server/store/proposal_sql.go index e3aa3ae2..b3b0136c 100644 --- a/server/store/proposal_sql.go +++ b/server/store/proposal_sql.go @@ -3,7 +3,6 @@ package store import ( "context" "database/sql" - "encoding/json" "fmt" "time" @@ -54,9 +53,9 @@ func CreateProposal(ctx context.Context, db *sql.DB, proposal domain.Proposal, t } players = append(players, participant.PlayerID) } - payload, err := json.Marshal(map[string]any{ - "event": "proposal_changed", "revision": uint64(0), "resource_id": proposal.ProposalID, - "occurred_at": now, "state": string(proposal.State), "player_ids": players, + payload, err := MarshalOutboxEnvelope(OutboxEnvelope{ + Event: "proposal_changed", ResourceID: proposal.ProposalID, Revision: 0, + OccurredAt: now, State: string(proposal.State), PlayerIDs: players, }) if err != nil { return err