nats

package
v0.0.0-...-9204231 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 11, 2026 License: AGPL-3.0 Imports: 40 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// MaxMessageRetries is the maximum number of processing attempts before a
	// message is routed to the dead-letter queue.
	MaxMessageRetries = 3

	// DLQSubjectPrefix is prepended to the original subject when publishing
	// failed messages to the dead-letter stream.
	DLQSubjectPrefix = "dlq."

	// Exponential backoff delays for retry attempts. Each delay corresponds
	// to the wait time before Nack'ing the message at that attempt number.
	// Watermill's Message interface does not support NakWithDelay, so the
	// backoff is implemented as a sleep before Nack.
	//
	// These delays are intentionally kept short (max 10s) because the sleep
	// blocks the consumer goroutine and holds the per-entity lock for the
	// entire duration. A 2-minute sleep (the previous RetryDelay3) would
	// block all events for that entity for 2 minutes, which is unacceptable
	// for subscription lifecycle events that trigger Remnawave provisioning.
	// If longer backoff is needed in the future, move to a scheduled re-queue
	// pattern (write a retry record with attempt_at, poll in background).
	RetryDelay1 = 2 * time.Second
	RetryDelay2 = 5 * time.Second
	RetryDelay3 = 10 * time.Second
)

Dead-letter queue and retry constants.

View Source 
const (

	// ConsumerDrainTimeout is the maximum duration Drain waits for in-flight
	// messages to finish processing before returning. This bounds the shutdown
	// delay when handlers are blocked on slow external calls (e.g. Remnawave).
	ConsumerDrainTimeout = 30 * time.Second

	// DLQDepthPollInterval is how often the consumer queries JetStream for
	// the current DLQ stream message count and updates the DLQDepth gauge.
	DLQDepthPollInterval = 30 * time.Second

	// DefaultMessageProcessingTimeout bounds the time for processing a single
	// event. If a handler exceeds this timeout (e.g., due to a hung DB query
	// or unresponsive Remnawave API), the context is cancelled, the handler
	// returns an error, and the message is Nack'd for retry.
	//
	// Default: 60 seconds — covers DB lookup + Remnawave API call + retry.
	// Configure via BillingEventConsumer.messageTimeout for environments
	// with slower upstream APIs.
	DefaultMessageProcessingTimeout = 60 * time.Second
)

Consumer lifecycle constants.

View Source 
const (
	// OutboxRelayBaseInterval is the starting poll interval. The relay
	// doubles this on each empty batch up to OutboxRelayMaxInterval, and
	// resets to base on any non-empty batch.
	OutboxRelayBaseInterval = 1 * time.Second

	// OutboxRelayMaxInterval caps exponential backoff so idle polling never
	// exceeds this frequency.
	OutboxRelayMaxInterval = 30 * time.Second

	// OutboxRelayBackoffMultiplier doubles the interval on each empty poll.
	OutboxRelayBackoffMultiplier = 2

	// OutboxRelayBatchSize is the base number of events fetched per tick.
	// During burst load the relay dynamically scales this up to
	// OutboxRelayMaxBatchSize.
	OutboxRelayBatchSize = 100

	// OutboxRelayMaxBatchSize is the upper bound for dynamic batch scaling.
	// When consecutive batches are full, the relay doubles the batch size
	// each tick up to this cap.
	OutboxRelayMaxBatchSize = 500

	// OutboxCleanupInterval is how often the relay purges old published events.
	OutboxCleanupInterval = 1 * time.Hour

	// OutboxRetentionPeriod is how long published events are kept before deletion.
	OutboxRetentionPeriod = 7 * 24 * time.Hour
)

Outbox relay constants control polling frequency, batch size, and retention.

View Source 
const (
	// HeaderOutboxSequence carries the outbox sequence number for ordering
	// verification and debugging on the consumer side.
	HeaderOutboxSequence = "X-Outbox-Sequence"

	// HeaderEventType carries the domain event type so consumers can inspect
	// it without deserialising the payload.
	HeaderEventType = "X-Event-Type"

	// HeaderTraceParent carries the W3C traceparent value extracted from the
	// domain event payload. Consumers use this header (via OTel's
	// TextMapPropagator) to link their processing spans to the originating
	// business operation's trace. The header name follows the W3C Trace
	// Context specification.
	HeaderTraceParent = "traceparent"
)

NATS message metadata header keys for outbox relay.

View Source 
const (
	// MinOutboxRelayWorkers is the lower bound for worker count to ensure at
	// least one goroutine always processes the outbox.
	MinOutboxRelayWorkers = 1

	// MaxOutboxRelayWorkers caps the number of parallel relay goroutines to
	// prevent exhausting the database connection pool.
	MaxOutboxRelayWorkers = 16
)
View Source 
const (
	StreamIdentity  = "IDENTITY"
	StreamBilling   = "BILLING"
	StreamRemnawave = "REMNAWAVE"
	StreamPayment   = "PAYMENT"
	StreamInfra     = "INFRA"
	StreamReseller  = "RESELLER"
	StreamPlugins   = "PLUGINS"
	StreamDLQ       = "DLQ"
)

Stream name constants identify each JetStream stream in the platform.

View Source 
const (
	RetentionDay   = 24 * time.Hour
	RetentionWeek  = 7 * RetentionDay
	RetentionMonth = 30 * RetentionDay

	// RetentionQuarter is 90 days — used for DLQ where failed events must
	// survive long enough for investigation and manual replay.
	RetentionQuarter = 90 * RetentionDay
)

Retention duration constants used by stream configurations.

View Source 
const DedupWindow = 60 * time.Minute

DedupWindow is the JetStream message deduplication window.

Set to 60 minutes to cover relay lag scenarios:

  • If MarkPublishedBatch succeeds but TX COMMIT fails, events are re-published on next tick. Dedup prevents consumer duplicates.
  • If the relay falls behind by more than DedupWindow, duplicates MAY reach consumers. Consumer-side IdempotencyChecker provides additional protection for billing events.

Memory cost: ~64 bytes per msg ID.

  • At 10K events/hour = ~640KB/hour.
  • At 100K events/hour = ~6.4MB -- well within NATS server capacity.

If the relay consistently lags > 60 minutes, this indicates a systemic issue that should trigger the OutboxBackpressureThreshold alert, not be masked by a larger dedup window.

The default JetStream dedup window (2 minutes) is far too short for the outbox relay circuit breaker backoff (up to 60s) plus deploy time.

View Source 
const MaxPluginAsyncRetries = 3

MaxPluginAsyncRetries is the maximum number of processing attempts before a plugin async message is routed to the dead-letter queue.

View Source 
const MaxReconnects = -1

MaxReconnects is set to -1 so the client retries indefinitely.

View Source 
const OutboxBacklogPollInterval = 30 * time.Second

OutboxBacklogPollInterval is how often the relay queries the outbox table for the count of unpublished events and updates the OutboxUnpublishedCount gauge.

Suggested Prometheus alert:

  • alert: OutboxBacklogGrowing expr: delta(platform_outbox_unpublished_count[5m]) > 100 for: 10m
View Source 
const OutboxBackpressureThreshold int64 = 10000

OutboxBackpressureThreshold is the number of unpublished events that triggers a backpressure warning. When exceeded, an error is logged each poll cycle and the OutboxBackpressureTriggered counter is incremented. This is observability-only — events are never rejected.

Suggested Prometheus alert:

  • alert: OutboxBackpressure expr: increase(platform_outbox_backpressure_triggered_total[5m]) > 0 for: 5m
View Source 
const ReconciliationCheckInterval = 5 * time.Minute

ReconciliationCheckInterval is how often the reconciliation check runs. Set to 5 minutes to balance observability with query overhead.

Suggested Prometheus alert:

  • alert: OutboxReconciliationGap expr: max(platform_outbox_reconciliation_sequence_gap) > 100 for: 15m

Variables

View Source 
var Module = fx.Module("nats",
	fx.Provide(NewConnection),
	fx.Provide(NewEventPublisher),
	fx.Provide(
		fx.Annotate(
			func(conn *nc.Conn) health.Checker {
				return NewHealthChecker(conn)
			},
			fx.ResultTags(`group:"health.checkers"`),
		),
	),
	fx.Invoke(EnsureStreams),
	fx.Invoke(registerMetrics),
)

Module provides the NATS connection, event publisher, stream provisioning, and Prometheus metrics collector to the Fx dependency graph.

Functions

func BillingSubscriptionSubjects 

func BillingSubscriptionSubjects() []string

BillingSubscriptionSubjects returns the NATS subjects this consumer listens to. Includes billing lifecycle events, traffic lifecycle events, and payment events that trigger billing-side effects (checkout completion). Exported for architecture tests that verify catalog-vs-consumer subscription completeness.

func EnsureStreams 

func EnsureStreams(conn *nc.Conn, logger *slog.Logger) error

EnsureStreams creates or updates every JetStream stream the platform needs. The operation is idempotent: existing streams whose configuration matches are left untouched, and those that differ are updated in place.

func NewConnection 

func NewConnection(lc fx.Lifecycle, cfg *config.Config, logger *slog.Logger) (*nc.Conn, error)

NewConnection dials the NATS server described in cfg and registers lifecycle hooks to close the connection on shutdown.

func StreamConfigs 

func StreamConfigs() []jetstream.StreamConfig

StreamConfigs returns every JetStream stream configuration the platform requires. EnsureStreams iterates this slice on startup to create or update each stream idempotently.

Types

type BillingEventConsumer 

type BillingEventConsumer struct {
	// contains filtered or unexported fields
}

BillingEventConsumer subscribes to billing domain events on NATS and routes them to the SubscriptionEventHandler (MultiSubOrchestrator) for Remnawave provisioning and deprovisioning.

Correctness guarantees:

  • Per-event-instance idempotency: events are deduplicated by the domain event ID (UUIDv7), not Watermill message UUID. This allows legitimate repeated operations on the same aggregate to be processed independently.
  • Per-entity ordering: events for the same entity are processed serially via entityLocks, while different entities run concurrently.
  • Retry + DLQ: failed messages are retried up to MaxMessageRetries times; permanently failing messages are sent to the dead-letter queue.

Event version handling: all events are normalized to the current schema version via SchemaRegistry.Upcast before processing. This allows producers to evolve payloads (adding fields, renaming) while consumers always work with the latest version.

To add a new event version:

  1. Create a struct implementing domainevent.Upcaster in the domain aggregate package
  2. Register it in the SchemaRegistry (internal/app/wiring_nats.go)
  3. Bump DefaultEventVersion if ALL producers now emit the new version
  4. Old events in NATS will be automatically upcasted on consumption.

Events with unknown versions are not rejected — they pass through unchanged and are processed best-effort with a warning log.

func NewBillingEventConsumer 

func NewBillingEventConsumer(
	subscriber *EventSubscriber,
	handler SubscriptionEventHandler,
	checkout CheckoutCompleter,
	plans multisub.PlanProvider,
	subs multisub.SubscriptionProvider,
	idempotency IdempotencyChecker,
	publisher *EventPublisher,
	schemaRegistry *domainevent.SchemaRegistry,
	logger *slog.Logger,
	clk clock.Clock,
	metrics *observability.Metrics,
	runner txmanager.Runner,
	conn *nc.Conn,
) *BillingEventConsumer

NewBillingEventConsumer creates a BillingEventConsumer with the given dependencies. The publisher is used to route permanently failed messages to the dead-letter queue. Plan and subscription data are resolved through multisub domain ports (PlanProvider + SubscriptionProvider). The checkout completer handles payment.charge_completed events by completing the billing checkout flow. The schema registry upcasts old event payloads to the latest version before processing. The NATS connection is used for DLQ depth polling via JetStream API.

func (*BillingEventConsumer) Drain 

func (c *BillingEventConsumer) Drain()

Drain waits for all in-flight message handlers to complete, bounded by ConsumerDrainTimeout. Call Drain after cancelling the context passed to Start so that consumeLoop stops reading new messages while existing handlers finish. If the timeout expires, a warning is logged — this is a best-effort mechanism that avoids blocking shutdown indefinitely.

func (*BillingEventConsumer) Start 

func (c *BillingEventConsumer) Start(ctx context.Context) error

Start subscribes to billing subscription events and processes them in background goroutines. It returns immediately; the goroutines run until the context is cancelled.

type BillingEventPublisher 

type BillingEventPublisher struct {
	// contains filtered or unexported fields
}

BillingEventPublisher adapts the EventPublisher to the domainevent.Publisher interface, routing billing domain events to the appropriate NATS topics.

func NewBillingEventPublisher 

func NewBillingEventPublisher(publisher *EventPublisher) *BillingEventPublisher

NewBillingEventPublisher creates a BillingEventPublisher backed by the given NATS EventPublisher.

func (*BillingEventPublisher) Publish 

func (p *BillingEventPublisher) Publish(ctx context.Context, event domainevent.Event) error

Publish routes a domain event to the correct NATS topic based on its type. The topic is derived directly from the event type string (e.g. "invoice.created" -> NATS subject "invoice.created").

func (*BillingEventPublisher) PublishBatch 

func (p *BillingEventPublisher) PublishBatch(ctx context.Context, events []domainevent.Event) error

PublishBatch publishes events sequentially to NATS. This is used for direct NATS publish (plugin async, internal notifications) — NOT for domain events which go through the transactional outbox. Partial publish is possible on error; callers should treat this as best-effort notification, not guaranteed delivery.

type BillingSubscriptionLookup 

type BillingSubscriptionLookup struct {
	// contains filtered or unexported fields
}

BillingSubscriptionLookup implements the multisub domain ports (PlanProvider + SubscriptionProvider) by delegating to billing domain read-only interfaces. It bridges the NATS consumer's enrichment needs with the billing bounded context and serves as the Anti-Corruption Layer that translates billing types into multisub-local types.

func NewBillingSubscriptionLookup 

func NewBillingSubscriptionLookup(
	subs billing.SubscriptionReader,
	plans billing.PlanReader,
	families billing.FamilyReader,
) *BillingSubscriptionLookup

NewBillingSubscriptionLookup creates a BillingSubscriptionLookup with the given billing readers.

func (*BillingSubscriptionLookup) GetFamilyMemberIDs 

func (l *BillingSubscriptionLookup) GetFamilyMemberIDs(ctx context.Context, ownerID string) ([]string, error)

GetFamilyMemberIDs fetches the user IDs of family members for the given owner. Returns nil (not an error) if no family group exists for the owner.

func (*BillingSubscriptionLookup) GetPlanSnapshot 

func (l *BillingSubscriptionLookup) GetPlanSnapshot(ctx context.Context, id string) (multisub.PlanSnapshot, error)

GetPlanSnapshot fetches a billing plan by ID and translates it into the multisub Anti-Corruption Layer type. This is the boundary where billing domain types are converted into multisub-local types.

func (*BillingSubscriptionLookup) GetSubscriptionInfo 

func (l *BillingSubscriptionLookup) GetSubscriptionInfo(ctx context.Context, id string) (multisub.SubscriptionInfo, error)

GetSubscriptionInfo fetches minimal subscription data for event enrichment. It satisfies multisub.SubscriptionProvider.

type CheckoutCompleter 

type CheckoutCompleter interface {
	CompleteCheckout(ctx context.Context, invoiceID string) error
}

CheckoutCompleter abstracts the billing context's CompleteCheckout operation. The BillingEventConsumer uses this to complete checkout asynchronously when it receives a payment.charge_completed event from the payment context.

type DLQLogConsumer 

type DLQLogConsumer struct {
	// contains filtered or unexported fields
}

DLQLogConsumer subscribes to the DLQ stream and logs every message at ERROR level with structured fields for centralized logging and alerting visibility. It is a read-only observer -- it acknowledges messages after logging but does not replay or modify them. The consumer runs as a background goroutine started via Start and exits when the context is cancelled.

This consumer is complementary to the DLQ depth gauge (polled by BillingEventConsumer.pollDLQDepth) and the DLQ replay tool (ReplayDLQMessages). Together they provide three layers of DLQ observability:

  • Depth gauge: how many messages are pending
  • Log consumer: structured ERROR log per message for alerting pipelines
  • Replay tool: manual or automated re-processing of DLQ messages

func NewDLQLogConsumer 

func NewDLQLogConsumer(
	subscriber *EventSubscriber,
	logger *slog.Logger,
) *DLQLogConsumer

NewDLQLogConsumer creates a DLQLogConsumer with the given dependencies.

func (*DLQLogConsumer) Start 

func (c *DLQLogConsumer) Start(ctx context.Context) error

Start subscribes to the DLQ stream and processes messages in a background goroutine. It returns immediately; the goroutine runs until the context is cancelled or the subscription channel is closed.

type DLQPayload 

type DLQPayload struct {
	OriginalSubject string `json:"original_subject"`
	OriginalPayload string `json:"original_payload"`
	Error           string `json:"error"`
	MsgID           string `json:"msg_id"`
	FailedAt        string `json:"failed_at"`
	RetryCount      int    `json:"retry_count"`
	EntityID        string `json:"entity_id,omitempty"`
	EventType       string `json:"event_type,omitempty"`
}

DLQPayload is the JSON envelope written to dead-letter queue topics.

type EventPublisher 

type EventPublisher struct {
	// contains filtered or unexported fields
}

EventPublisher wraps a Watermill NATS publisher to provide a simple JSON-based publish API on top of JetStream.

func NewEventPublisher 

func NewEventPublisher(conn *nc.Conn) (*EventPublisher, error)

NewEventPublisher creates an EventPublisher backed by the given NATS connection with JetStream enabled and automatic stream provisioning.

func (*EventPublisher) Close 

func (p *EventPublisher) Close() error

Close shuts down the underlying Watermill publisher.

func (*EventPublisher) Publish 

func (p *EventPublisher) Publish(ctx context.Context, topic string, payload any) error

Publish serializes payload to JSON and publishes it to the given topic.

Context handling: ctx is accepted for domainevent.Publisher interface compatibility but is not used to cancel the underlying NATS publish. PubAck timeout is managed by NATS client options configured in NewConnection (default 5s). The outbox relay checks ctx.Done() between batches, providing cancellation at the batch boundary.

A goroutine-based ctx.Done() wrapper was considered and rejected: if the context is cancelled mid-publish, the wrapper returns immediately but the background goroutine continues until PubAck timeout, leaking a goroutine per cancelled publish. The NATS client timeout is the correct cancellation mechanism for individual publishes.

func (*EventPublisher) PublishRaw 

func (p *EventPublisher) PublishRaw(topic string, msg *message.Message) error

PublishRaw publishes a pre-serialized Watermill message to a topic.

func (*EventPublisher) PublishWithID 

func (p *EventPublisher) PublishWithID(ctx context.Context, id string, topic string, payload []byte) error

PublishWithID serializes payload to JSON and publishes it with a deterministic message ID. When TrackMsgId is enabled on the publisher, Watermill uses the message UUID as the JetStream Nats-Msg-Id header, enabling server-side deduplication of retransmissions.

Context handling: see Publish godoc for rationale. ctx is accepted for interface compatibility; PubAck timeout is controlled by NATS client options.

type EventSubscriber 

type EventSubscriber struct {
	// contains filtered or unexported fields
}

EventSubscriber wraps a Watermill NATS subscriber to provide JetStream-backed subscriptions with durable consumer groups.

func NewEventSubscriber 

func NewEventSubscriber(conn *nc.Conn, consumerGroup string) (*EventSubscriber, error)

NewEventSubscriber creates an EventSubscriber using the given NATS connection. consumerGroup identifies the durable consumer group so that multiple instances of the same service share work.

func (*EventSubscriber) Close 

func (s *EventSubscriber) Close() error

Close shuts down the underlying Watermill subscriber.

func (*EventSubscriber) Subscribe 

func (s *EventSubscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

Subscribe returns a channel of Watermill messages for the given topic.

type HealthChecker 

type HealthChecker struct {
	// contains filtered or unexported fields
}

HealthChecker implements health.Checker for the NATS connection.

func NewHealthChecker 

func NewHealthChecker(conn *nc.Conn) *HealthChecker

NewHealthChecker returns a NATS health checker that reports status based on the underlying connection state.

func (*HealthChecker) HealthCheck 

func (c *HealthChecker) HealthCheck(_ context.Context) health.ComponentCheck

HealthCheck inspects the NATS connection status and returns the component health. A reconnecting state is reported as degraded rather than unhealthy because the client is actively recovering.

type IdempotencyChecker 

type IdempotencyChecker interface {
	// TryAcquire returns true if the key is new, false if it was already seen.
	TryAcquire(ctx context.Context, key string) (bool, error)

	// Release removes an idempotency key so that a redelivered message can be
	// processed again. This MUST be called when event processing fails,
	// otherwise the redelivered message will be silently skipped as a
	// duplicate.
	Release(ctx context.Context, key string) error

	// IncrementRetry atomically increments and returns the retry count for
	// the given key. The count is persisted in the database so it survives
	// NATS redeliveries (Watermill metadata is lost across Nack cycles).
	IncrementRetry(ctx context.Context, key string) (int, error)
}

IdempotencyChecker provides event-level deduplication and retry tracking. The adapter layer owns this interface; the postgres.IdempotencyRepository satisfies it.

Keys are the domain event ID (UUIDv7), unique per event instance. This deduplicates at the event level rather than the transport level (Watermill message UUID). Events without an ID (backward compat) use "{event_type}:{entity_id}" as the key.

type MetricsCollector 

type MetricsCollector struct {
	// contains filtered or unexported fields
}

MetricsCollector implements prometheus.Collector and reports NATS connection statistics on every Prometheus scrape. Metrics are fetched lazily — no background goroutine needed.

func NewMetricsCollector 

func NewMetricsCollector(conn *nc.Conn) *MetricsCollector

NewMetricsCollector returns a collector that exposes NATS connection stats.

func (*MetricsCollector) Collect 

func (c *MetricsCollector) Collect(ch chan<- prometheus.Metric)

Collect fetches current NATS connection stats and sends them as metrics. Uses conn.Stats() which returns a race-safe copy of the statistics.

func (*MetricsCollector) Describe 

func (c *MetricsCollector) Describe(ch chan<- *prometheus.Desc)

Describe sends the metric descriptors to the channel.

type MultiSubEventPublisher 

type MultiSubEventPublisher struct {
	// contains filtered or unexported fields
}

MultiSubEventPublisher adapts the EventPublisher to the domainevent.Publisher interface, routing multisub domain events to the appropriate NATS topics.

func NewMultiSubEventPublisher 

func NewMultiSubEventPublisher(publisher *EventPublisher) *MultiSubEventPublisher

NewMultiSubEventPublisher creates a MultiSubEventPublisher backed by the given NATS EventPublisher.

func (*MultiSubEventPublisher) Publish 

func (p *MultiSubEventPublisher) Publish(ctx context.Context, event domainevent.Event) error

Publish routes a domain event to the correct NATS topic based on its type.

func (*MultiSubEventPublisher) PublishBatch 

func (p *MultiSubEventPublisher) PublishBatch(ctx context.Context, events []domainevent.Event) error

PublishBatch publishes events sequentially to NATS. This is used for direct NATS publish (plugin async, internal notifications) — NOT for domain events which go through the transactional outbox. Partial publish is possible on error; callers should treat this as best-effort notification, not guaranteed delivery.

type OutboxHealthChecker 

type OutboxHealthChecker struct {
	// contains filtered or unexported fields
}

OutboxHealthChecker implements health.Checker for the transactional outbox. It reports degraded when the unpublished event backlog exceeds the backpressure threshold, indicating the relay is falling behind.

func NewOutboxHealthChecker 

func NewOutboxHealthChecker(outbox *postgres.OutboxRepository) *OutboxHealthChecker

NewOutboxHealthChecker returns an outbox health checker.

func (*OutboxHealthChecker) HealthCheck 

func (c *OutboxHealthChecker) HealthCheck(ctx context.Context) health.ComponentCheck

HealthCheck queries the unpublished event count and returns degraded when the backlog exceeds the backpressure threshold.

type OutboxRelay 

type OutboxRelay struct {
	// contains filtered or unexported fields
}

OutboxRelay polls the transactional outbox table for unpublished domain events and forwards them to NATS via the EventPublisher. It runs as a background goroutine managed by the Fx lifecycle.

Row locking: each relay batch runs inside a database transaction with FOR UPDATE SKIP LOCKED, ensuring multiple relay instances never process the same rows concurrently.

Startup catch-up: Run executes one immediate relay pass before entering the ticker loop, so events stuck from a prior crash are forwarded without waiting for the first tick.

Delivery guarantee: at-least-once with documented edge cases.

Normal flow: GetUnpublished -> NATS Publish -> MarkPublishedBatch -> Commit. JetStream deduplication (1-hour window) prevents duplicate processing.

Edge case 1 (most common): Commit fails after NATS publish succeeded. Events are re-published on next tick. JetStream Msg-Id deduplication ensures consumers see each event exactly once.

Edge case 2 (rare): MarkPublishedBatch succeeds but PG commit fails (network partition). Events are marked published in PG (on reconnect PG may have committed the transaction), but NATS may not have received them. This is at-most-once for this specific failure mode. Mitigation: reconciliation check compares outbox sequence with JetStream last sequence, and the relay_failed mechanism catches permanently stuck events. The ReconciliationCheck method detects this gap via observability.

Consumers must be idempotent regardless of delivery guarantee.

Circuit breaker: the relay wraps NATS publishes in a circuit breaker. When NATS is unreachable, the breaker opens after relayCBConsecutiveFailures consecutive failures and the relay skips DB polling entirely until the breaker transitions to half-open after relayCBTimeout. This prevents wasteful DB locks and log spam during NATS outages.

JetStream deduplication: each message is published with the outbox event ID as the Watermill message UUID. Because TrackMsgId is enabled on the publisher, Watermill sets the Nats-Msg-Id header to this UUID, enabling server-side deduplication of retransmissions after transaction rollbacks.

func NewOutboxRelay 

func NewOutboxRelay(
	outbox *postgres.OutboxRepository,
	publisher *EventPublisher,
	txRunner txmanager.Runner,
	clk clock.Clock,
	logger *slog.Logger,
	workerCount int,
	cbCfg circuitbreaker.Config,
	metrics *observability.Metrics,
	conn *nc.Conn,
) *OutboxRelay

NewOutboxRelay creates an OutboxRelay with the given dependencies. workerCount controls the number of parallel relay goroutines; values below MinOutboxRelayWorkers are clamped to MinOutboxRelayWorkers. cbCfg configures the NATS circuit breaker; pass circuitbreaker.DefaultConfigNoInterval() for defaults matching the previously hardcoded values. metrics may be nil; metric recording is skipped when nil (safe for tests). conn is the NATS connection used by ReconciliationCheck to query JetStream stream state; it may be nil if reconciliation is not needed.

func (*OutboxRelay) ReconciliationCheck 

func (r *OutboxRelay) ReconciliationCheck(ctx context.Context) error

ReconciliationCheck compares per-stream JetStream LastSeq deltas between reconciliation intervals. A stream whose LastSeq stops growing while the relay is actively publishing to its subjects indicates a delivery gap.

The previous implementation summed LastSeq across all streams into a single number and compared it against the outbox sequence. This was an apples-to-oranges comparison: the outbox uses a single autoincrement sequence spanning all event types, while each JetStream stream has its own independent LastSeq counter. A quiet stream could mask a gap in a busy stream (and vice versa).

Per-stream tracking eliminates this class of false negatives by reporting each stream's delta independently.

On the first invocation, all previous values are zero, so the current state becomes the baseline and no gap is reported. This prevents false alarms at startup.

This is observability only -- it does not correct or republish events. The per-stream gap metric (platform_outbox_reconciliation_sequence_gap with a "stream" label) can be used to trigger manual investigation when it remains non-zero for an extended period.

func (*OutboxRelay) Run 

func (r *OutboxRelay) Run(ctx context.Context)

Run spawns workerCount relay goroutines plus a single cleanup goroutine. Each worker independently polls the outbox table with FOR UPDATE SKIP LOCKED, so rows are never processed by more than one worker. Run blocks until the context is cancelled and all goroutines have exited.

An immediate relay pass is executed by each worker on startup to catch up on any events that were written but not yet relayed before a previous shutdown or crash.

type PluginAsyncConsumer 

type PluginAsyncConsumer struct {
	// contains filtered or unexported fields
}

PluginAsyncConsumer subscribes to plugin.hook.* topics on NATS JetStream and dispatches each message to the appropriate plugin runtime handlers. Messages that fail processing are retried up to MaxPluginAsyncRetries times with exponential backoff before being sent to the dead-letter queue.

func NewPluginAsyncConsumer 

func NewPluginAsyncConsumer(
	subscriber *EventSubscriber,
	dispatcher *plugin.HookDispatcher,
	runtime *plugin.RuntimePool,
	publisher *EventPublisher,
	clk clock.Clock,
	logger *slog.Logger,
) *PluginAsyncConsumer

NewPluginAsyncConsumer creates a consumer that bridges NATS async hook events to the plugin runtime pool. The publisher is used to route permanently failed messages to the dead-letter queue.

func (*PluginAsyncConsumer) Start 

func (c *PluginAsyncConsumer) Start(ctx context.Context) error

Start begins consuming async hook events from the "plugin.hook.>" subject. It blocks until the context is cancelled.

type ReplayOptions 

type ReplayOptions struct {
	// Limit caps the number of messages replayed. Zero means replay all available messages.
	Limit int

	// FilterSubject restricts replay to messages whose OriginalSubject contains
	// this substring. Empty means all subjects match.
	FilterSubject string

	// FilterEntity restricts replay to messages whose EntityID equals this value
	// exactly. Empty means all entities match.
	FilterEntity string

	// DryRun when true collects matching DLQ payloads without re-publishing them.
	DryRun bool

	// DelayBetween inserts a pause between consecutive re-publishes for rate
	// limiting. Zero means no delay.
	DelayBetween time.Duration
}

ReplayOptions configures how DLQ messages are replayed.

type ReplayResult 

type ReplayResult struct {
	// Replayed is the number of messages successfully re-published (or matched in dry-run mode).
	Replayed int

	// Skipped is the number of messages that did not match the filter criteria.
	Skipped int

	// Errors is the number of messages that failed to re-publish. The replay
	// stops on the first publish error; this count will be 0 or 1.
	Errors int

	// Matched holds the DLQ payloads that were replayed (or would be replayed
	// in dry-run mode). This is always populated so callers can inspect what
	// was processed.
	Matched []DLQPayload
}

ReplayResult summarises the outcome of a DLQ replay operation.

func ReplayDLQMessages 

func ReplayDLQMessages(ctx context.Context, subscriber *EventSubscriber, publisher *EventPublisher, opts ReplayOptions) (ReplayResult, error)

ReplayDLQMessages reads messages from the DLQ stream and re-publishes the original payload to the original subject, filtered and rate-limited according to the provided options.

Messages that cannot be unmarshalled are acknowledged and skipped to prevent poison pills from blocking the replay. If a re-publish fails, the DLQ message is Nack'd so it remains available for a subsequent replay attempt.

When DryRun is true, matching messages are collected and acknowledged without re-publishing. The original MsgID from the DLQ payload is preserved as the Watermill message UUID to enable JetStream server-side deduplication on replay.

type SubscriptionEventHandler 

type SubscriptionEventHandler interface {
	OnSubscriptionActivated(
		ctx context.Context,
		subscriptionID string,
		platformUserID string,
		plan multisub.PlanSnapshot,
		addonIDs []string,
		familyMemberIDs []string,
	) error
	OnSubscriptionCancelled(ctx context.Context, subscriptionID string) error
	OnSubscriptionPaused(ctx context.Context, subscriptionID string) error
	OnSubscriptionResumed(ctx context.Context, subscriptionID string) error
	OnBindingTrafficExceeded(ctx context.Context, bindingID, subscriptionID string, usedBytes, limitBytes int64) error
	OnBindingTrafficReset(ctx context.Context, bindingID, subscriptionID string) error
	OnTrafficWarning(ctx context.Context, bindingID, subscriptionID string, usedBytes, limitBytes int64, thresholdPct int)
}

SubscriptionEventHandler defines the contract for handling billing subscription lifecycle events. The MultiSubOrchestrator satisfies this interface, keeping the NATS adapter decoupled from the multisub domain.

Plan data is passed as multisub.PlanSnapshot (an Anti-Corruption Layer type) so that the handler never depends on billing/aggregate types.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.