messaging

package
v0.66.0 Latest Latest
Warning

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

 Go to latest Published: Sep 19, 2026 License: MIT Imports: 40 Imported by: 0

Documentation

Overview

Package messaging provides a unified interface for message queue operations. It abstracts the underlying messaging implementation to allow for easy testing and future extensibility.

Index

Constants

View Source 
const (
	ExchangeTypeDirect  = amqp.ExchangeDirect
	ExchangeTypeTopic   = amqp.ExchangeTopic
	ExchangeTypeFanout  = amqp.ExchangeFanout
	ExchangeTypeHeaders = amqp.ExchangeHeaders
)

AMQP 0-9-1 core exchange types for ExchangeDeclaration.Type, spelled direct, topic, fanout and headers (case-sensitive). Validate admits these and any "x-" plugin type.

View Source 
const (

	// QueueTypeQuorum and QueueTypeClassic are the x-queue-type values a
	// DeadLetterSpec.QueueType may name; an empty QueueType resolves to
	// QueueTypeQuorum. Any other value fails Validate.
	QueueTypeQuorum  = "quorum"
	QueueTypeClassic = "classic"
)

AMQP declaration/consume argument keys, shared by the declaration constructors in helpers.go, the validators in declarations.go, and the consumer session in registry.go.

View Source 
const (
	SealTenancyDisabled  = sealruntime.TenancyDisabled
	SealTenancyShared    = sealruntime.TenancyShared
	SealTenancyPerTenant = sealruntime.TenancyPerTenant
)
View Source 
const HeaderEventID = "x-outbox-event-id"

HeaderEventID is the AMQP header the outbox relay stamps with the event id a consumer dedups on. outbox.HeaderEventID aliases it; the constant lives here because Metadata.DedupKey reads it and outbox imports this package.

View Source 
const SealTagName = "seal"

SealTagName is the struct-tag key of the sealing family. Spelled here rather than imported from jose/sealed: the probe must not pull the codec into every build, and jose must not import messaging. messaging/sealed pins the two spellings together.

View Source 
const TenantStampHeader = tenantstamp.Header

TenantStampHeader is the header the framework writes the publishing context's tenant into, and the one a consumer reads it back from.

Variables

View Source 
var (
	// ErrNotConnected is returned by a publish when the client is not connected to the broker.
	ErrNotConnected = errors.New("not connected to AMQP broker")
	// ErrShutdown is returned when a publish is interrupted by client shutdown. The outbox
	// relay treats this (and context.Canceled) as a shutdown abort that must NOT advance an
	// event's retry_count — it is the one publish error the relay branches on.
	ErrShutdown = errors.New("AMQP client is shutting down")
	// ErrPublishRetriesExhausted is returned by publishBytes once the bounded retry loop
	// reaches maxPublishAttempts. It wraps the last attempt's cause (one of ErrPublishNacked,
	// ErrPublishConfirmTimeout, or the raw publish error) so a caller can see WHY it gave up.
	ErrPublishRetriesExhausted = errors.New("amqp: publish retries exhausted")
	// ErrPublishNacked is the cause when the broker negatively acknowledged a publish. A
	// basic.nack on a publish-confirm is a transient broker condition (disk alarm, mirror
	// resync, failover), not a statement that the message is bad. These cause sentinels are
	// informational (logging / direct-publisher branching) — the outbox relay does NOT
	// classify on them (see ADR-033); it retries every publish failure.
	ErrPublishNacked = errors.New("amqp: publish nacked by broker")
	// ErrPublishConfirmTimeout is the cause when a confirmed publish never received an
	// ACK/NACK within connectionTimeout.
	ErrPublishConfirmTimeout = errors.New("amqp: publish confirmation timed out")
)
View Source 
var (
	// ErrPayloadUndecodable reports a message body that could not be decoded into
	// the consumer's payload type. Match it with errors.Is.
	ErrPayloadUndecodable = errors.New("messaging: payload could not be decoded")

	// ErrPayloadInvalid reports a payload that decoded but failed struct
	// validation. Match it with errors.Is.
	ErrPayloadInvalid = errors.New("messaging: payload failed validation")

	// ErrPayloadOpenRefused reports a sealed message the opener refused —
	// poison, like the two above; a provisioning-recoverable refusal
	// (SEAL_KID_UNKNOWN_GENERATION) still takes this path and keeps its code.
	// Match it with errors.Is.
	ErrPayloadOpenRefused = errors.New("messaging: sealed payload refused by the opener")
)
View Source 
var ErrInvalidEventID = errors.New("messaging: event id is outside the ledger grammar [A-Za-z0-9_-]{1,128}")

ErrInvalidEventID is returned when an event id headed for the inbox ledger is outside ^[A-Za-z0-9_-]{1,128}$ — absent, empty, over 128 bytes, or carrying any other byte. Match it with errors.Is. The wrapped message names the byte LENGTH, never the id: the value is publisher-controlled and this error reaches logs and spans. A handler returning it takes the standard poison path (nack without requeue → DLQ); the remedy is to re-mint conforming ids or to move the producer to the sealed typed door.

View Source 
var ErrInvalidPublishDestination = errors.New("amqp: publish destination exceeds the AMQP shortstr limit")

ErrInvalidPublishDestination is returned when a caller-supplied field that the AMQP wire format carries as a shortstr cannot fit one. Two doors return it: Publish/PublishToExchange, for the exchange and routing key of the basic.publish METHOD frame and the header keys of the CONTENT-HEADER frame beside it; and Declarations.Validate, for a declared name or routing key, which fails startup rather than the first publish. Match it with errors.Is; the wrapped message names the FIELD and its byte length, never the value — an over-long destination is usually built from request data, and this error reaches logs and spans.

The publish is refused outright rather than retried: the frame is unwritable whatever the broker's state, so a retry only re-tears the connection it just brought back. ErrPublishRetriesExhausted is deliberately NOT involved.

View Source 
var ErrManagerClosed = errors.New("messaging: manager closed")

ErrManagerClosed is returned by Manager's EnsureConsumers and Publisher methods once Close has been called, rather than resurrecting a consumer or publisher on a shut-down manager (backlog F22). Publisher additionally returns it from a zero-value Manager that was never built via NewMessagingManager. Callers can use errors.Is(err, ErrManagerClosed) to distinguish "manager is gone" from a per-key failure and decide whether to abort or fall back to a non-messaging path.

View Source 
var ErrNotSealTagged = errors.New("messaging: Seal on a type that carries no seal tags; a plain event goes to the outbox as a struct payload")

ErrNotSealTagged is returned by Publisher.Seal for a type that carries no seal tags.

View Source 
var ErrPublishDoorUnavailable = errors.New("messaging: client carries no byte publish door; publish through a framework-built client")

ErrPublishDoorUnavailable is returned when a publish is attempted through a client that carries no byte door: one built by app.Options.MessagingClientFactory, a hand-written AMQPClient, or a testing/mocks double. Match it with errors.Is. The remedy is to publish through a framework-built client (deps.Messaging) or, in a test, to swap the handle for messaging/testing's capture publisher.

View Source 
var ErrSealingNotLinked = sealruntime.ErrNotLinked

ErrSealingNotLinked is the startup error for a seal-tagged type declared in a build that never imported messaging/sealed.

View Source 
var ErrTenantStampConflict = tenantstamp.ErrConflict

ErrTenantStampConflict reports a publish whose tenant stamp was supplied by the caller and disagrees with the one the framework resolved. The framework is the stamp's only writer: a caller-supplied value is an unauthenticated claim to act for a tenant, so the publish fails rather than being silently overwritten.

The streams lane re-exports this same value, so errors.Is holds across lanes.

Functions

func ConfigureSealing added in v0.63.0 

func ConfigureSealing(rt *SealRuntime)

ConfigureSealing records the runtime facts sealing needs. The app calls it once before collecting declarations; a seal-tagged declaration collected before it fails Validate.

func IsSealTagged added in v0.63.0 

func IsSealTagged(t reflect.Type) bool

IsSealTagged reports whether t (pointers unwrapped) is a struct that carries a `seal` tag anywhere the typed publish door would refuse to publish as plaintext: on its own fields or the members an untagged embedded struct promotes (the depth jose/sealed.ScanType inspects), OR misplaced on a named nested field or a tagged embed (which DeclareTypedPublisher refuses outright). It is a probe, not a scan: the codec judges the declaration; the probe must say yes wherever the codec would speak or the door would fail closed. It is the one detector every lane guard shares — the typed publish door, the streams typed declarations (which refuse a sealed T in v1) and the outbox door (which refuses a sealed struct payload) — so a struct is "sealed" in exactly one way.

The outbox door asks on every Publish, so the answer is memoized per Go type: the set of payload types a process publishes is small and fixed, and a type's tags never change.

func IsSealedDelivery added in v0.63.0 

func IsSealedDelivery(ctx context.Context) bool

IsSealedDelivery reports whether ctx belongs to a delivery the sealed typed door opened — the framework's own marker, unreachable from a header or from consumer code. True when the context carries that delivery's sealed DedupKey. The ledger door cross-checks a Sealed DedupKey against it by equality.

The marker travels with the handler's context: a handler that calls inbox.ProcessOnce from a goroutine or with a context NOT derived from the one it was handed (context.Background() instead of context.WithoutCancel(ctx)) loses it and gets ErrInvalidEventID — fail closed. Derive the context.

func RegisterSealCodec added in v0.63.0 

func RegisterSealCodec(c SealCodec)

RegisterSealCodec installs the sealing codec. A blank import of messaging/sealed does this from init; a second registration panics.

func ResolveTenantStamp added in v0.63.0 

func ResolveTenantStamp(ctx context.Context, headers map[string]any) (string, error)

ResolveTenantStamp is the publish-side stamp rule for a writer that persists a message for later delivery instead of publishing it now — the outbox, whose Publish must snapshot the tenant while the originating context is still live, the same way it snapshots the trace keys (ADR-087 §3).

It applies the exact rule stampingPublisher applies: a caller-supplied TenantStampHeader in headers is refused with ErrTenantStampConflict (equal value included), then the context tenant is the stamp. There is no pool key to fall back on, so the context is the only source; an empty stamp with a nil error means no tenant is in play and nothing must be written.

func ValidateDedupKey added in v0.63.0 

func ValidateDedupKey(ctx context.Context, key DedupKey) error

ValidateDedupKey checks a key at the ledger door. Admission is by the key's provenance, not its spelling: the zero DedupKey is refused, and a Sealed key is admitted only when ctx carries a sealed delivery key AND that key equals the one being validated. A key retained from delivery A is therefore refused while handling delivery B, whose jti differs, as is a sealed key under a plain context. Only the sealed branch of Metadata.DedupKey mints a Sealed key, so a caller can only hold a key some sealed delivery composed; the equality check binds it to the one in hand. A wire key passes under either context; its grammar ran when WireDedupKey built it.

The two sealed refusals are reported apart, because they send an operator to different code: ctx carries no sealed delivery at all (a lost marker — a detached goroutine, context.Background()), or it carries one whose key differs (a key held past its delivery). Every refusal wraps ErrInvalidEventID and names only its arm: a sealed key spells a jti, so neither the offered key nor the bound one is ever rendered.

func ValidateEventID added in v0.63.0 

func ValidateEventID(id string) error

ValidateEventID checks id against the ledger grammar. WireDedupKey runs it, so every wire key — Metadata.DedupKey's, on the header and on the message_id property alike, or one a consumer builds — passed it at construction.

func ValidatePublishDestination added in v0.61.0 

func ValidatePublishDestination(exchange, routingKey string, headers map[string]any) error

ValidatePublishDestination checks every caller-supplied shortstr a publish puts on the wire: the exchange and routing key, which travel in the basic.publish METHOD frame, and every header KEY, which travels in the CONTENT-HEADER frame that follows it (the same frame carrying CorrelationId, which ADR-070 already guards). One operation, two frames, one ceiling (nested tables included — a table's keys are shortstrs at every depth).

Length only. The charset is deliberately NOT checked: unlike the consume side (ADR-070, `[C60.17]`), where the value is a foreign publisher's, these are the service's OWN destinations, and a broker that dislikes one answers with a CHANNEL error — recoverable, and not the connection-wide failure this guard exists to prevent. Empty is legal: the default exchange and a fanout binding both use it.

It is exported for callers that record a destination now and publish it later — the outbox writes exchange, routing key and header keys to a ledger row, and a row the frame can never carry is better refused at the INSERT than parked by the relay after MaxRetries. They run the rule rather than restating the ceiling.

func ValidatePublishEventType added in v0.64.0 

func ValidatePublishEventType(eventType string) error

ValidatePublishEventType checks one caller-supplied event type against the same ceiling, for callers that record an event type now and publish it later. It leaves as the `type` property of the CONTENT-HEADER frame (ADR-105), so it is a shortstr on the same frame as the header keys ValidatePublishDestination judges — and it fails that frame, and with it the shared Connection, the same way.

It exists as a door of its own because the property travels beside a destination the recorder already validated: the outbox writes exchange, routing key and event type to one ledger row, and its EventType column bounds 255 of whatever the vendor counts — PostgreSQL `VARCHAR(255)` counts characters, Oracle `VARCHAR2(255)` counts bytes by default and characters under CHAR semantics — so on PostgreSQL, and on a CHAR-semantics Oracle, a multibyte type the column accepts can still exceed 255 BYTES. Same reasoning as the destination door — a row the frame can never carry is better refused at the INSERT than parked by the relay after MaxRetries.

Length only, and the error names the field and its byte size, never the value.

Types

type AMQPClient 

type AMQPClient interface {
	Client

	// ConsumeFromQueue consumes messages from a queue with specific options.
	ConsumeFromQueue(ctx context.Context, options ConsumeOptions) (<-chan amqp.Delivery, error)

	// DeclareQueue declares a queue from the given declaration.
	// ctx is checked before the broker operation (amqp091 declares are not context-aware on the wire).
	// The declaration's Args carries optional AMQP queue arguments (x-dead-letter-exchange, x-queue-type, ...).
	DeclareQueue(ctx context.Context, queue *QueueDeclaration) error

	// DeclareExchange declares an exchange from the given declaration (ctx: pre-flight check, as DeclareQueue).
	DeclareExchange(ctx context.Context, exchange *ExchangeDeclaration) error

	// BindQueue binds a queue to an exchange from the given declaration (ctx: pre-flight check, as DeclareQueue).
	BindQueue(ctx context.Context, binding *BindingDeclaration) error
}

AMQPClient extends the basic Client interface with AMQP-specific functionality. This allows for more advanced AMQP features while maintaining the simple interface. It is the module-facing type — what ModuleDeps.Messaging and scheduler.JobContext.Messaging return — and carries no byte publish door: the framework's own client and its stamping wrapper implement an unexported one that Publisher[T].Publish and the outbox relay reach (ADR-096).

type AMQPClientImpl 

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

AMQPClientImpl provides an AMQP implementation of the messaging client interface. It includes automatic reconnection, retry logic, and AMQP-specific features.

func NewAMQPClient 

func NewAMQPClient(brokerURL string, log logger.Logger, opts ...ClientOption) *AMQPClientImpl

NewAMQPClient creates a new AMQP client instance. It automatically attempts to connect to the broker and handles reconnections. Optional Option values (e.g. WithConnectionTimeout) override the defaults.

func (*AMQPClientImpl) BindQueue 

func (c *AMQPClientImpl) BindQueue(ctx context.Context, binding *BindingDeclaration) error

BindQueue binds a queue to an exchange from the given declaration (ctx: pre-flight check, see DeclareQueue).

func (*AMQPClientImpl) Close 

func (c *AMQPClientImpl) Close() error

Close gracefully shuts down the AMQP client.

func (*AMQPClientImpl) Consume 

func (c *AMQPClientImpl) Consume(ctx context.Context, destination string) (<-chan amqp.Delivery, error)

Consume starts consuming messages from the specified destination (queue name).

func (*AMQPClientImpl) ConsumeFromQueue 

func (c *AMQPClientImpl) ConsumeFromQueue(_ context.Context, options ConsumeOptions) (<-chan amqp.Delivery, error)

ConsumeFromQueue consumes messages from a queue with specific options.

func (*AMQPClientImpl) DeclareExchange 

func (c *AMQPClientImpl) DeclareExchange(ctx context.Context, exchange *ExchangeDeclaration) error

DeclareExchange declares an exchange from the given declaration (ctx: pre-flight check, see DeclareQueue).

func (*AMQPClientImpl) DeclareQueue 

func (c *AMQPClientImpl) DeclareQueue(ctx context.Context, queue *QueueDeclaration) error

DeclareQueue declares a queue from the given declaration. ctx is honored as a pre-flight check: amqp091 declare/bind operations are not context-aware on the wire, so a canceled context fails fast before the call.

func (*AMQPClientImpl) IsReady 

func (c *AMQPClientImpl) IsReady() bool

IsReady returns true if the client is connected and ready to send/receive messages.

type BindingDeclaration 

type BindingDeclaration struct {
	Queue      string         // Queue name
	Exchange   string         // Exchange name
	RoutingKey string         // Routing key pattern
	NoWait     bool           // Do not wait for server confirmation
	Args       map[string]any // Additional arguments
}

BindingDeclaration defines a queue-to-exchange binding

func NewBinding added in v0.14.1 

func NewBinding(queue, exchange, routingKey string) *BindingDeclaration

NewBinding creates a binding declaration between a queue and exchange.

Parameters:

  • queue: Queue name to bind
  • exchange: Exchange name to bind to
  • routingKey: Routing key pattern (e.g., "order.*", "user.created")

type BrokerURLProvider added in v0.26.0 

type BrokerURLProvider interface {
	// BrokerURL returns the AMQP broker URL for the given key.
	// For single-tenant apps, key will be "". For multi-tenant, key will be the tenant ID.
	BrokerURL(ctx context.Context, key string) (string, error)
}

BrokerURLProvider provides per-key AMQP configurations. This interface abstracts where tenant-specific messaging configs come from.

type Client 

type Client interface {
	// Consume starts consuming messages from the specified destination.
	// Returns a channel that delivers messages and an error if consumption setup fails.
	// Messages should be acknowledged by the consumer.
	Consume(ctx context.Context, destination string) (<-chan amqp.Delivery, error)

	// Close gracefully shuts down the messaging client.
	// It should clean up all connections and resources.
	Close() error

	// IsReady returns true if the client is connected and ready to send/receive messages.
	IsReady() bool
}

Client defines the interface for messaging operations. It provides a simple API for consuming messages while hiding the complexity of connection management, retries, and protocol-specific details. It carries no publish method: a module publishes through the Publisher[T] handle that DeclareTypedPublisher returns, never by handing bytes to the client (ADR-096).

type ClientFactory added in v0.9.0 

type ClientFactory func(string, logger.Logger) AMQPClient

ClientFactory creates AMQP clients from URLs

type ClientOption added in v0.42.0 

type ClientOption func(*AMQPClientImpl)

ClientOption configures an AMQPClientImpl at construction time.

func WithAppName added in v0.64.0 

func WithAppName(name string) ClientOption

WithAppName sets the application identity the client stamps as the AMQP app_id property on every publish (ADR-105). The framework passes app.name here; a client built without it publishes with no app_id.

func WithConnectionTimeout added in v0.42.0 

func WithConnectionTimeout(d time.Duration) ClientOption

WithConnectionTimeout overrides the per-publish broker confirmation timeout — the wait for an ACK/NACK after a confirmed publish (see publishBytes). Non-positive values are ignored, leaving the 30s default in place.

func WithMaxPublishAttempts added in v0.45.0 

func WithMaxPublishAttempts(n int) ClientOption

WithMaxPublishAttempts bounds the per-publish retry loop: after n failed attempts publishBytes returns ErrPublishRetriesExhausted instead of retrying forever. Non-positive values are ignored, leaving the default (5).

func WithPublishTimeout added in v0.64.0 

func WithPublishTimeout(d time.Duration) ClientOption

WithPublishTimeout sets the aggregate per-publish bound: publishBytes runs the readiness pre-flight and the entire retry loop under a derived context deadline of d, layered under any tighter caller deadline (the shorter wins). A value below maxPublishAttempts x connectionTimeout deliberately lowers the effective retry count. Non-positive values are ignored, leaving the publish unbounded — the default.

func WithReadyTimeout added in v0.49.0 

func WithReadyTimeout(d time.Duration) ClientOption

WithReadyTimeout bounds publishBytes's pre-flight wait for a not-yet-ready client to become ready before the publish attempt begins (see waitForReady). The wait does not consume a maxPublishAttempts slot. Non-positive values are ignored, leaving the 5s default in place.

func WithReconnectDelay added in v0.49.0 

func WithReconnectDelay(d time.Duration) ClientOption

WithReconnectDelay overrides the base of the full-jitter reconnect backoff: each wait is a uniform random sample in [0, min(base*2^attempt, max)), so this is the first attempt's upper bound, not a minimum spacing between attempts. Non-positive values are ignored, leaving the 5s default in place.

func WithReconnectMaxDelay added in v0.49.0 

func WithReconnectMaxDelay(d time.Duration) ClientOption

WithReconnectMaxDelay overrides the ceiling of the connection-reconnect backoff. The consumer re-subscribe loop (registry.go) keeps its own fixed cap. Non-positive values are ignored, leaving the 60s default in place.

func WithReinitDelay added in v0.49.0 

func WithReinitDelay(d time.Duration) ClientOption

WithReinitDelay overrides the delay before channel reinitialization after failure. Non-positive values are ignored, leaving the 2s default in place.

func WithResendDelay added in v0.49.0 

func WithResendDelay(d time.Duration) ClientOption

WithResendDelay overrides the wait between retries after a channel-level publish error only; broker NACKs retry on the fixed 100ms nackBackoff and confirmation timeouts retry immediately. Non-positive values are ignored, leaving the 5s default in place.

type ConsumeOptions 

type ConsumeOptions struct {
	Queue         string // Queue name to consume from
	Consumer      string // Consumer tag
	AutoAck       bool   // Auto-acknowledge messages
	Exclusive     bool   // Exclusive consumer
	NoLocal       bool   // No-local flag
	NoWait        bool   // No-wait flag
	PrefetchCount int    // RabbitMQ prefetch count (0 or negative = default to 1)
	// Args are per-consumer arguments forwarded to basic.consume
	// (x-stream-offset, x-priority, ...).
	Args map[string]any
}

ConsumeOptions contains options for consuming messages with AMQP-specific features.

type ConsumerDeclaration 

type ConsumerDeclaration struct {
	Queue         string         // Queue to consume from
	Consumer      string         // Consumer tag
	AutoAck       bool           // Automatically acknowledge messages
	Exclusive     bool           // Exclusive consumer
	NoLocal       bool           // Do not deliver to the connection that published
	NoWait        bool           // Do not wait for server confirmation
	EventType     string         // Event type identifier
	Description   string         // Human-readable description
	Handler       MessageHandler // Message handler (optional for documentation-only declarations)
	Workers       int            // Number of concurrent workers (0 = auto-scale to NumCPU*4, >0 = explicit)
	PrefetchCount int            // RabbitMQ prefetch count (0 = auto-scale to Workers*10, capped at 500)
	Args          map[string]any // Per-consumer arguments forwarded to basic.consume (x-stream-offset, x-priority, ...)
	// TenantOptional lets this consumer run a delivery that carries no tenant stamp,
	// for a control-plane consumer whose events belong to no tenant. The default is
	// false — fail closed — and it never admits a stamp that is present but unusable.
	TenantOptional bool
}

ConsumerDeclaration defines what a module consumes and how to handle messages

func DeclareTypedConsumer added in v0.56.0 

func DeclareTypedConsumer[T any](decls *Declarations, opts *ConsumerOptions, fn func(context.Context, T) error) *ConsumerDeclaration

DeclareTypedConsumer registers a consumer whose handler is built by NewTypedHandler from fn, so T is inferred from fn and never spelled out.

The queue is not declared here: pass it to DeclareQueue or DeclareQueueWithDLQ separately, exactly as an untyped DeclareConsumer with a nil queue does. A consumer naming a queue nobody declared surfaces at Declarations.Validate() as "consumer references non-existent queue", not here.

It panics on a nil decls, a nil opts, or an opts that already carries a Handler — all three are declaration-time wiring mistakes, and the package already fails startup that way for duplicate consumer registrations.

Failure and concurrency semantics are NewTypedHandler's: decode and validation failures return a *PayloadError that nacks WITHOUT requeue, so pair the queue with DeclareQueueWithDLQ, and fn must be safe for concurrent use.

func DeclareTypedConsumerWithMeta added in v0.57.0 

func DeclareTypedConsumerWithMeta[T any](decls *Declarations, opts *ConsumerOptions, fn func(context.Context, T, Metadata) error) *ConsumerDeclaration

DeclareTypedConsumerWithMeta is DeclareTypedConsumer for a metadata-aware fn. Same panics, same queue rules, same PayloadError semantics.

For a seal-tagged T this is the sealed typed door (ADR-097): every delivery is opened — verified, tid-judged, decrypted — BEFORE decode and validation, a refused one is a *PayloadError at PayloadStageOpen (nacked without requeue), and fn's Metadata carries the verified envelope, so meta.Sealed() is (envelope, true) and meta.DedupKey() is a Sealed key spelled `<SignFamily>:<jti>`. A sealed consumer that cannot start — codec not linked, runtime not configured, no key material, a family missing a provisioned generation in its inherited role — fails Validate.

func NewConsumer added in v0.14.1 

func NewConsumer(opts *ConsumerOptions) *ConsumerDeclaration

NewConsumer creates a consumer declaration from options.

type ConsumerOptions added in v0.14.1 

type ConsumerOptions struct {
	Queue         string         // Queue name to consume from
	Consumer      string         // Consumer tag
	EventType     string         // Event type identifier
	Description   string         // Human-readable description
	Handler       MessageHandler // Message handler (optional for documentation-only declarations)
	AutoAck       bool           // Automatically acknowledge messages (default: false)
	Exclusive     bool           // Exclusive consumer (default: false)
	NoLocal       bool           // Don't deliver to the connection that published (default: false)
	Workers       int            // Number of concurrent workers (0 = auto-scale to NumCPU*4, >0 = explicit)
	PrefetchCount int            // RabbitMQ prefetch count (0 = auto-scale to Workers*10, capped at 500)
	Args          map[string]any // Per-consumer arguments forwarded to basic.consume (x-stream-offset, x-priority, ...)
	// TenantOptional lets this consumer run a delivery that carries no tenant stamp
	// (a control-plane consumer). Default false: a consumer that needs a tenant fails
	// the delivery closed rather than running the handler without one.
	TenantOptional bool
}

ConsumerOptions contains configuration for creating a consumer declaration.

type ConsumerState added in v0.65.0 

type ConsumerState struct {
	// Key is the manager key the consumer's registry was leased under: the tenant id
	// under per-tenant replay, "" for the control plane. Registry.ConsumerStates leaves
	// it empty — a registry does not know the key it was leased under.
	Key       string
	Queue     string
	Consumer  string // consumer tag
	EventType string
	// Subscribed flips false only once the session has fully ended: the handler
	// pool drains first, so a consumer whose delivery channel the broker already
	// closed still reads subscribed while its slowest handler runs.
	Subscribed        bool
	Resubscribes      uint64
	LastResubscribeAt time.Time // zero until the first successful re-subscribe
	// FailStreak counts failed re-subscribe attempts in the CURRENT outage only:
	// the next success clears it and a restarted consumer starts a fresh count, so
	// it never carries a previous outage's, or a previous session's, total.
	FailStreak int
}

ConsumerState is a snapshot of one declared consumer's subscription state.

The identity fields — Key, Queue, Consumer, EventType — are for an operator reading /_sys/health-debug or a caller of ConsumerStates, never for the unauthenticated /ready body or a log line: Manager.Stats() reduces these rows to counts precisely so no tenant key, queue name, consumer tag or event type leaves through them.

func (*ConsumerState) GivenUp added in v0.65.0 

func (s *ConsumerState) GivenUp() bool

GivenUp reports a consumer whose outage stopped looking like a routine flap: it is unsubscribed and its consecutive re-subscribe failures have reached consumerResubscribeWarnFromAttempt, the same threshold that escalates the re-subscribe log to WARN. The supervisor keeps retrying; this is the point at which the outage is worth reporting.

The receiver is a pointer because the identity fields make the struct too heavy to copy per call; a snapshot read out of a slice is addressable, so callers write states[i].GivenUp() unchanged.

type DeadLetterSpec added in v0.53.0 

type DeadLetterSpec struct {
	// Exchange is the dead-letter exchange name. Empty derives "<queue>.dlx".
	// The exchange is declared as a durable fanout so the parking queue
	// receives every dead-lettered message regardless of routing key.
	Exchange string

	// ParkingQueue is the queue that collects dead-lettered messages.
	// Empty derives "<queue>.dlq". Declared with the production defaults of
	// NewQueue (durable, non-exclusive).
	ParkingQueue string

	// RoutingKey, when non-empty, is set as x-dead-letter-routing-key so
	// dead-lettered messages are re-published with it instead of their
	// original routing key. Rarely needed with the fanout DLX default.
	RoutingKey string

	// QueueType is the x-queue-type both the primary and the parking queue are
	// declared with: QueueTypeQuorum (the default an empty value resolves to)
	// or QueueTypeClassic. Any other value fails Validate. A type already set
	// on either queue's Args wins — the helper never overwrites one.
	QueueType string
}

DeadLetterSpec configures the declarative dead-letter opt-in for a queue. Zero-value fields get derived defaults; see DeclareQueueWithDLQ.

type DeclarationStats added in v0.9.0 

type DeclarationStats struct {
	Exchanges  int
	Queues     int
	Bindings   int
	Publishers int
	Consumers  int
}

DeclarationStats holds counts of each declaration type.

type Declarations added in v0.9.0 

type Declarations struct {
	Exchanges  map[string]*ExchangeDeclaration
	Queues     map[string]*QueueDeclaration
	Bindings   []*BindingDeclaration
	Publishers []*PublisherDeclaration
	// contains filtered or unexported fields
}

Declarations stores messaging infrastructure declarations made by modules at startup. This is a pure data structure (no client dependencies) that can be validated once and replayed to multiple per-tenant registries.

func NewDeclarations added in v0.9.0 

func NewDeclarations() *Declarations

NewDeclarations creates a new empty declarations store.

func (*Declarations) Clone added in v0.9.0 

func (d *Declarations) Clone() *Declarations

Clone copies the declarations for per-tenant replay. Each declaration and its Args/Headers map is new, but the copy is one level deep: a map or slice stored as a VALUE inside one of those maps is shared with the original.

func (*Declarations) Consumers added in v0.9.0 

func (d *Declarations) Consumers() []*ConsumerDeclaration

Consumers returns all consumer declarations in registration order. Used internally by ReplayToRegistry and for observability/metrics.

func (*Declarations) DeclareBinding added in v0.14.1 

func (d *Declarations) DeclareBinding(queue, exchange, routingKey string) *BindingDeclaration

DeclareBinding creates and registers a binding in one step. RegisterBinding copies the Args map into a new declaration, so — as with DeclareQueue — the returned pointer is not the stored one and Args set on it afterwards are discarded. Build with NewBinding, set Args, then RegisterBinding.

Note that bindings are APPENDED, never merged by name — unlike queues and exchanges, which are map-backed and idempotent. Declaring one twice adds a second entry, which issues a redundant BindQueue at startup and changes Hash(), so a caller that may run more than once has to guard the call itself (DeclareQueueWithDLQ's hasParkingBinding is the in-tree precedent).

func (*Declarations) DeclareConsumer added in v0.14.1 

func (d *Declarations) DeclareConsumer(opts *ConsumerOptions, queue *QueueDeclaration) *ConsumerDeclaration

DeclareConsumer creates and registers a consumer in one step.

A non-nil queue is registered, merging with any existing declaration of the same name; an incompatible shape keeps the incumbent and becomes a startup conflict (see RegisterQueue). This hybrid approach allows consumers to optionally declare their dependencies.

Usage:

  • Pass nil if queue is already registered separately
  • Pass queue declaration to auto-register (convenience for simple cases)

RegisterConsumer copies Args into a new declaration, so the returned pointer is not the stored one: set Args on opts BEFORE this call. The consumer index is unexported and re-registering one key panics, so reaching the stored entry afterwards means d.Consumers(), which returns the stored pointers.

func (*Declarations) DeclareDirectExchange added in v0.66.0 

func (d *Declarations) DeclareDirectExchange(name string) *ExchangeDeclaration

DeclareDirectExchange creates and registers a direct exchange in one step. RegisterExchange copies the Args map into a new declaration, so — as with DeclareQueue — the returned pointer is not the stored one and Args set on it afterwards are discarded. Set them on d.Exchanges[name].Args, or build with NewDirectExchange and register that.

func (*Declarations) DeclarePublisher added in v0.14.1 

func (d *Declarations) DeclarePublisher(opts *PublisherOptions, exchange *ExchangeDeclaration) *PublisherDeclaration

DeclarePublisher creates and registers a publisher in one step.

A non-nil exchange is registered, merging with any existing declaration of the same name; an incompatible shape keeps the incumbent and becomes a startup conflict (see RegisterExchange). This hybrid approach allows publishers to optionally declare their dependencies.

Usage:

  • Pass nil if exchange is already registered separately
  • Pass exchange declaration to auto-register (convenience for simple cases)

RegisterPublisher copies Headers into a new declaration, so the returned pointer is not the stored one: set Headers on opts BEFORE this call. Publishers are stored in the d.Publishers slice rather than keyed by name, so reaching the stored entry afterwards means indexing that slice.

func (*Declarations) DeclareQueue added in v0.14.1 

func (d *Declarations) DeclareQueue(name string) *QueueDeclaration

DeclareQueue creates and registers a queue in one step. It registers IMMEDIATELY, and RegisterQueue copies Args into a new declaration, so the returned pointer is NOT the stored one: setting Args on it afterwards changes nothing and reports no error.

Use either of the following instead: 

q := NewQueue(name); q.Args["x-max-length"] = 1000; d.RegisterQueue(q)
d.DeclareQueue(name); d.Queues[name].Args["x-max-length"] = 1000

The copy is one level deep: the Args map is new, but a map or slice stored as a value inside it is still shared with the caller. The returned value is for reading the declaration's shape, not for editing it.

func (*Declarations) DeclareQueueWithDLQ added in v0.53.0 

func (d *Declarations) DeclareQueueWithDLQ(name string, dl *DeadLetterSpec) *QueueDeclaration

DeclareQueueWithDLQ declares a queue whose failed deliveries are parked instead of dropped: the framework's nack-without-requeue on handler error (see wiki/messaging.md) dead-letters into the spec's exchange, and the parking queue bound to it retains the message with the x-death header. Lowers to ordinary exchange/queue/binding declarations plus queue Args, so per-tenant replay, validation, and topology hashing behave as if declared by hand.

Both queues are declared with the spec's QueueType — quorum unless the spec or an existing declaration of either name says otherwise.

Returns the primary queue declaration. As with DeclareQueue, RegisterQueue copies Args into a new declaration, so Args set on the returned pointer afterwards are discarded — this helper works because it fills Args BEFORE registering. To add your own, set them on d.Queues[name].Args, or build the queue with NewQueue and register it (re-declaring one name merges compatible shapes).

func (*Declarations) DeclareStreamQueue added in v0.59.0 

func (d *Declarations) DeclareStreamQueue(name string, spec *StreamQueueSpec) *QueueDeclaration

DeclareStreamQueue declares a RabbitMQ stream queue (x-queue-type: stream): an append-only replicated log read non-destructively at a client-chosen offset, instead of a classic queue's destructive consume. Consumers pick a start position with the x-stream-offset consumer Arg (see wiki/messaging.md). A nil spec declares the queue with broker-default retention.

Returns the caller's declaration, NOT the registered one: RegisterQueue copies Args into a new declaration, so — as with DeclareQueue — Args set on the returned pointer afterwards are discarded. This helper works because it fills Args before registering; add your own via d.Queues[name].Args.

func (*Declarations) DeclareTopicExchange added in v0.14.1 

func (d *Declarations) DeclareTopicExchange(name string) *ExchangeDeclaration

DeclareTopicExchange creates and registers a topic exchange in one step. RegisterExchange copies the Args map into a new declaration, so — as with DeclareQueue — the returned pointer is not the stored one and Args set on it afterwards are discarded. Set them on d.Exchanges[name].Args, or build with NewTopicExchange and register that.

func (*Declarations) Hash added in v0.15.0 

func (d *Declarations) Hash() uint64

Hash generates a deterministic hash of all declarations. This is used by Manager to detect duplicate replay attempts (idempotency). The hash is stable across multiple calls and does not include handler functions.

func (*Declarations) IsEmpty added in v0.30.0 

func (d *Declarations) IsEmpty() bool

IsEmpty reports whether no declarations have been registered. Comparing the whole zero value keeps a declaration kind added to Stats() later from being forgotten here.

func (*Declarations) RegisterBinding added in v0.9.0 

func (d *Declarations) RegisterBinding(b *BindingDeclaration)

RegisterBinding adds a binding declaration to the store.

func (*Declarations) RegisterConsumer added in v0.9.0 

func (d *Declarations) RegisterConsumer(c *ConsumerDeclaration)

RegisterConsumer adds a consumer declaration to the store. Panics if a consumer with the same queue+consumer+event_type already exists.

func (*Declarations) RegisterExchange added in v0.9.0 

func (d *Declarations) RegisterExchange(e *ExchangeDeclaration)

RegisterExchange adds an exchange declaration to the store.

Re-declaring a name merges when the shapes are compatible, so which call ran last no longer decides the type the broker gets. An incompatible re-declaration keeps the incumbent and records a conflict that Validate reports, because declaration order across modules is invisible at any single call site (ADR-118). RegisterQueue's Args caveat applies here unchanged: the conflict is reported by rendering the contested values, so exchange Args are broker topology and must not carry secrets either.

func (*Declarations) RegisterPublisher added in v0.9.0 

func (d *Declarations) RegisterPublisher(p *PublisherDeclaration)

RegisterPublisher adds a publisher declaration to the store.

func (*Declarations) RegisterQueue added in v0.9.0 

func (d *Declarations) RegisterQueue(q *QueueDeclaration)

RegisterQueue adds a queue declaration to the store.

Re-declaring a name merges when the shapes are compatible, so DeclareQueueWithDLQ and DeclareQueue on one name compose instead of whichever ran last silently dropping the other's dead-letter args. An incompatible re-declaration keeps the incumbent and records a conflict that Validate reports, because declaration order across modules is invisible at any single call site. That report prints the contested Args values verbatim into a startup error, which the logger's key-based filter cannot mask — queue Args are broker topology and must not carry secrets.

func (*Declarations) ReplayToRegistry added in v0.9.0 

func (d *Declarations) ReplayToRegistry(reg RegistryInterface) error

ReplayToRegistry applies all declarations to a runtime registry. The order is important: exchanges first, then queues, then bindings, then publishers/consumers.

func (*Declarations) Stats added in v0.9.0 

func (d *Declarations) Stats() DeclarationStats

Stats returns counts of each declaration type.

func (*Declarations) Validate added in v0.9.0 

func (d *Declarations) Validate() error

Validate checks the integrity of all declarations. It ensures that references between declarations are valid.

type DedupKey added in v0.65.0 

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

DedupKey is an inbox ledger key carrying which door produced it. A wire key comes from WireDedupKey and has passed ValidateEventID's grammar; a sealed key is composed by the sealed typed door alone — no exported function mints one. The zero value is invalid and inbox.ProcessOnce refuses it.

func WireDedupKey added in v0.65.0 

func WireDedupKey(id string) (DedupKey, error)

WireDedupKey builds a ledger key from a wire-sourced or consumer-composed id, applying ValidateEventID's grammar at construction. A refused id returns the invalid zero DedupKey and an error wrapping ErrInvalidEventID. Its result is never Sealed, whatever the id spells.

func (DedupKey) Sealed added in v0.65.0 

func (k DedupKey) Sealed() bool

Sealed reports whether the sealed typed door produced this key.

func (DedupKey) String added in v0.65.0 

func (k DedupKey) String() string

String returns the key's persisted spelling: the wire id verbatim, or `<SignFamily>:<jti>` for a sealed key.

type EventPublisher added in v0.63.0 

type EventPublisher[T any] interface {
	Publish(ctx context.Context, client AMQPClient, evt T) error
}

EventPublisher is the seam a module depends on when it wants to swap the handle in a test: *Publisher[T] satisfies it, and so does the capture double in messaging/testing. Production code declares the handle with DeclareTypedPublisher and stores it behind this interface.

type ExchangeDeclaration 

type ExchangeDeclaration struct {
	Name       string         // Exchange name
	Type       string         // Exchange type: an ExchangeType* constant or an "x-" plugin type
	Durable    bool           // Survive server restart
	AutoDelete bool           // Delete when no longer used
	Internal   bool           // Internal exchange
	NoWait     bool           // Do not wait for server confirmation
	Args       map[string]any // Additional arguments
}

ExchangeDeclaration defines an exchange to be declared

func NewDirectExchange added in v0.66.0 

func NewDirectExchange(name string) *ExchangeDeclaration

NewDirectExchange creates a direct exchange with production-safe defaults. Direct exchanges route messages to bindings whose routing key matches exactly.

Production defaults:

  • Durable: true (survives broker restart)
  • AutoDelete: false (won't delete when unused)
  • Internal: false (can be published to directly)
  • NoWait: false (waits for broker confirmation)

func NewTopicExchange added in v0.14.1 

func NewTopicExchange(name string) *ExchangeDeclaration

NewTopicExchange creates a topic exchange with production-safe defaults. Topic exchanges route messages based on routing key patterns (e.g., "order.*", "user.#").

Production defaults:

  • Durable: true (survives broker restart)
  • AutoDelete: false (won't delete when unused)
  • Internal: false (can be published to directly)
  • NoWait: false (waits for broker confirmation)

type Manager added in v0.9.0 

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

Manager manages AMQP clients by string keys with different lifecycle strategies. Publishers are cached with idle eviction (can be recreated easily). Consumers are long-lived (must stay alive to receive messages). The manager is key-agnostic - it doesn't know about tenants, just manages named clients.

The publisher side is a thin adapter over internal/resourcepool.Pool, which owns the ADR-032 lease/evict/close protocol (seed leases, LRU eviction, idle cleanup, and the closed-pool guard). The manager keeps only the AMQP-specific client creation. The consumer side is long-lived and managed directly (not pool-shaped).

func NewMessagingManager added in v0.9.0 

func NewMessagingManager(resourceSource BrokerURLProvider, log logger.Logger, opts ManagerOptions, clientFactory ClientFactory) *Manager

NewMessagingManager creates a new messaging manager. The idle-publisher sweep starts here and stops in Close, so callers need not drive it (ADR-067); StartCleanup remains available and idempotent.

func (*Manager) AnyConsumerGivenUp added in v0.65.0 

func (m *Manager) AnyConsumerGivenUp() bool

AnyConsumerGivenUp reports whether any consumer on any tenant key has stopped being able to re-subscribe. It answers the readiness probe's question directly rather than through ConsumerStates: /ready asks on every poll, and a snapshot would copy every declared consumer's row — four identifier strings apiece — to compute one bool, carrying coordinates that must never reach the unauthenticated body into the package that renders it.

func (*Manager) Close added in v0.9.0 

func (m *Manager) Close() error

Close closes all clients and stops the idle-publisher sweep the constructor started. Publisher closes go through the pool (which stops its own cleanup loop and joins every per-publisher close failure); consumer closes are handled directly. A publisher client still borrowed by in-flight work is closed at its final release instead of by this call, and that deferred close failure (if any) is excluded from this return value — it is counted in Stats()["errors"] instead (wiki/migrations.md C581.3). Every failure returned here, from BOTH sides, is surfaced under the historical "errors closing messaging clients" prefix.

func (*Manager) ConsumerStates added in v0.65.0 

func (m *Manager) ConsumerStates() []ConsumerState

ConsumerStates returns the subscription state of every consumer declared on every tenant key that holds a consumer registry. Order is declaration order within a key; across keys it is map order.

func (*Manager) EnsureConsumers added in v0.9.0 

func (m *Manager) EnsureConsumers(ctx context.Context, key string, decls *Declarations) error

EnsureConsumers creates and starts consumers for the given key using the provided declarations. This should be called once per key to set up long-lived consumers. Subsequent calls for the same key are idempotent.

Concurrent calls for the same key collapse onto one setup pass, but each caller waits on ITS OWN context. DoChan (not Do) is what makes that possible: Do blocks uncancelably, so a caller whose budget was already spent still sat through the full setup — bounded by infraSetupTimeout (45s) while the realistic caller is a lazy first-touch request carrying a ~5s deadline. Giving up does NOT cancel the setup: ensureConsumersInternal runs on a WithoutCancel budget, so it completes and installs the consumers for whoever asks next.

One testing caveat: singleflight neither recovers nor forwards a runtime.Goexit from the shared call, so a test double that calls t.Fatal or require.* inside the setup path hangs every waiter. Use t.Errorf and return, as the repo's httptest handlers do.

func (*Manager) Publisher added in v0.19.0 

func (m *Manager) Publisher(ctx context.Context, key string) (AMQPClient, ReleaseFunc, error)

Publisher returns a publisher client for the given key plus a ReleaseFunc the caller must invoke when finished with it for the current unit of work (typically deferred). Publishers are cached with LRU eviction and lazy initialization; the lease prevents a publisher that is evicted while in use from being closed under an active caller (the #606 race). Once Close begins, Publisher fails closed rather than resurrecting a publisher (F22) — except a caller already mid-Publisher on a fresh client another borrower holds, who may still receive that live handle after Close returns; it closes exactly once, at its final release. On error the returned ReleaseFunc is nil — check err first.

func (*Manager) StartCleanup added in v0.9.0 

func (m *Manager) StartCleanup(interval time.Duration)

StartCleanup starts the background cleanup routine for idle publishers. A non-positive interval substitutes the documented 2-minute default. The constructor already started a sweep, so this is a no-op unless StopCleanup ran first (the pool's loop is single-instance).

func (*Manager) Stats added in v0.9.0 

func (m *Manager) Stats() map[string]any

Stats returns statistics about the messaging manager. Publisher counters come from the pool; the consumer counters come from the consumer map and the per-consumer subscription state its registries keep. consumer_registries counts tenant keys, not consumers, and consumer_max_fail_streak is the largest current-outage re-subscribe streak across them.

func (*Manager) StopCleanup added in v0.9.0 

func (m *Manager) StopCleanup()

StopCleanup stops the background cleanup routine

func (*Manager) StopConsumers added in v0.42.0 

func (m *Manager) StopConsumers()

StopConsumers stops every consumer registry from accepting new messages (canceling their consume contexts) WITHOUT closing the underlying AMQP connections — Close does that. The framework calls this during shutdown before tearing down modules so it stops delivering fresh messages to modules that are about to shut down. Cancellation propagates to in-flight handlers via their context, but they are not synchronously joined here. Idempotent: Registry.StopConsumers guards on its active flag, so a subsequent Close (which also stops consumers) is safe. Unlike Close it does not mark the manager closed and leaves the replay state intact, so a Stop is recoverable while a Close is terminal.

type ManagerOptions added in v0.9.0 

type ManagerOptions struct {
	MaxPublishers int           // Maximum number of publisher clients to keep cached
	IdleTTL       time.Duration // Time after which idle publishers are evicted
	// CleanupInterval is how often the idle-publisher sweep runs; <=0 uses the documented
	// 2-minute default. The manager starts that sweep itself at construction (ADR-067).
	CleanupInterval time.Duration
	// ConnectionTimeout is the per-publish broker confirmation timeout applied to
	// clients created by the default factory. Zero leaves the client default (30s).
	ConnectionTimeout time.Duration
	// MaxPublishAttempts bounds the per-publish retry loop for clients created by the
	// default factory. Zero (or negative) leaves the client default (5).
	MaxPublishAttempts int
	// ReadyTimeout bounds the pre-flight readiness wait for clients created by the
	// default factory. Zero (or negative) leaves the client default (5s).
	ReadyTimeout time.Duration
	// PublishTimeout is the aggregate per-publish bound (messaging.publishtimeout)
	// applied to clients created by the default factory: the readiness pre-flight
	// plus the whole retry loop run under it, layered under any tighter caller
	// deadline. Zero (or negative) leaves the publish unbounded.
	PublishTimeout time.Duration
	// Reconnect delays for clients created by the default factory. Zero (or negative)
	// leaves the client defaults (5s/60s/2s/5s). See the WithReconnect*/WithResendDelay
	// option docs for each knob's exact scope (jitter semantics, publish-error-only).
	ReconnectDelay    time.Duration
	ReconnectMaxDelay time.Duration
	ReinitDelay       time.Duration
	ResendDelay       time.Duration
	// AppName is the app.name config value stamped as the AMQP app_id property on every
	// publish by clients created by the default factory (ADR-105). Empty stamps no app_id.
	AppName string
	// ConsumerResubscribeDelay is the backoff floor between a consumer's re-subscribe
	// attempts, applied to every registry this manager builds. Zero (or negative) leaves
	// the registry default (5s), which is what a broker flap should be paced at.
	// Deliberately a Go-only seam with NO config mapping: BuildMessagingOptions never sets
	// it, so no YAML key reaches it. It exists for an embedder — or a test — that drives a
	// re-subscribe streak directly and cannot wait out the default's jittered ladder.
	ConsumerResubscribeDelay time.Duration
	// TenantStamps makes consumers read the tenant stamp off each delivery and seed
	// the handler context with it. True only under multitenant.enabled together with
	// messaging.tenancy: shared — under per-tenant tenancy the replay key is already
	// the tenant, and in single-tenant mode there is none to read.
	TenantStamps bool
}

ManagerOptions configures the Manager

type MessageHandler 

type MessageHandler interface {
	// Handle processes a message and returns an error if processing fails.
	// If an error is returned, the message will be negatively acknowledged (nack).
	// If no error is returned, the message will be acknowledged (ack).
	// The delivery is passed by pointer for performance reasons.
	Handle(ctx context.Context, delivery *amqp.Delivery) error

	// EventType returns the event type this handler can process.
	// This is used for routing messages to the correct handler.
	EventType() string
}

MessageHandler defines the interface for processing consumed messages. Handlers should implement this interface to process specific message types.

func NewTypedHandler added in v0.56.0 

func NewTypedHandler[T any](eventType string, fn func(context.Context, T) error) MessageHandler

NewTypedHandler adapts a typed function to the MessageHandler contract: decode (JSON) → validate (go-playground struct tags) → fn. It is the consumer mirror of the typed HTTP handlers registered with server.POST.

Failures short-circuit with a *PayloadError, which the worker loop nacks WITHOUT requeue like any other handler error — decode and validation failures are not retryable, so pair the queue with DeclareQueueWithDLQ to park them. Match them with errors.Is against ErrPayloadUndecodable or ErrPayloadInvalid. fn's own error is returned unwrapped, so a consumer's errors.Is against its business sentinels still works.

The returned handler holds no mutable state and is safe to share across workers and tenants. fn must therefore be safe for concurrent use too.

func NewTypedHandlerWithMeta added in v0.57.0 

func NewTypedHandlerWithMeta[T any](eventType string, fn func(context.Context, T, Metadata) error) MessageHandler

NewTypedHandlerWithMeta is NewTypedHandler for consumers that also need delivery metadata — the outbox-dedup shape: take meta.DedupKey() (a wire key holding the grammar-validated x-outbox-event-id, or the message_id property when no such header is present, or an error to return) and wrap the business logic in inbox.ProcessOnce. Failure and concurrency semantics are identical to NewTypedHandler; fn must be safe for concurrent use.

type Metadata added in v0.57.0 

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

Metadata exposes read-only delivery facts to a typed consumer without widening fn's payload contract. It is a per-delivery value constructed by the adapter AFTER the nil-delivery guard; the zero value is inert (every accessor returns its zero). Headers returns the live amqp.Table — treat it as read-only, it is shared with the worker loop.

func (Metadata) DedupKey added in v0.63.0 

func (m Metadata) DedupKey() (DedupKey, error)

DedupKey returns the key the inbox ledger should be keyed on for this delivery. For a sealed consumer it is a Sealed key spelled `<SignFamily>:<jti>` — the Logical sign family, never the concrete Generation, so a rotation does not re-open the replay window — composed from the verified envelope; that branch always returns a nil error.

For a plain typed consumer it is a wire key (WireDedupKey) holding the x-outbox-event-id header, or — when the delivery carries no such header at all — the AMQP message_id property, so a producer that follows the standard without being go-bricks is still processable through inbox.ProcessOnce. The stamp is tried first and a stamp that is present but malformed errors rather than falling through: on a go-bricks producer the stamp is framework-written while the property is caller-written, so a caller must not be able to shadow it by spoiling it.

The framework validates the SHAPE of either source, never its uniqueness: AMQP obliges no producer to make message_id unique per message, so a producer reusing one across distinct events makes the ledger skip them as duplicates. A queue whose producer does that wants the stamp, or the consumer's own key.

Return the error from the handler: the delivery is nacked without requeue, like any other poison message. AMQP header values arrive as string or []byte depending on the broker and client, so both are accepted.

func (Metadata) EventType added in v0.57.0 

func (m Metadata) EventType() string

EventType returns the wire-level message type stamped by the publisher.

func (Metadata) Headers added in v0.57.0 

func (m Metadata) Headers() amqp.Table

Headers returns the AMQP delivery headers. Nil when no headers were published. For the outbox dedup key prefer DedupKey, which validates as it extracts. The table is the delivery's own, not a copy — read it, do not mutate it. Values are publisher-controlled, so treat them as untrusted input on a queue fed from outside this service.

func (Metadata) MessageID added in v0.64.0 

func (m Metadata) MessageID() string

MessageID returns the AMQP message_id property the publisher set, or empty when the delivery carries none. It is exposed for a consumer making its OWN judgement about an unstamped delivery, never as a ledger key: take DedupKey's answer for inbox.ProcessOnce. Passed through WireDedupKey it can only become a wire key, never a Sealed one (ADR-097 §4).

func (Metadata) Redelivered added in v0.57.0 

func (m Metadata) Redelivered() bool

Redelivered reports the broker's redelivery flag.

func (Metadata) Sealed added in v0.63.0 

func (m Metadata) Sealed() (SealedEnvelope, bool)

Sealed reports whether this delivery arrived through the sealed typed door and, when it did, what its verified envelope asserts. The answer is a property of the consumer TYPE, never of the message: a sealed consumer gets (envelope, true) for every delivery it runs — the opener refused every other one before the handler — and a plain typed consumer gets (zero, false) for every delivery, whatever headers the publisher wrote, so a handler branching on ok cannot be steered by a caller-written header.

type PayloadError added in v0.56.0 

type PayloadError struct {
	// EventType is the declared consumer event type the body was routed to.
	EventType string

	// Stage is where the failure happened. It is exported to label logs and
	// metrics with "open", "decode" or "validate"; for control flow, match
	// errors.Is against ErrPayloadOpenRefused, ErrPayloadUndecodable or
	// ErrPayloadInvalid instead.
	Stage PayloadStage
	// contains filtered or unexported fields
}

PayloadError describes why a message body could not be turned into a typed payload. It is this lane's thin surface over payloaderr.Body, which owns the rendering rules and the SECURITY rationale behind them: Error() and Fields() are safe to log, Unwrap() is not.

func (*PayloadError) Error added in v0.56.0 

func (e *PayloadError) Error() string

func (*PayloadError) Fields added in v0.56.0 

func (e *PayloadError) Fields() []string

Fields returns the validator field namespaces that failed, e.g. ["CreateReq.Amount"], with every bracketed span redacted. It is empty for decode failures and for a nil receiver.

func (*PayloadError) Is added in v0.56.0 

func (e *PayloadError) Is(target error) bool

Is maps the stage onto its sentinel so consumers can discriminate the two failure modes without reading Stage.

func (*PayloadError) Unwrap added in v0.56.0 

func (e *PayloadError) Unwrap() error

Unwrap exposes the underlying decode or validation error so errors.As can reach the cause.

SECURITY: the returned error MAY carry payload-derived text. It is the deliberate escape hatch for a caller that needs the raw diagnostic; logging it is opt-in and on the caller.

type PayloadStage added in v0.56.0 

type PayloadStage string

PayloadStage names the half of the typed-payload pipeline that failed. 

const (
	PayloadStageDecode   PayloadStage = PayloadStage(payloaderr.StageDecode)
	PayloadStageValidate PayloadStage = PayloadStage(payloaderr.StageValidate)
	// PayloadStageOpen is a sealed message the opener refused before decode
	// (ADR-097): the rule's code is in Error() and the opener's own error is in
	// the chain for errors.As.
	PayloadStageOpen PayloadStage = PayloadStage(payloaderr.StageOpen)
)

Stage values carried by PayloadError.Stage. A PayloadError whose Stage is none of these matches neither sentinel.

type Publisher added in v0.63.0 

type Publisher[T any] struct {
	// contains filtered or unexported fields
}

Publisher is the module-facing handle DeclareTypedPublisher returns: a publisher bound at declaration time to ONE destination — the declared exchange, routing key, default headers and delivery flags — so a call site never re-spells them. It is the publish mirror of the typed consumer built by DeclareTypedConsumer.

Every field is set once at construction and read-only afterwards: one handle is shared by every goroutine of a module AND, under multi-tenant messaging, by every tenant's client, so any mutable state here would be a data race. Publish therefore hands the client a fresh headers map on every call and never writes back into the handle. The copy is one level deep, the same depth RegisterPublisher, Declarations.Clone and the stamping wrapper copy at: declared default headers are startup config, expected to be scalars (a string, a number, a bool), not nested tables a publish would mutate.

func DeclareTypedPublisher added in v0.63.0 

func DeclareTypedPublisher[T any](decls *Declarations, opts *PublisherOptions) *Publisher[T]

DeclareTypedPublisher registers a publisher exactly as DeclarePublisher does — same registry entry, same replay, validation and hash path — and returns a Publisher[T] handle bound to the declared destination. Keep the handle on the module (there is no deps accessor to look one up again) and publish through it from handlers and services.

The exchange is not declared here: pass it to DeclareTopicExchange separately, exactly as a DeclarePublisher with a nil exchange does.

It panics on a nil decls or a nil opts — both are declaration-time wiring mistakes, and the package already fails startup that way for the typed consumer entries.

func (*Publisher[T]) Publish added in v0.63.0 

func (h *Publisher[T]) Publish(ctx context.Context, client AMQPClient, evt T) error

Publish encodes evt and publishes it to the DECLARED exchange and routing key with the declared default headers, through client — the tenant-aware client a handler already holds (the getMessaging(ctx) idiom), so the framework's tenant stamping and trace injection apply unchanged.

A plain T is JSON-marshaled. A seal-tagged T is sealed (ADR-097) — once, here, before the client's retry loop, so every attempt and every redelivery carries the same bytes and the same signed jti. A caller-side retry after this call fails is a new seal and a new jti.

An encode or seal failure is returned wrapped and publishes nothing. Every other error is the client's own (ErrInvalidPublishDestination, ErrPublishRetriesExhausted, ...) and is returned unwrapped.

Safe for concurrent use: the handle is never written after construction and the client receives a fresh copy of the declared headers on every call.

func (*Publisher[T]) Seal added in v0.63.0 

func (h *Publisher[T]) Seal(ctx context.Context, evt T) ([]byte, error)

Seal returns the sealed wire bytes for evt without publishing them — the outbox lane persists them as-is (persisted-sealed, ADR-097) and the relay moves them byte-identical. A plain T has nothing to seal: it returns ErrNotSealTagged, and the event goes to the outbox as a struct payload the outbox already marshals.

type PublisherDeclaration 

type PublisherDeclaration struct {
	Exchange    string         // Target exchange
	RoutingKey  string         // Default routing key
	EventType   string         // Event type identifier
	Description string         // Human-readable description
	Mandatory   bool           // Message must be routed to a queue
	Immediate   bool           // Message must be delivered immediately
	Headers     map[string]any // Default headers
}

PublisherDeclaration defines what a module publishes

func NewPublisher added in v0.14.1 

func NewPublisher(opts *PublisherOptions) *PublisherDeclaration

NewPublisher creates a publisher declaration from options. If Headers is nil, an empty map is created.

type PublisherOptions added in v0.14.1 

type PublisherOptions struct {
	Exchange    string         // Target exchange name
	RoutingKey  string         // Routing key for messages
	EventType   string         // Event type identifier
	Description string         // Human-readable description
	Headers     map[string]any // Default headers (optional)
	Mandatory   bool           // Message must be routed to a queue (default: false)
	Immediate   bool           // Message must be delivered immediately (default: false)
}

PublisherOptions contains configuration for creating a publisher declaration.

type QueueDeclaration 

type QueueDeclaration struct {
	Name       string         // Queue name
	Durable    bool           // Survive server restart
	AutoDelete bool           // Delete when no consumers
	Exclusive  bool           // Only accessible by declaring connection
	NoWait     bool           // Do not wait for server confirmation
	Args       map[string]any // Additional arguments
}

QueueDeclaration defines a queue to be declared

func NewQueue added in v0.14.1 

func NewQueue(name string) *QueueDeclaration

NewQueue creates a queue with production-safe defaults.

Production defaults:

  • Durable: true (survives broker restart)
  • AutoDelete: false (won't delete when consumers disconnect)
  • Exclusive: false (can be accessed by multiple connections)
  • NoWait: false (waits for broker confirmation)

type Registry 

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

Registry manages messaging infrastructure declarations across modules. It ensures queues, exchanges, and bindings are properly declared before use. It also manages consumer lifecycle and handles message routing to handlers.

func NewRegistry 

func NewRegistry(client AMQPClient, log logger.Logger) *Registry

NewRegistry creates a new messaging registry

func (*Registry) Bindings added in v0.19.0 

func (r *Registry) Bindings() []*BindingDeclaration

Bindings returns all registered bindings (for testing/monitoring)

func (*Registry) ConsumerStates added in v0.65.0 

func (r *Registry) ConsumerStates() []ConsumerState

ConsumerStates returns a snapshot of every declared consumer's subscription state in declaration order. A consumer declared without a handler (documentation only) never subscribes, so it reports Subscribed false forever.

func (*Registry) Consumers added in v0.19.0 

func (r *Registry) Consumers() []*ConsumerDeclaration

Consumers returns all registered consumers (for documentation/monitoring)

func (*Registry) DeclareInfrastructure 

func (r *Registry) DeclareInfrastructure(ctx context.Context) error

DeclareInfrastructure declares all registered messaging infrastructure

func (*Registry) Exchanges added in v0.19.0 

func (r *Registry) Exchanges() map[string]*ExchangeDeclaration

Exchanges returns all registered exchanges (for testing/monitoring)

func (*Registry) Publishers added in v0.19.0 

func (r *Registry) Publishers() []*PublisherDeclaration

Publishers returns all registered publishers (for documentation/monitoring)

func (*Registry) Queues added in v0.19.0 

func (r *Registry) Queues() map[string]*QueueDeclaration

Queues returns all registered queues (for testing/monitoring)

func (*Registry) RegisterBinding 

func (r *Registry) RegisterBinding(declaration *BindingDeclaration)

RegisterBinding registers a binding for declaration

func (*Registry) RegisterConsumer 

func (r *Registry) RegisterConsumer(declaration *ConsumerDeclaration)

RegisterConsumer registers a consumer declaration

func (*Registry) RegisterExchange 

func (r *Registry) RegisterExchange(declaration *ExchangeDeclaration)

RegisterExchange registers an exchange for declaration

func (*Registry) RegisterPublisher 

func (r *Registry) RegisterPublisher(declaration *PublisherDeclaration)

RegisterPublisher registers a publisher declaration

func (*Registry) RegisterQueue 

func (r *Registry) RegisterQueue(declaration *QueueDeclaration)

RegisterQueue registers a queue for declaration

func (*Registry) StartConsumers 

func (r *Registry) StartConsumers(ctx context.Context) error

StartConsumers starts all registered consumers with handlers. This should be called after DeclareInfrastructure and before starting the main application.

func (*Registry) StopConsumers 

func (r *Registry) StopConsumers()

StopConsumers gracefully stops all running consumers.

func (*Registry) ValidateConsumer 

func (r *Registry) ValidateConsumer(queue string) bool

ValidateConsumer checks if a consumer is registered for the given queue

func (*Registry) ValidatePublisher 

func (r *Registry) ValidatePublisher(exchange, routingKey string) bool

ValidatePublisher checks if a publisher is registered for the given exchange/routing key

type RegistryInterface added in v0.8.0 

type RegistryInterface interface {
	// Registration methods
	RegisterExchange(declaration *ExchangeDeclaration)
	RegisterQueue(declaration *QueueDeclaration)
	RegisterBinding(declaration *BindingDeclaration)
	RegisterPublisher(declaration *PublisherDeclaration)
	RegisterConsumer(declaration *ConsumerDeclaration)

	// Infrastructure lifecycle
	DeclareInfrastructure(ctx context.Context) error
	StartConsumers(ctx context.Context) error
	StopConsumers()

	// Accessor methods for testing/monitoring
	Exchanges() map[string]*ExchangeDeclaration
	Queues() map[string]*QueueDeclaration
	Bindings() []*BindingDeclaration
	Publishers() []*PublisherDeclaration
	Consumers() []*ConsumerDeclaration

	// Validation methods
	ValidatePublisher(exchange, routingKey string) bool
	ValidateConsumer(queue string) bool
}

RegistryInterface defines the contract for messaging infrastructure management. This interface allows for easy mocking and testing of messaging infrastructure.

type ReleaseFunc added in v0.43.0 

type ReleaseFunc func()

ReleaseFunc releases a lease obtained from Publisher. Callers must invoke it (typically deferred) when finished with the publisher for the current unit of work. It is idempotent. Release does NOT close the shared publisher; it signals this borrower is done, so a publisher evicted while leased is closed only once its last lease is released. See ADR-032.

type SealCodec added in v0.63.0 

type SealCodec = sealruntime.Codec

SealCodec is what messaging/sealed registers.

type SealEnvelope added in v0.63.0 

type SealEnvelope = sealruntime.Envelope

SealEnvelope is the seam's envelope; SealedEnvelope is its Metadata twin.

type SealKeyStore added in v0.63.0 

type SealKeyStore = sealruntime.KeyStore

SealKeyStore is the app.KeyStore subset sealing needs.

type SealOpenRefusedError added in v0.63.0 

type SealOpenRefusedError = sealruntime.OpenRefusedError

SealOpenRefusedError is the seam's refusal, found in a PayloadError's chain.

type SealOpener added in v0.63.0 

type SealOpener = sealruntime.Opener

SealOpener opens one sealed delivery.

type SealOpenerProvider added in v0.63.0 

type SealOpenerProvider = sealruntime.OpenerProvider

SealOpenerProvider is the OPTIONAL consume side of a SealCodec.

type SealRuntime added in v0.63.0 

type SealRuntime = sealruntime.Runtime

SealRuntime is what the app configures at bootstrap.

func SealingRuntime added in v0.63.0 

func SealingRuntime() *SealRuntime

SealingRuntime returns the facts ConfigureSealing recorded, or nil before it ran.

type SealSpec added in v0.63.0 

type SealSpec = sealruntime.Spec

SealSpec is a codec's scanned declaration as messaging sees it: the two Logical kids.

type SealTenancy added in v0.63.0 

type SealTenancy = sealruntime.Tenancy

SealTenancy is the tenancy fact the opener's tid rule reads.

type SealTenantRule added in v0.63.0 

type SealTenantRule = sealruntime.TenantRule

SealTenantRule is the tid expectation the sealed door derives per delivery.

type SealedEnvelope added in v0.63.0 

type SealedEnvelope struct {
	// JTI is the token id the sealed dedup key `<SignFamily>:<jti>` is built from.
	JTI string
	// IssuedAt is the protected header's iat claim.
	IssuedAt time.Time
	// EventType is the event type asserted INSIDE the envelope, which the
	// framework has matched against the delivery's wire-level type.
	EventType string
	// TenantID is the tenant asserted inside the envelope (empty single-tenant).
	TenantID string
	// SignKid and SignFamily identify the verifying key and its key family.
	SignKid    string
	SignFamily string
	// EncKid identifies the key the envelope was decrypted with.
	EncKid string
}

SealedEnvelope is what a sealed (JWE-of-JWS) message's protected header asserts about itself once the framework has verified it. Plain data: a consumer that never seals reads this type without linking go-jose (the jose side has its own envelope type; the sealed door maps between them). It is reachable only through Metadata.Sealed, filled by the sealed typed door (DeclareTypedConsumerWithMeta on a seal-tagged T) and zero everywhere else.

type Sealer added in v0.63.0 

type Sealer = sealruntime.Sealer

Sealer turns one event into its sealed wire bytes.

type StreamQueueSpec added in v0.59.0 

type StreamQueueSpec struct {
	// MaxAge -> x-max-age ("<n>s"), truncated to whole seconds. A non-zero
	// sub-second value floors to "1s": second granularity is RabbitMQ's, so
	// anything briefer is inexpressible and would otherwise render "0s",
	// discarding the retention the caller asked for.
	MaxAge              time.Duration
	MaxLengthBytes      int64 // -> x-max-length-bytes
	MaxSegmentSizeBytes int64 // -> x-stream-max-segment-size-bytes
}

StreamQueueSpec configures retention for a stream queue. Zero-value fields are omitted (broker defaults apply). MaxAge is rendered as whole seconds.

Source Files

View all Source files

Directories

Path Synopsis
internal
delivery
Package delivery runs the delivery pipeline both messaging lanes share: everything that happens to one consumed message between "bytes arrived" and "outcome recorded" — trace extraction from the lane's carrier, the consumer span, the per-message lease scope, handler invocation, panic-to-error, one consumed record at completion, and the lane's own outcome line.
 Package delivery runs the delivery pipeline both messaging lanes share: everything that happens to one consumed message between "bytes arrived" and "outcome recorded" — trace extraction from the lane's carrier, the consumer span, the per-message lease scope, handler invocation, panic-to-error, one consumed record at completion, and the lane's own outcome line.
lanecontract
Package lanecontract holds the contract every messaging lane driving the delivery pipeline must satisfy, and the fixture that drives it.
 Package lanecontract holds the contract every messaging lane driving the delivery pipeline must satisfy, and the fixture that drives it.
payloaderr
Package payloaderr holds the payload-error core both messaging lanes build their typed consumers on: the decode and struct-validation steps, and a failure rendering that never echoes the body that caused it.
 Package payloaderr holds the payload-error core both messaging lanes build their typed consumers on: the decode and struct-validation steps, and a failure rendering that never echoes the body that caused it.
sealruntime
Package sealruntime is the link-time seam between the messaging package and the payload-sealing codec (ADR-097).
 Package sealruntime is the link-time seam between the messaging package and the payload-sealing codec (ADR-097).
tenantstamp
Package tenantstamp owns the tenant stamp: the carrier entry that carries a tenant identity between a producer and a consumer.
 Package tenantstamp owns the tenant stamp: the carrier entry that carries a tenant identity between a producer and a consumer.
tracking
Package sealed wires the jose/sealed codec into the messaging typed doors.
 Package sealed wires the jose/sealed codec into the messaging typed doors.
Package streams publishes to and consumes RabbitMQ streams over the native stream protocol (default port 5552, rabbitmq_stream plugin).
 Package streams publishes to and consumes RabbitMQ streams over the native stream protocol (default port 5552, rabbitmq_stream plugin).
Package testing provides test doubles for the messaging package's module-facing surface.
 Package testing provides test doubles for the messaging package's module-facing surface.

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.