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, }) }