CosmicClash/server/store/outbox_envelope.go

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