messaging

README

foi-messaging-go

Standardized asynchronous messaging for FOI platform services — a transport-agnostic Go library built on Watermill and Redis Streams.

Status: released as v0.1.0, pre-1.0. The library is feature-complete against PRD v1.1, but the API is not frozen: under v0.x breaking changes arrive as minor bumps and the Go toolchain will not auto-upgrade across them. Pin an exact version. v1.0.0 follows the first FOI service integration.

---

Why this exists

FOI services communicate asynchronously, but today each one implements serialization, routing, retries, correlation, and Redis configuration on its own. This library provides those concerns once, behind a small typed API: serialization, routing, correlation, Redis configuration, retry, and dead-lettering.

Application code interacts only with this library. Watermill, Redis Streams, and go-redis are internal implementation details and never cross the library boundary.

Features

• Standard, strongly-typed event envelope with generic payloads
• Publisher and consumer APIs keyed on a shared typed EventDef (topic + type + version)
• Automatic serialization, routing, and handler dispatch
• At-least-once delivery with three-layer retry and poison-message protection
• Dead Letter Queue with a defined wrapper contract
• Correlation-ID propagation across service chains
• OpenTelemetry tracing and OTel metrics, exportable to Prometheus
• Structured logging via slog
• A testing/ package for unit-testing handlers and publish paths without Redis

Requirements

• Go 1.25+ (generics, slog)
• Redis 7.0+ (Redis Streams consumer groups)

Installation

go get github.com/bcgov/foi-messaging-go

Quick start

Define the contract

Event contracts live in a shared package and are imported by both producers and consumers:

package contracts

import messaging "github.com/bcgov/foi-messaging-go"

type DocumentCreatedPayload struct {
    EntityID string `json:"entity_id"`
    Name     string `json:"name"`
}

var DocumentCreated = messaging.EventDef{
    Topic:   "documents",
    Type:    "document.created",
    Version: "1.0.0",
}

Publish

cfg := messaging.Config{
    Source: "documents.service",
    Redis:  messaging.RedisConfig{Address: "redis:6379"},
}

publisher, err := messaging.NewPublisher(cfg)
if err != nil {
    log.Fatal(err)
}

result, err := publisher.Publish(ctx,
    contracts.DocumentCreated,
    contracts.DocumentCreatedPayload{EntityID: "42", Name: "Report.pdf"},
)
// result.EventID, result.Timestamp

The library generates event_id and timestamp. Correlation IDs are resolved from publish options, context, or generated as a UUIDv7.

Consuming events

consumer, err := messaging.NewConsumer(cfg)  // cfg.Consumer.Group is required

err = messaging.RegisterHandler(consumer, contracts.DocumentCreated, documentHandler{})

// Blocks until ctx is cancelled, then drains in-flight handlers.
err = consumer.Run(ctx)

A handler is invoked for every event on its topic whose event type matches and whose major schema version matches. Within a major version, payload changes must be backward-compatible, so a handler registered for 1.0.0 receives 1.4.2 too; breaking changes require a new major version and a new EventDef.

Events on a subscribed topic that no handler matches are acknowledged and skipped — topics are shared, and services consume only the event types they care about.

Handlers must be idempotent. Delivery is at-least-once and EventID is the deduplication key.

See examples/consumer for a complete service.

Core concepts

Envelope

Every event is wrapped in a standard envelope carrying business and workflow metadata only. Transport state (retry counters, delivery attempts, trace context) never appears in the envelope — it travels in message metadata.

type Envelope[T any] struct {
    EventID       string    `json:"event_id"`
    EventType     string    `json:"event_type"`
    Timestamp     time.Time `json:"timestamp"`
    SchemaVersion string    `json:"schema_version"`
    CorrelationID string    `json:"correlation_id"`
    Source        string    `json:"source"`
    Payload       T         `json:"payload"`
}

Routing

Routing is two-level: EventDef.Topic maps to a Redis stream, and within that stream messages dispatch to handlers by event_type + major schema version. A handler registered for 1.0.0 receives 1.x.y events, so producers can add optional fields without a coordinated consumer release. See Schema versioning.

Delivery semantics

The library is at-least-once. Three consequences are application obligations:

• Handlers must be idempotent. The same event may be delivered more than once; event_id is the deduplication key.
• Ordering is per-stream and only with Concurrency: 1 (the default). Reclaimed messages arrive out of order. Concurrency bounds in-flight handlers per subscribed topic, so a consumer registered on three topics at Concurrency: 3 can be running nine handlers. Immediate retries run inside the message's slot, so a retrying message holds its topic's slot for the whole retry window — which is what preserves ordering at Concurrency: 1.
• Publishing is not transactional with your database. A crash between a DB write and a publish loses the event. Transactional outbox support is on the roadmap, not in the initial release.

Error handling

A handler that returns an error is retried in-process — Retry.MaxImmediateRetries times, with exponential backoff and full jitter — before the message NACKs and is left pending for the reclaim loop. Redelivery is bounded by Consumer.MaxDeliveryAttempts: on the delivery whose attempt exceeds it, the event is dead-lettered and ACKed without being decoded or dispatched.

Handlers steer that path by classifying the error they return:

return messaging.AsPermanent(err) // → Dead Letter Queue, then ACK
return messaging.AsRetryable(err) // → retried in-process, then NACKed
return messaging.AsDiscard(err)   // → acknowledged without retry or DLQ

An unclassified error is retryable. Classification is re-read on every attempt, so a handler may fail transiently and then return AsPermanent once it knows better. AsPermanent and AsDiscard both skip the remaining retries; AsPermanent skips the delivery cap too, since the verdict is already final.

Classification is additive: errors.Is and errors.As see straight through the wrapper, and a classified error may itself be wrapped with %w without losing its verdict.

Dead Letter Queue

Permanent failures, messages exceeding the delivery cap, and events that could not be deserialized are published to <topic>.dlq using the exported messaging.DeadLetter wrapper, which carries failure metadata — reason, error, delivery_attempts, consumer_group, consumer_name, original_topic, dead_lettered_at — alongside the original event.

The event travels in one of two fields, never both. event holds the original bytes verbatim when they were valid JSON, so replay tooling can republish without transformation; event_raw holds them when they were not parseable, which is the case for a malformed entry or an envelope that failed validation. Splicing unparseable bytes into event would make the dead letter itself invalid JSON, unreadable by the very tooling the DLQ exists for.

A failed DLQ write NACKs rather than ACKs: while the DLQ is unwritable the entry stays pending and the next reclaim sweep retries it, which is preferable to acknowledging an event into nothing.

Configuration

A minimal config is three fields; every reclaim and concurrency knob has a working default.

cfg := messaging.Config{
    Source:   "billing.service",                          // required
    Redis:    messaging.RedisConfig{Address: "redis:6379"}, // Address required
    Consumer: messaging.ConsumerConfig{Group: "billing-service"}, // required for consumers
}

Redis auth/TLS, pool sizing, consumer concurrency, and claim intervals are all configurable with working defaults, as are the delivery cap (MaxDeliveryAttempts, default 5) and retry backoff (RetryConfig, default 3 retries from 100ms to 5s). Because retries sleep inside the message's concurrency slot, NewConsumer rejects a config whose worst-case backoff reaches Consumer.ClaimMinIdle — such a config guarantees the entry is reclaimed, and processed a second time by the same process, before the first delivery has finished retrying. See the full configuration reference.

Observability

Structured logging

Consumers emit structured slog logs when an envelope cannot be decoded, fails validation, or carries an unparseable schema version, a warning whenever an event is dead-lettered or discarded, and a debug log when no registered handler matches an event. Logged fields include topic, event_id, event_type (when a handler is not found), error, and trace_id/span_id so a log line and its trace are navigable from each other. Correlation IDs propagate through handler contexts via context.Context.

Payload contents are not logged by default. Set Telemetry.LogPayloads: true to include them on every consume-path log line for a delivery — not just its error lines, since the payload attaches to that delivery's base logger. They are never placed in span or metric attributes regardless of that setting, because spans and metrics routinely leave the trust boundary that logs stay inside.

Tracing

Publish opens a producer span; dispatch opens a consumer span parented to it, so handlers receive a context.Context whose span is a child of the publisher's. Trace context travels as traceparent transport metadata, injected and extracted by Telemetry.Propagator.

That field defaults to propagation.TraceContext{} directly, not to otel.GetTextMapPropagator() — the OTel global is a no-op until the application sets it, and defaulting from it would silently disable cross-service trace continuity with no error anywhere to explain why.

One span per delivery, not per attempt: immediate retries are recorded as span events. Span volume therefore scales with redelivery — a persistently failing event produces up to MaxDeliveryAttempts spans. Sampling is the application's decision.

Metrics

The library records through the OpenTelemetry metric API only. It has no Prometheus dependency; hand it a MeterProvider and export however you like. examples/telemetry is a working Prometheus wiring.

Prometheus name	Type	Attributes
messaging_events_published_total	counter	topic, event_type
messaging_publish_failures_total	counter	topic, event_type, stage
messaging_events_received_total	counter	topic
messaging_events_processed_total	counter	topic, event_type, group
messaging_events_failed_total	counter	topic, event_type*, group, error_category
messaging_events_skipped_total	counter	topic, group, reason
messaging_retries_total	counter	topic, event_type, group
messaging_dlq_total	counter	topic, group, reason
messaging_dlq_publish_failures_total	counter	topic, group, reason
messaging_processing_duration_seconds	histogram	topic, event_type, group
messaging_queue_latency_seconds	histogram	topic

Every delivery increments exactly one of processed, failed, or skipped, and records processing_duration exactly once. dlq is orthogonal and fires alongside failed on the dead-letter paths reached through dispatch — it answers "what are we giving up on", not "what failed".

The one exception is a stream entry Watermill's own marshaller cannot read at all: it never reaches dispatch, so it increments received and dlq and none of processed/failed/skipped, and never records processing_duration. This is deliberate — there is no envelope to attribute a terminal outcome or a duration to — but it means alerting on rate(messaging_events_failed_total) alone will not catch a producer that starts writing corrupt entries; watch messaging_dlq_total too.

* event_type is attached only when the event matched a typed handler registration, where it comes from a set fixed at registration time. On the no-handler, raw-handler, and deserialization paths it is whatever the wire said — unbounded, and one bad producer away from exploding your metric store — so it is omitted from metrics and recorded on the span instead.

These are per-delivery counters

A retryable failure NACKs, and the entry is later reclaimed and redelivered. One event therefore increments received once per delivery, up to MaxDeliveryAttempts + 1 times. received exceeding published is redelivery working as designed, not double-counting.

processing_duration includes the dead-letter write

On the five dead-letter paths reached through dispatch/runWithRetry (delivery-cap exceeded, three deserialization failures, and a permanent handler error) the histogram covers the DLQ publish as well as decode and handler time, because that write genuinely occupies the delivery's concurrency slot. A DLQ outage will therefore show up as a p99 processing_duration_seconds spike and in messaging_dlq_publish_failures_total. That correlation is expected; the second metric is the one that tells you which it is. The sixth dead-letter path — an entry the marshaller cannot read at all — never reaches dispatch, so it has no processing_duration to include a DLQ write time in; see the exception noted above.

queue_latency depends on clock sync

published_at is stamped by the publishing host and read by the consuming host, so messaging_queue_latency_seconds measures elapsed time plus clock skew. Negative values are clamped to zero. It is only as trustworthy as your fleet's NTP. A missing or unparseable published_at skips the observation rather than failing the delivery.

The histogram bucket View is required

The OTel Prometheus exporter's default histogram boundaries are millisecond-scaled (0, 5, 10, ... 10000). Both of the library's histograms are in seconds, so without an explicit View every realistic observation lands in the first bucket and both render as flat lines. The library cannot fix this — your application owns the MeterProvider and therefore owns the Views.

Copy the View from examples/telemetry. Omitting it is the most likely way to finish integrating and still be unable to see your own latency.

The exporter also adds otel_scope_name/otel_scope_version labels to every series and a target_info series. Both are normal.

Testing

Testing your own service

Import github.com/bcgov/foi-messaging-go/testing as messagingtest. Nothing in it needs Redis or Docker, and it drives the library's real code rather than a simulation of it — so a malformed EventDef or a misclassified error fails your unit test rather than production.

Recording publishes. messagingtest.Publisher wraps a real publisher with only its transport write redirected, so envelope construction, correlation-ID resolution, and validation are genuine:

pub, _ := messagingtest.NewPublisher()
defer pub.Close()

svc := NewService(pub) // your code, against your own narrow interface
svc.CreateOrder(ctx, order)

published := pub.Published()
payload, _ := messagingtest.PayloadAs[OrderCreated](published[0])

Testing a handler. Deliver reproduces the handler boundary — it installs the context values the router installs, invokes the handler, and returns its error. It does not retry, ack, or dead-letter:

err := messagingtest.Deliver(ctx, handler, env)

Testing what the library would do with it. Dispatch runs the real consume path against your own configured Consumer, covering the delivery-attempt cap, error classification, the retry loop, and dead-lettering:

c, _ := messaging.NewConsumer(messagingtest.Config())
messaging.RegisterHandler(c, contracts.OrderCreated, handler)

e, _ := messagingtest.NewEvent(contracts.OrderCreated, payload)
res, _ := messagingtest.Dispatch(ctx, c, e)

res.Outcome              // processed | skipped | dead_lettered | nacked
res.DeadLetters[0].Reason

Because Published() returns the same Event type Dispatch accepts, one service's publish is the next service's input — a two-service chain, no Redis:

res, _ := messagingtest.Dispatch(ctx, consumerB, pub.Published()[0])

Dispatch reports what the runtime would do with a delivery; it does not perform one. There is no ack, no pending entry, and no reclaim — for those, use the integration tier against real Redis.

The library's own suite

The library's own suite has three tiers: make test (unit), make test-examples (the nested examples/telemetry module, which ./... does not reach), and make test-integration (needs Docker). make test-all runs the first two.

Integration tests in this repository use Testcontainers against real Redis.

Project layout

foi-messaging-go/
├── config.go        envelope.go     eventdef.go
├── publisher.go     consumer.go     handler.go
├── validation.go    errors.go       context.go     dlq.go
├── telemetry.go     registry.go
├── testing/         examples/
└── internal/
    ├── watermill/
    └── redis/

Applications import the top-level package, and testing/ from their tests. All Watermill and Redis code stays in internal/, enforced by a golangci-lint depguard rule, which CI runs on every pull request.

Roadmap

Planned after v1.0: transactional outbox, a Redis-backed idempotency helper, delayed retry queues, DLQ replay tooling, a schema registry, and additional transports (Kafka, RabbitMQ, and others). Because applications depend only on the library interfaces, these can arrive without application changes.

Documentation

Full design and rationale live in the Product Requirements Document.

Releasing

Releases are cut by pushing a tag. A published Go module version is immutable — once proxy.golang.org has served it, it can never be corrected, only superseded — so the order here matters.

1. Add a ## [X.Y.Z] - YYYY-MM-DD section to CHANGELOG.md, and update the link definitions at the bottom of the file.

2. Run make verify. This is exactly what CI runs, including the -race integration tier, which needs Docker.

3. Confirm CI is green on main.

4. Tag and push:

   git tag v0.1.0
   git push origin v0.1.0
   

release.yml then re-runs every gate on the tagged commit and creates the GitHub Release from the matching changelog section. A tag with no changelog section fails the release rather than publishing empty notes, and a tag containing a hyphen (v0.1.0-rc.1) is marked as a prerelease.

If the gates fail, no Release is created — but the tag exists. Delete it immediately (git push --delete origin vX.Y.Z) and re-tag; that only works before anything has fetched the version through the module proxy. After that, ship the fix as the next patch version.

License

Apache License 2.0 — see LICENSE.

Documentation

Overview

Package messaging provides a transport-agnostic, strongly-typed asynchronous messaging library for FOI platform services, built on Watermill and Redis Streams.

See docs/foi-messaging-go-prd-v1.1.md for the full design. The publish path (EventDef, Envelope, Config, and Publisher) and the consume path (Consumer, Handler, and routing) are implemented, as is the failure path: handlers classify errors with AsPermanent, AsRetryable, or AsDiscard, a retryable failure is retried in-process with exponential backoff and full jitter before the message nacks, redelivery is bounded by Consumer.MaxDeliveryAttempts, and an event that reaches the end of that path is published to its topic's dead letter queue as a DeadLetter and acked.

Observability is live: Publish opens a producer span and dispatch opens a consumer span parented to it via traceparent transport metadata, and both paths record OpenTelemetry metrics — eleven instruments covering publish, receive, process, failure, skip, retry, and dead-letter counts plus processing-duration and queue-latency histograms. The library depends on the OTel metric API only; see examples/telemetry for Prometheus wiring, including the histogram bucket View that recipe requires.

The application-facing testing package is implemented: import github.com/bcgov/foi-messaging-go/testing to record publishes, invoke a handler at the router's boundary, and assert what the consume path would do with a delivery — all without Redis.

Index

• Constants
• func AsDiscard(err error) error
• func AsPermanent(err error) error
• func AsRetryable(err error) error
• func IsDiscard(err error) bool
• func IsPermanent(err error) bool
• func IsRetryable(err error) bool
• func RegisterHandler[T any](c *Consumer, def EventDef, h Handler[T]) error
• func RegisterRawHandler(c *Consumer, sel TopicSelector, h Handler[json.RawMessage]) error
• type Config
  • func (c *Config) Validate() error
• type Consumer
  • func NewConsumer(cfg Config) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) Run(ctx context.Context) error
• type ConsumerConfig
• type DeadLetter
• type Envelope
• type EventDef
• type Handler
• type PublishOption
  • func WithCorrelationID(id string) PublishOption
• type PublishResult
• type Publisher
  • func NewPublisher(cfg Config) (*Publisher, error)
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(ctx context.Context, def EventDef, payload any, opts ...PublishOption) (PublishResult, error)
• type RedisConfig
• type RetryConfig
• type TelemetryConfig
• type TopicSelector

Constants

const (
	// ReasonPermanent is a handler error classified with AsPermanent.
	ReasonPermanent = "permanent"
	// ReasonMaxAttemptsExceeded is the delivery-attempt cap firing,
	// regardless of how the failures were classified.
	ReasonMaxAttemptsExceeded = "max_attempts_exceeded"
	// ReasonDeserializationFailed covers every event that could not be read
	// well enough to dispatch: invalid JSON, an envelope failing validation,
	// an unparseable schema version, or a stream entry Watermill's own
	// marshaller rejected.
	ReasonDeserializationFailed = "deserialization_failed"
)

DLQ reason values, fixed by PRD §14. Exported so operational tooling consuming DLQ streams can compare against them rather than against string literals of its own.

Functions

func AsDiscard

func AsDiscard(err error) error

AsDiscard marks err as a failure the application wants dropped: the event is logged at warn and acked, with no retry and no DLQ entry.

func AsPermanent

func AsPermanent(err error) error

AsPermanent marks err as a failure that will not resolve on retry — a validation failure, an unresolvable reference. The consume path routes it straight to the DLQ and acks, skipping both immediate retry and the delivery-attempt cap.

A nil err returns nil: wrapping "no error" would turn a successful handler return into a dead letter.

func AsRetryable

func AsRetryable(err error) error

AsRetryable marks err as transient. It is optional — an unclassified error is already retryable (PRD §15) — and exists so handlers can say so deliberately rather than by omission.

func IsDiscard

func IsDiscard(err error) bool

IsDiscard reports whether err is classified discard.

func IsPermanent

func IsPermanent(err error) bool

IsPermanent reports whether err is classified permanent.

func IsRetryable

func IsRetryable(err error) bool

IsRetryable reports whether err should be retried, which an unclassified error is by default (PRD §15).

The retry loop does not call this — retrying is its fallthrough — but operational tooling and tests do, and the default has to be stated somewhere executable rather than only in a comment.

func RegisterHandler

func RegisterHandler[T any](c *Consumer, def EventDef, h Handler[T]) error

RegisterHandler registers a typed handler for def. It is a function rather than a method because Go does not permit generic methods.

The handler is invoked for every event on def.Topic whose event type matches def.Type and whose major schema version matches def.Version's. Minor and patch differences do not affect dispatch: producers may add optional fields without a coordinated consumer release, so handlers must tolerate any additive change within their major version.

func RegisterRawHandler

func RegisterRawHandler(c *Consumer, sel TopicSelector, h Handler[json.RawMessage]) error

RegisterRawHandler registers a handler that receives every event on a topic with its payload left as raw JSON. Use it when several payload shapes share an event type, or to consume events without a typed contract.

A topic may have typed handlers or one raw handler, never both.

Types

type Config

type Config struct {
	Source       string
	StreamPrefix string
	Redis        RedisConfig
	Consumer     ConsumerConfig
	Retry        RetryConfig
	Telemetry    TelemetryConfig
}

Config is the library's single configuration object. A minimal config is three fields: Source, Redis.Address, and — for consumers — Consumer.Group. Every other field has a working default.

func (*Config) Validate

func (c *Config) Validate() error

Validate checks required fields and fills in defaults for everything else. Called by NewPublisher, and by NewConsumer — which then also calls validateConsumer for the consumer-only fields.

type Consumer

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

Consumer subscribes to the topics its registered handlers cover and dispatches each event to the handler matching its event type and major schema version.

A Consumer is created from a Config, has handlers registered against it, and is then run. Registration after Run has started is an error.

func NewConsumer

func NewConsumer(cfg Config) (*Consumer, error)

NewConsumer validates cfg and returns a Consumer with no handlers registered. It opens no connections — the Redis client, subscriber, and router are built by Run, once the set of topics is known.

func (*Consumer) Close

func (c *Consumer) Close() error

Close releases the Redis client held by a Consumer that was constructed but never run. It is idempotent, returns nil when there is nothing to release, and is safe to defer unconditionally.

Close is not how a running consumer is stopped. Cancel the context passed to Run: Run drains its handlers and releases everything itself before returning. While Run is in progress — and after it has returned, when there is nothing left to release — Close does nothing and returns nil.

That guard matters. Closing the client under a live read loop wedged the consumer permanently and silently: "redis: client is closed" is neither ctx.Done nor a shutdown signal, so the read loop backed off and retried it forever, Run never returned, and the process never exited.

func (*Consumer) Run

func (c *Consumer) Run(ctx context.Context) error

Run subscribes to every registered topic and blocks until ctx is cancelled, then drains in-flight handlers within cfg.Consumer.ShutdownTimeout before returning.

type ConsumerConfig

type ConsumerConfig struct {
	// Group is the Redis consumer group name. Required.
	Group string

	// ConsumerName identifies this instance within Group. Defaults to
	// <hostname>-<random suffix>, which stays unique across replicas and
	// restarts on one host.
	ConsumerName string

	// Concurrency bounds how many messages may be in flight at once *per
	// subscribed topic*, not across the consumer as a whole: a consumer
	// registered on three topics at Concurrency 3 can be running nine
	// handlers. The bound is per topic because each topic has its own read
	// loop, and a shared bound would let an idle topic's blocking read
	// throttle a busy one. Defaults to 1, at which per-topic ordering is
	// preserved.
	Concurrency int

	// ClaimInterval is how often the reclaim sweep runs. Defaults to 30s.
	ClaimInterval time.Duration

	// ClaimMinIdle is how long an entry must sit unacknowledged before
	// another consumer may reclaim it. It must be >= ClaimInterval, and it
	// must also exceed the longest one delivery can occupy its concurrency
	// slot — which, with immediate retry, is
	//
	//	(1 + Retry.MaxImmediateRetries) × handler duration + worst-case backoff
	//
	// not one handler run. A 20s handler at the defaults occupies its slot
	// for up to 80.7s, so the 60s default is already too short for it: the
	// entry is reclaimed and processed concurrently by this same process
	// while the first delivery is still retrying.
	//
	// The backoff term alone is checked at construction; the handler term
	// cannot be. Defaults to 60s.
	ClaimMinIdle time.Duration

	// MaxDeliveryAttempts bounds redeliveries. On the delivery whose
	// attempt exceeds it, the event is dead-lettered and acked before it is
	// even decoded — regardless of how its failures were classified
	// (PRD §13 Layer 3). Attempts 1..MaxDeliveryAttempts dispatch;
	// attempt MaxDeliveryAttempts+1 is dead-lettered. Defaults to 5.
	MaxDeliveryAttempts int

	// ShutdownTimeout bounds the drain of in-flight handlers after the Run
	// context is cancelled.
	//
	// Immediate retry extends the drain: a message that fails on the last
	// attempt before shutdown can still spend
	// Retry.MaxImmediateRetries × handler duration + worst-case backoff
	// finishing, across Concurrency × (number of subscribed topics)
	// messages. Retries are deliberately not interrupted by shutdown —
	// message contexts stay live for the whole drain — so budget for it
	// here. Defaults to 30s.
	ShutdownTimeout time.Duration
}

ConsumerConfig configures a Consumer. Its defaults and validation are applied by validateConsumer, which NewConsumer calls after Validate.

type DeadLetter

type DeadLetter struct {
	DeadLetteredAt   time.Time       `json:"dead_lettered_at"`
	Reason           string          `json:"reason"`
	Error            string          `json:"error"`
	DeliveryAttempts int64           `json:"delivery_attempts"`
	ConsumerGroup    string          `json:"consumer_group"`
	ConsumerName     string          `json:"consumer_name"`
	OriginalTopic    string          `json:"original_topic"`
	Event            json.RawMessage `json:"event,omitempty"`
	EventRaw         []byte          `json:"event_raw,omitempty"`
}

DeadLetter is the wrapper written to a topic's DLQ stream.

PRD §5 forbids failure metadata inside the event, so the original event travels verbatim alongside the metadata rather than being modified to carry it. That is what lets replay tooling republish the event without transformation.

Exactly one of Event and EventRaw is set. Event holds the original bytes when they were valid JSON; EventRaw holds them — base64-encoded by encoding/json — when they were not, which is the only case where preserving unparseable input is what matters.

type Envelope

type Envelope[T any] struct {
	EventID       string    `json:"event_id"`
	EventType     string    `json:"event_type"`
	Timestamp     time.Time `json:"timestamp"`
	SchemaVersion string    `json:"schema_version"`
	CorrelationID string    `json:"correlation_id"`
	Source        string    `json:"source"`
	Payload       T         `json:"payload"`
}

Envelope wraps every published and consumed event. It carries business and workflow metadata only — transport state (retry counters, delivery attempts, trace context) travels in message metadata, never here.

type EventDef

type EventDef struct {
	Topic   string
	Type    string
	Version string
}

EventDef identifies an event contract: which topic it publishes on, its event type, and its schema version. Declared once per event in the contract package that owns the payload type, and shared by publishers and consumers.

type Handler

type Handler[T any] interface {
	Handle(context.Context, Envelope[T]) error
}

Handler processes a typed event payload. Applications implement this interface and register implementations with a Consumer.

Handlers must be idempotent: delivery is at-least-once, so the same event may arrive more than once (PRD §6). EventID is the deduplication key.

type PublishOption

type PublishOption func(*publishOptions)

PublishOption customizes a single Publish call.

func WithCorrelationID

func WithCorrelationID(id string) PublishOption

WithCorrelationID explicitly sets the correlation ID for a publish call, taking priority over any correlation ID carried on the context.

type PublishResult

type PublishResult struct {
	EventID   string
	Timestamp time.Time
}

PublishResult is returned by a successful Publish.

type Publisher

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

Publisher publishes typed payloads to Redis streams without exposing Watermill or go-redis to callers.

func NewPublisher

func NewPublisher(cfg Config) (*Publisher, error)

NewPublisher validates cfg and builds a Publisher backed by it.

func (*Publisher) Close

func (p *Publisher) Close() error

Close releases the Publisher's underlying resources.

func (*Publisher) Publish

func (p *Publisher) Publish(ctx context.Context, def EventDef, payload any, opts ...PublishOption) (PublishResult, error)

Publish builds a standard envelope around payload and writes it to the stream named by cfg.StreamPrefix + ":" + def.Topic. Errors are returned synchronously; the library does not buffer or retry publishes.

type RedisConfig

type RedisConfig struct {
	Address  string
	Username string
	Password string
	TLS      *tls.Config
	DB       int
	PoolSize int
}

RedisConfig configures the Redis connection the library uses internally. Applications never construct a go-redis client themselves.

type RetryConfig

type RetryConfig struct {
	// MaxImmediateRetries is the number of retries after the first
	// attempt, so a message gets 1+MaxImmediateRetries invocations per
	// delivery. Defaults to 3.
	MaxImmediateRetries int

	// InitialBackoff is the upper bound of the first retry's jittered
	// sleep, doubling per retry. Defaults to 100ms.
	InitialBackoff time.Duration

	// MaxBackoff caps that doubling. Defaults to 5s.
	MaxBackoff time.Duration
}

RetryConfig configures the in-process immediate-retry layer (PRD §13 Layer 1): the retries a handler gets within one delivery, before the message is nacked and left for the reclaim loop.

Zero values mean "use the default" — 3 retries, 100ms initial, 5s max — so there is no way to express "no retries" by zeroing MaxImmediateRetries. Set InitialBackoff and MaxBackoff to a nanosecond in tests that need the loop to run without waiting.

Retries sleep inside the handler's concurrency slot, so these values interact with Consumer.ClaimMinIdle; see its documentation.

type TelemetryConfig

type TelemetryConfig struct {
	TracerProvider trace.TracerProvider
	MeterProvider  metric.MeterProvider
	// Propagator injects trace context at publish and extracts it at
	// consume (PRD §5). It defaults to propagation.TraceContext{}
	// directly, NOT to otel.GetTextMapPropagator(), which returns a no-op
	// unless the application has called otel.SetTextMapPropagator. A
	// no-op here would leave traceparent unwritten and every consumer
	// span a disconnected root, with nothing reporting a fault — the
	// failure would surface only during the first incident the tracing
	// was bought for.
	//
	// Baggage is deliberately not composited in: PRD §5's transport
	// metadata table lists traceparent and tracestate and nothing else.
	// Applications wanting baggage pass their own composite here.
	Propagator  propagation.TextMapPropagator
	Logger      *slog.Logger
	LogPayloads bool
}

TelemetryConfig configures observability integration. Logger is defaulted by Validate and is used throughout the consume path — the subscriber's loops, dispatch, and watermill's own router logs all go through it. Propagator is defaulted here and carries trace context across the publish/consume boundary. TracerProvider and MeterProvider default to otel.GetTracerProvider()/otel.GetMeterProvider() when nil, and are live: Publish and the consume path both open spans and record the full set of PRD §16 metrics through them. An application that never calls otel.SetTracerProvider/otel.SetMeterProvider gets the OTel SDK's no-op implementations, which is a silent way to end up with no telemetry rather than an error — set these fields explicitly, or call the otel.Set* functions before constructing a Publisher or Consumer.

type TopicSelector

type TopicSelector struct {
	Topic string
}

TopicSelector identifies a topic for raw handler registration, for cases where several payload shapes share an event type or a service consumes events it has no typed contract for.