messenger

package module
v0.2.2 Latest Latest
Warning

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

 Go to latest Published: Aug 29, 2026 License: MIT Imports: 24 Imported by: 7

README

GoMessenger

CI Release Go Reference Go 1.27

Typed durable messaging for Go.

Build commands, local queries, and events with explicit transaction, delivery, retry, and broker semantics.

Transactional Outbox/Inbox, NATS JetStream, Kafka, bounded retries, DLQ/replay, tracing, and managed lifecycle.

Release status: v0.2.1 is the current release line. Source validation, dependency-ordered tag publication, and the clean published-consumer gate are separate release evidence. The real-service pilot remains pending, so controlled repository gates are not a production-readiness claim. The current checkout may accumulate follow-up changes beyond that release line.

What it is

GoMessenger is a typed facade for Go 1.27+ services that need both process-local messaging and durable one-way delivery. Commands, local request/reply queries, and events share explicit descriptors, middleware, lifecycle, and observations; one-way commands and events can additionally use transactional Outbox staging, NATS JetStream or Kafka delivery, and a durable SQL Inbox.

It keeps transport truth visible. An Outbox receipt means staged in the caller's transaction, a NATS receipt means JetStream returned PubAck, a Kafka receipt means the producer transaction committed, and a consumer ACK/offset follows the committed Inbox transaction. The root module remains standard library plus gobus; broker, SQL, Prometheus, and OpenTelemetry dependencies live in optional nested modules.

Why it exists

Use GoMessenger when a service needs at least one of these boundaries:

  • a business write and outgoing command/event must commit or roll back together;
  • broker redelivery must not repeat an already committed SQL handler effect;
  • retry, terminal failure, DLQ, and replay must be bounded and explicit;
  • local commands, local queries, and durable events should use stable typed descriptors without pretending that NATS and Kafka have identical semantics.

If all work is process-local, GoBus or direct function calls may be enough. If the problem is a durable multi-step workflow with timers and compensation, use a workflow engine. See the use-case comparison before choosing an abstraction.

Install

GoMessenger requires Go 1.27+. For local commands, queries, and events:

go get github.com/assurrussa/gomessenger@v0.2.1

For durable NATS JetStream delivery with Inbox and transactional Outbox integration:

go get github.com/assurrussa/gomessenger@v0.2.1 \
  github.com/assurrussa/gomessenger/adapters/inbox@v0.2.1 \
  github.com/assurrussa/gomessenger/adapters/nats@v0.2.1 \
  github.com/assurrussa/gomessenger/adapters/outbox@v0.2.1

For durable Kafka delivery with Inbox and transactional Outbox integration:

go get github.com/assurrussa/gomessenger@v0.2.1 \
  github.com/assurrussa/gomessenger/adapters/inbox@v0.2.1 \
  github.com/assurrussa/gomessenger/adapters/kafka@v0.2.1 \
  github.com/assurrussa/gomessenger/adapters/outbox@v0.2.1

Optional telemetry and CLI modules use the same release version:

go get github.com/assurrussa/gomessenger/observability@v0.2.1
go install github.com/assurrussa/gomessenger/tools/gomessengerctl@v0.2.1

These commands target the exact path-qualified v0.2.1 tags. The rest of this README tracks the current checkout and may describe unreleased APIs that are not present in that release line. Use the versioned Go Reference for the exact release API, or use the checkout workflow below when evaluating unreleased changes.

Keep every GoMessenger module in one consumer on the same version. The Outbox adapter requires Outbox v0.12.0; the host selects and installs its matching database backend separately. To evaluate the current checkout instead:

git clone https://github.com/assurrussa/gomessenger.git
cd gomessenger
GOWORK=off go test ./...

30-line local quickstart

The smallest local command is:

type ResizeMedia struct {
	JobID int64 `json:"jobId"`
}

resize := messenger.MustCommand("media.resize", 1, messenger.JSON[ResizeMedia]())
builder := messenger.NewBuilder(messenger.WithSource("urn:service:media-resizer"))
builder.HandleCommandFunc(resize, "media-worker", func(_ context.Context, payload ResizeMedia) error {
	fmt.Println("resize", payload.JobID)
	return nil
})
builder.RouteCommand(resize, messenger.NewLocalSyncRoute())

bus, _, err := builder.Build()
if err != nil {
	return err
}
resizeSender := messenger.BindSender(bus, resize)
_, err = resizeSender.Send(ctx, ResizeMedia{JobID: 42})

The equivalent runnable API is compiled as ExampleMessenger_Send; testdata/consumer separately compiles the complete public facade from an external Go module.

For the durable path, run the PostgreSQL + NATS demo:

make demo-durable-postgres-nats

It performs a business write and Outbox stage in one PostgreSQL transaction, relays through JetStream, retries an intentional handler failure, suppresses a distinct duplicate delivery in the Inbox, moves a permanent failure to the DLQ, and confirms replay. The example is a checkout-level demonstration with local GoMessenger replacements; it is not evidence of published-module resolution or production readiness.

The example also contains an opt-in open-loop NATS capacity experiment over HTTP -> business transaction + Outbox -> JetStream -> Inbox -> business projection:

make capacity-nats
make capacity-nats-site
make capacity-inbox-postgres

The default command retains the four-Outbox/four-consumer PostgreSQL 18 profile. The site-shaped command uses PostgreSQL 17 with two Outbox workers, reservation batch 1, one consumer, isolated Outbox producer/relay pools fixed at 9 + 1, and a separate ten-connection Inbox/measurement pool; the PostgreSQL-only command isolates the real Inbox ProcessAttempt transaction without Outbox or NATS. They report unique committed business effects, canonical envelope bytes, Inbox/ACK latency, and PostgreSQL statement/WAL/I/O telemetry. See the example capacity contract. Results describe only the recorded checkout, host, and local Docker topology; they are not production benchmark claims.

Set OUTBOX_RESERVATION_BATCH_SIZE=16 (valid 1..1000) to A/B only the reservation/prefetch width. The default remains 1. Each Outbox worker still publishes and acknowledges jobs sequentially, so this does not multiply handler concurrency. Capacity report spec 1.3 records the batch and exact Outbox module version in JSON and Markdown. It reports relay throughput from published envelopes, consumer throughput and MiB/s from committed projections, and separates staged - published Outbox lag from published - committed consumer lag. Warm-up and drain remain outside every throughput denominator.

Versioned capacity baselines, raw-evidence rules, and defensible claim boundaries are recorded in the performance evidence registry.

Guarantees

The durable contract is at-least-once:

  • an Outbox route reports success only after the envelope is staged in the caller's database transaction;
  • a direct NATS route reports success only after a JetStream PubAck;
  • a direct Kafka route reports success only after its producer transaction commits;
  • a durable consumer acknowledges only after its Inbox transaction and handler commit;
  • stable message identity and an Inbox suppress a second execution of the same committed consumer transaction;
  • concurrency, envelopes, headers, handler attempts, retry delays, and shutdown waits are bounded by explicit configuration or documented limits.

Non-goals

  • No exactly-once external effects: HTTP calls, email, object storage, and writes to another database still need their own idempotency or durable hand-off.
  • No distributed queries: Query[Q,R] is process-local. Remote request/reply remains the separate, unimplemented contract in ADR-0003.
  • No workflow, saga, event-sourcing, service-discovery, or streaming-analytics engine.
  • No automatic ownership of host database connections, broker credentials, migrations, topology policy, supervision, or deployment.
  • No claim of universal broker abstraction or production-proven maturity; the real-service pilot in ADR-0002 is still pending.

Choose a route

Start with an explicit descriptor, choose the success boundary, build the messenger, and inject a narrow Sender, Querier, or Publisher into business code. Durable consumers are separate managed services.

Route A successful call means Important boundary
NewLocalSyncRoute the in-process handler completed no restart durability
NewLocalAsyncRoute bounded runtime work completed for queries or was accepted for one-way work no restart durability
natsadapter.NewRoute JetStream returned PubAck not atomic with a separate business database write
kafkaadapter.NewRoute the producer transaction committed not atomic with a separate business database write
outboxadapter.NewProducer the envelope was staged in the active host transaction provisional until that transaction commits 
business code -> Messenger.Query -> local result handler -> typed result
             `-> Send/Publish -> local handler
                              `-> Outbox -> relay -> JetStream/Kafka -> consumer -> Inbox transaction -> handler -> ACK/offset

For a durable flow, provision topology, apply Outbox and Inbox migrations, register the relay before producer traffic, start the managed runtimes, and expose readiness before accepting traffic. The practical usage guide shows the full composition and shutdown order.

Modules and release status

GoMessenger requires Go 1.27 because the builder and messenger expose generic methods.

Module Responsibility
github.com/assurrussa/gomessenger descriptors, local queries, envelopes, local routes, runtime, manifest
.../adapters/outbox transactional staging and broker relay job
.../adapters/nats JetStream producers, consumers, topology, CloudEvents
.../adapters/kafka transactional Kafka producers, consumers, topology, retry/DLQ
.../adapters/inbox atomic PostgreSQL and SQLite consumer deduplication
.../observability Prometheus, OpenTelemetry spans, W3C Trace Context
.../tools/gomessengerctl manifest/topology validation, plan/apply, DLQ inspect/replay

The module set uses synchronized path-qualified v0.2.1 tags. Release completion requires every tag above plus the clean post-publication consumer probe; neither is inferred from source-only checks. Outbox root and its PostgreSQL/SQLite backend tags at v0.12.0 are the pinned durable-producer dependencies. During repository development go.work selects local GoMessenger modules; published consumers use matching path-qualified tags and no local replace directives. See the release process for dependency order and verification.

Minimal local query example

type FindArticle struct{ ID int64 }
type ArticleView struct {
	ID    int64
	Title string
}

findArticle := messenger.MustQuery[FindArticle, ArticleView](
	"article.find", 1, messenger.JSON[FindArticle](),
)
builder := messenger.NewBuilder(messenger.WithSource("urn:service:catalog"))
builder.HandleQueryFunc(findArticle, "article-reader", func(_ context.Context, query FindArticle) (ArticleView, error) {
	return ArticleView{ID: query.ID, Title: "CQRS in Go"}, nil
})
builder.RouteQuery(findArticle, messenger.NewLocalSyncRoute())

bus, _, err := builder.Build()
if err != nil {
	return err
}
reader := messenger.BindQuerier(bus, findArticle)
article, err := reader.Query(ctx, FindArticle{ID: 42})

Query[Q,R] uses the codec and schema only for request identity; R is an in-process type identity and is not written to the manifest. Every registered query must have exactly one handler and one built-in local route. The async route uses the caller context for admission, execution, and waiting, so cancellation returns ctx.Err() and accepted result delivery cannot block runtime drain. The runnable version is compiled as ExampleMessenger_Query.

Descriptors use explicit stable wire names, schema versions, content types, data encodings, and optional schema URIs. Go type names and package paths never become the wire contract implicitly. Native envelopes carry dataEncoding as json, text, or binary, so custom codecs do not infer their representation from a media-type prefix.

Local method-call benchmarks

Representative GitHub-hosted benchmark results for the synchronous local public path (Linux/amd64, Intel Xeon Platinum 8573C, Go 1.27, ten samples):

Public method Scenario Median time/op Approx. calls/s B/op allocs/op
Messenger.Send one local command handler 1.066 µs ~938k 1472 9
Messenger.Query one local query handler and typed result 937.6 ns ~1.07M 1296 10
Messenger.Publish one local event subscriber 1.086 µs ~921k 1472 9

The calls/s column is the reciprocal of the median single-thread time/op, not a concurrency or durable-throughput claim. These benchmarks exercise NewLocalSyncRoute with no-op application handlers. They do not include NATS, Kafka, Outbox, Inbox, SQL, network latency, retries, or telemetry exporters. Reproduce the sample with:

GOWORK=off go test -run '^$' -bench '^BenchmarkMessengerLocal' -benchmem -count=10 .

GitHub Actions retains the raw samples and benchstat comparison. Command and query results in this snapshot did not change significantly from the base commit.

Transactional outbox producer

Use adapters/outbox when a business write and message publication must commit or roll back together. Register the broker relay before the producer can stage its job:

natsRoute, err := natsadapter.NewRoute(natsConnection, natsadapter.RouteConfig{
	Name:      "nats.integration-events",
	Namespace: "prod",
	WireMode:  natsadapter.WireNative,
})
if err != nil {
	return err
}
relay, err := outboxadapter.NewRelayJob(natsRoute, outboxadapter.RelayJobConfig{})
if err != nil {
	return err
}
if err := outboxRuntime.Service().RegisterJob(relay); err != nil {
	return err
}

route, err := outboxadapter.NewProducer(
	outboxRuntime.Service(),
	outboxadapter.ProducerConfig{Name: "outbox.integration-events"},
)
if err != nil {
	return err
}
builder := messenger.NewBuilder(
	messenger.WithSource("urn:service:media-resizer"),
)
builder.RouteEvent(mediaResized, route)

bus, _, err := builder.Build()
if err != nil {
	return err
}

err = outboxRuntime.Transactor().RunInTx(ctx, func(txCtx context.Context) error {
	if err := mediaRepository.Save(txCtx, media); err != nil {
		return err
	}
	_, err := bus.Publish(txCtx, mediaResized, MediaResized{JobID: media.ID})
	return err
})

The repository write must use the transaction carried by txCtx. ReceiptStaged reports staging in that active transaction; a callback rollback still removes the row. The producer stores the canonical envelope under its MessageID; the relay later publishes those exact bytes and waits for PubAck. Repeating identical content resolves the same outbox tombstone, while reusing the identity with different content fails closed. Immediate messages use their immutable message time as availableAt, preventing retry-time fingerprint drift.

messenger.RetryAfter from the broker route becomes a persisted outbox.RetryAt; permanent envelope failures move directly to the outbox DLQ.

Durable JetStream consumer and inbox

A consumer owns one stable ConsumerID, explicit concurrency and retry bounds, and one SQL inbox:

if err := inboxpgsql.Migrate(ctx, database); err != nil {
	return err
}
store, err := inboxpgsql.New(database)
if err != nil {
	return err
}
consumer, err := natsadapter.NewEventConsumer(
	natsConnection,
	store,
	mediaResized,
	func(ctx context.Context, message messenger.Message[MediaResized]) error {
		tx, ok := inbox.SQLTxFromContext(ctx)
		if !ok {
			return errors.New("missing inbox transaction")
		}
		return projection.ApplyTx(ctx, tx, message.Metadata.ID.String(), message.Payload)
	},
	natsadapter.HandlerConfig{
		Stream: "MESSAGES", Namespace: "prod", ConsumerID: "media-projection", WireMode: natsadapter.WireNative,
		Concurrency: 8, Timeout: 30 * time.Second, FinalizationTimeout: 5 * time.Second,
		AckWait: 30 * time.Second, MaxAttempts: 10,
		DLQSubject: "prod.dlq",
	},
)
if err != nil {
	return err
}
consumerBuilder := messenger.NewBuilder(
	messenger.WithSource("urn:service:media-projection"),
)
consumerBuilder.Use("consumer.media-projection", consumer)

_, runtime, err := consumerBuilder.Build()
if err != nil {
	return err
}

Run the embedded additive inbox migrations explicitly with inboxpgsql.Migrate or inboxsqlite.Migrate before constructing consumers. They include the durable handler-attempt count and permanent-outcome state required by durable consumers. Hosts own database connections, migration ordering, NATS connections, credentials, and process supervision.

Each active consumer worker holds one SQL transaction for the full handler invocation. This is the atomic boundary for business writes and the Inbox completion marker; do not move the handler outside it. The adapter adds no connection semaphore: database/sql is the host-managed backpressure boundary. When several consumers share one sql.DB, configure SetMaxOpenConns to at least the sum of their HandlerConfig.Concurrency values, plus explicit headroom for application and maintenance queries. A smaller pool becomes the bottleneck before Go dispatch and can exhaust handler deadlines before application code runs.

Timeout bounds application handler execution. The Inbox transaction receives an additional FinalizationTimeout (5 seconds by default) to commit or roll back after that deadline. Increase it when a remote or otherwise slow database needs more finalization time; it does not extend the handler deadline.

Both SQL backends default to the gomessenger_ prefix and therefore preserve gomessenger_inbox, gomessenger_inbox_attempts, and gomessenger_inbox_attempt_generations. Pass the same options to migration and runtime construction when a host needs another namespace:

postgresInbox := []inboxpgsql.Option{
	inboxpgsql.WithSchema("messaging"), // the host creates the schema and grants access
	inboxpgsql.WithTablePrefix("site_"),
}
if err := inboxpgsql.Migrate(ctx, database, postgresInbox...); err != nil {
	return err
}
store, err := inboxpgsql.New(database, postgresInbox...)

WithTablePrefix is also available for SQLite; SQLite has no schema option. PostgreSQL qualifies every relation instead of relying on search_path, and its migrator never creates the configured schema. Changing the prefix or schema selects a separate Inbox: migrations do not rename, copy, or otherwise transfer existing deduplication history.

Provision source and DLQ capacity separately. DevStream keeps native source messages at the 1 MiB envelope bound; DevDLQStream reserves DefaultMaxDLQMessageBytes for the expanded JSON DLQ record. The NATS server/account max_payload must be at least the same value. Consumer.Run rejects a missing or undersized DLQ route before starting; the low-frequency Consumer.DeepHealth probe detects later topology drift without making ordinary readiness expensive.

Middleware, logging, and tracing

Global middleware wraps local query/command/event handlers and can be reused by durable NATS consumers. Registration order is execution order: the first middleware is outermost. A middleware may replace the context or short-circuit, but next may be called at most once. Typed one-way decorators use messenger.ChainHandler; typed query decorators use messenger.ChainQueryHandler and may return a cached or synthetic R. Global middleware cannot synthesize a typed result: successful completion without one returns ErrQueryResultMissing.

logger := messenger.AdaptSlog(slog.Default())
builder := messenger.NewBuilder(
	messenger.WithSource("urn:service:media-resizer"),
	messenger.WithLogger(logger),
	messenger.WithObserver(messenger.NewLoggingObserver(logger)),
	messenger.WithContextPropagator(observability.NewTraceContextPropagator()),
)
builder.UseMiddleware(func(
	ctx context.Context,
	metadata messenger.Metadata,
	handlerID string,
	next messenger.HandlerFunc,
) error {
	return next(ctx)
})

Logging is disabled by default. AdaptSlog(nil) is a safe no-op adapter; direct WithLogger(nil) is a configuration error. Observer registrations are additive. A panic in one observer is logged and isolated from the remaining observers. Kafka TransportConfig.Logger reports adapter-owned startup/readiness, producer and consumer transaction, abort/fencing, topology failures or applied changes, and retry partition deferrals. Core logging contains infrastructure state only and never logs record keys, payloads, message bodies, or arbitrary headers. WithClientLogger is a separate explicit opt-in to franz-go's own client logs.

Recovered handler panic values and stacks are dropped by default; ordinary errors satisfy the transport-neutral HandlerPanicError interface and can be classified with errors.As across independently versioned adapters. Configure WithPanicReporter (or the corresponding durable consumer field) only for a trusted diagnostic sink. Consumer observations and DLQ errors use the conservative DefaultFailureSanitizer unless the host explicitly supplies another sanitizer.

The observability propagator carries only W3C traceparent and tracestate. It works through native envelopes, CloudEvents structured/binary modes, and transactional Outbox storage. Baggage is intentionally not supported yet.

Runtime lifecycle

Run the managed services under the host supervisor. On a signal, close admission first and then wait for accepted work within a deadline:

signalCtx, stopSignals := signal.NotifyContext(
	context.Background(),
	os.Interrupt,
	syscall.SIGTERM,
)
defer stopSignals()

runErr := make(chan error, 1)
go func() {
	runErr <- runtime.Run(context.Background())
}()

select {
case err := <-runErr:
	return err
case <-signalCtx.Done():
}

runtime.BeginDrain()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
if err := runtime.Shutdown(shutdownCtx); err != nil {
	return err
}
return <-runErr

Runtime.Readiness is the lightweight admission/connectivity probe for the runtime and every attached service. Runtime.DeepHealth performs explicit topology diagnostics; Runtime.Liveness does not require readiness. The selected Outbox backend runtime has its own host-supervised lifecycle; GoMessenger does not open connections or supervise it implicitly. Never call Shutdown from a handler executing on the same runtime.

Failure semantics

  • Return messenger.Permanent(err) for a non-retryable handler failure.
  • Return messenger.RetryAfter(err, delay) for an explicit durable delay.
  • Other errors use bounded full-jitter exponential retry in the NATS and Kafka adapters.
  • MaxAttempts bounds handler invocations, not broker deliveries. A NotBefore deferral does not invoke the handler or consume an attempt.
  • AckWait is a broker redelivery deadline, not a handler timeout. It must be at least 100 ms and may be shorter than Timeout; active handlers refresh acknowledgement progress every AckWait / 3.
  • The NATS adapter publishes and confirms a DLQ record before broker-confirming acknowledgement of the original message. Failed terminal hand-off operations retry on the current delivery without invoking the application handler beyond MaxAttempts. A permanent outcome is persisted independently of the attempt count, so an interrupted hand-off cannot invoke that handler again after restart.
  • NotBefore becomes a retry delay until due; ExpiresAt becomes a permanent expired outcome.

Kafka

The independent Kafka module provides native-envelope transactional publish, Outbox relay, read-committed consumers, consumer-specific retry topics, atomic retry/DLQ offset hand-off, protected replay, static worker identity, and non-destructive topic planning. It intentionally does not reuse the NATS engine or support CloudEvents/query transport. Retry may be overtaken by later source records, so ordering is guaranteed only before the first failure. See the Kafka adapter guide and ADR-0004.

Level 2 roadmap

The next maturity level is an operational messaging platform, not a workflow framework. Work is ordered evidence-first: pilot and performance baseline, schema compatibility, broker capability declarations, partition-key and ordering semantics, batching only when profiling justifies it, then load, soak, and chaos validation. Saga engines, workflow orchestration, and generic distributed request/reply remain out of scope.

See the Level 2 roadmap for workstreams and exit criteria.

CloudEvents

The NATS adapter supports the native envelope for commands and events, plus CloudEvents 1.0 structured and binary modes for events. Native metadata is mapped explicitly; delivery-attempt state never enters the canonical envelope or fingerprint. An omitted CloudEvents time is derived deterministically from a UUIDv7 event ID; events without time and with a non-UUIDv7 ID are rejected so retries cannot change the canonical fingerprint. The required dataencoding extension carries json, text, or binary; missing, invalid, or descriptor-conflicting values are rejected before the payload codec runs.

Topology and CLI

Topology management is declarative and non-destructive. It may create missing resources or apply compatible additive changes. Retention/storage/replica changes, subject removal, reduced limits, or consumer delivery-contract drift are conflicts; the tool never deletes and recreates resources automatically. Compatible updates preserve broker settings outside the manifest's managed subset.

gomessengerctl manifest validate --file manifest.json
gomessengerctl topology validate --file topology.json
gomessengerctl topology plan --file topology.json --server nats://localhost:4222
gomessengerctl topology apply --file topology.json --server nats://localhost:4222
gomessengerctl dlq inspect --file record.json
gomessengerctl dlq replay --file record.json
gomessengerctl dlq replay --file record.json --confirm --server nats://localhost:4222
gomessengerctl kafka topology validate --file kafka-topology.json
gomessengerctl kafka topology plan --file kafka-topology.json --brokers localhost:9092 --instance-id ops-a
gomessengerctl kafka topology apply --file kafka-topology.json --brokers localhost:9092 --instance-id ops-a
gomessengerctl kafka dlq inspect --file kafka-record.json
gomessengerctl kafka dlq replay --file kafka-record.json
gomessengerctl kafka dlq replay --file kafka-record.json --confirm --brokers localhost:9092 --instance-id ops-a

topology plan exits with code 3 when the printed plan contains a conflict. dlq inspect prints a safe summary and replayability status without handler error text, wire bytes, or header values. dlq replay is offline by default and prints a payload-free deterministic JSON plan. --confirm republishes the original subject, wire bytes, and bounded headers with a deterministic, DLQ-record-specific Nats-Msg-Id, then waits for JetStream PubAck. The target consumer starts one fresh bounded attempt generation even if post-ACK Inbox cleanup was interrupted; broker redeliveries of that replay retain the same generation and do not reset MaxAttempts. Replay does not support subject substitution, record deletion, or payload/header output. Its internal replay headers are reserved transport metadata and must not be injected by ordinary publishers; enforce that boundary with NATS publish permissions when producers can access subjects directly.

Development and verification

make prepare
make check
make test-e2e
make test-integration
make test-kafka
GOMESSENGER_POSTGRES_DSN='postgres://...' make test-postgres
make bench-all

make check covers every module, static lint, race and checkptr builds, a 90% root coverage gate, an isolated clean-consumer module, and the Docker-free durable pipeline E2E. make test-e2e reruns only that full Outbox-to-JetStream-to-Inbox path. make test-kafka is the local Docker entry point against official Kafka 4.1.2 and 4.3.1 images; hosted CI runs each version in an independent matrix job. Release requirements are prepared and checked explicitly:

make check
make release-ready VERSION=v0.2.1 OUTBOX_VERSION=v0.12.0
make release-readiness VERSION=v0.2.1 OUTBOX_VERSION=v0.12.0

Run the full source gate before release-ready removes development replacements. release-readiness then checks the exact tag declarations without trying to download GoMessenger tags that do not exist yet.

A published release is verified separately, after all dependency-ordered module tags exist:

make test-consumer-release VERSION=v0.2.1

GitHub Actions runs static, race, checkptr, PostgreSQL, and Kafka integration shards; the aggregate Full gate requires all of them. A separate workflow compares base and head benchmarks with pinned benchstat, uploads raw samples, and reports when the base predates the Go module instead of failing the first comparison. It does not enforce an unstable cross-machine performance threshold.

See the practical usage guide, Kafka guide, contracts, architecture, durable pipeline E2E, use-case comparison, Level 2 roadmap, adoption and migration guide, release order, and the first real-project pilot decision. Distributed request/reply remains a separate, unimplemented boundary in ADR-0003.

Documentation

Overview

Package messenger provides typed commands, local queries, and events with explicit descriptors, GoBus dispatch, and transport-neutral one-way delivery contracts. Queries are process-local request/reply calls and never use the wire envelope or durable routes.

Index

Examples

Constants

View Source 
const (
	// EnvelopeSpecVersion is the current native envelope contract.
	EnvelopeSpecVersion = "1.0"
	// DefaultMaxEnvelopeBytes is the default encoded envelope limit.
	DefaultMaxEnvelopeBytes = 1 << 20
	// DefaultMaxHeaders is the default number of application headers.
	DefaultMaxHeaders = 64
	// DefaultMaxHeaderBytes is the default aggregate application-header limit.
	DefaultMaxHeaderBytes = 16 << 10
)
View Source 
const ManifestSpecVersion = "1.0"

ManifestSpecVersion is the current topology manifest contract.

Variables

View Source 
var (
	// ErrInvalidDescriptor reports an invalid command, event, or query descriptor.
	ErrInvalidDescriptor = errors.New("messenger: invalid descriptor")
	// ErrInvalidMessage reports invalid outgoing metadata or envelope data.
	ErrInvalidMessage = errors.New("messenger: invalid message")
	// ErrDescriptorConflict reports two incompatible descriptors with one wire identity.
	ErrDescriptorConflict = errors.New("messenger: descriptor conflict")
	// ErrHandlerConflict reports a duplicate command handler or subscription ID.
	ErrHandlerConflict = errors.New("messenger: handler conflict")
	// ErrHandlerNotFound reports a missing required local handler.
	ErrHandlerNotFound = errors.New("messenger: handler not found")
	// ErrQueryResultMissing reports successful global middleware completion without a query result.
	ErrQueryResultMissing = errors.New("messenger: query result missing")
	// ErrRouteConflict reports more than one primary route for a descriptor.
	ErrRouteConflict = errors.New("messenger: route conflict")
	// ErrRouteNotFound reports that a descriptor has no outbound route.
	ErrRouteNotFound = errors.New("messenger: route not found")
	// ErrUnsupportedCapability reports a requested semantic guarantee that a route cannot provide.
	ErrUnsupportedCapability = errors.New("messenger: unsupported route capability")
	// ErrMessageExpired reports a message whose expiration boundary has been reached.
	ErrMessageExpired = errors.New("messenger: message expired")
	// ErrMessageNotReady reports a message whose not-before boundary is still in the future.
	ErrMessageNotReady = errors.New("messenger: message not ready")
	// ErrServiceConflict reports a duplicate managed service ID.
	ErrServiceConflict = errors.New("messenger: service conflict")
	// ErrRuntimeNotRunning reports an operation that requires a running runtime.
	ErrRuntimeNotRunning = errors.New("messenger: runtime not running")
	// ErrRuntimeRunning reports a second concurrent call to Runtime.Run.
	ErrRuntimeRunning = errors.New("messenger: runtime already running")
	// ErrRuntimeClosed reports use after a runtime has shut down.
	ErrRuntimeClosed = errors.New("messenger: runtime closed")
	// ErrEnvelopeTooLarge reports an envelope beyond the configured wire limit.
	ErrEnvelopeTooLarge = errors.New("messenger: envelope too large")
)

Functions

func BoundedFailureText added in v0.2.0 

func BoundedFailureText(sanitizer FailureSanitizer, err error, limit int) string

BoundedFailureText returns sanitized, valid UTF-8 text no longer than limit bytes. It never splits a UTF-8 code point.

func CanonicalizeEnvelope 

func CanonicalizeEnvelope(data []byte) ([]byte, error)

CanonicalizeEnvelope parses and re-encodes an envelope using the native deterministic field order. Delivery metadata is never included.

func ContextWithMetadata 

func ContextWithMetadata(ctx context.Context, metadata Metadata) context.Context

ContextWithMetadata installs immutable message lineage for an adapter or terminal handler entering the typed messenger boundary. As with standard context helpers, ctx must be non-nil.

func DecodeCommandPayload 

func DecodeCommandPayload[T any](descriptor Command[T], data []byte) (T, error)

DecodeCommandPayload decodes codec bytes without an envelope.

func DecodeEventPayload 

func DecodeEventPayload[T any](descriptor Event[T], data []byte) (T, error)

DecodeEventPayload decodes codec bytes without an envelope.

func EncodeCommandEnvelope 

func EncodeCommandEnvelope[T any](descriptor Command[T], metadata Metadata, payload T) ([]byte, error)

EncodeCommandEnvelope encodes a typed command with already resolved metadata.

func EncodeEventEnvelope 

func EncodeEventEnvelope[T any](descriptor Event[T], metadata Metadata, payload T) ([]byte, error)

EncodeEventEnvelope encodes a typed event with already resolved metadata.

func EnvelopeFingerprint 

func EnvelopeFingerprint(data []byte) [sha256.Size]byte

EnvelopeFingerprint returns SHA-256 over canonical encoded envelope bytes.

func HandlerCompletionError added in v0.2.0 

func HandlerCompletionError(ctx context.Context, handlerErr error) error

HandlerCompletionError prevents a handler that returns nil after its context deadline from committing its transaction. Handler deadlines remain cooperative: a handler must still observe ctx.Done to stop promptly.

func IsPermanent 

func IsPermanent(err error) bool

IsPermanent reports whether err contains a Permanent marker.

func MarshalEnvelope 

func MarshalEnvelope(metadata Metadata, payload []byte, encoding DataEncoding) ([]byte, error)

MarshalEnvelope validates and encodes an envelope with a codec payload.

func Permanent 

func Permanent(err error) error

Permanent marks an error as non-retryable.

func ReportHandlerPanic added in v0.2.0 

func ReportHandlerPanic(
	ctx context.Context,
	reporter PanicReporter,
	handlerID string,
	recovered any,
	stack []byte,
) error

ReportHandlerPanic sends sensitive details to the optional reporter and returns a safe error suitable for retries, observations, logs, and DLQ data.

func RetryAfter 

func RetryAfter(err error, delay time.Duration) error

RetryAfter asks a durable transport to retry after an exact positive delay.

func RetryDelay 

func RetryDelay(err error) (time.Duration, bool)

RetryDelay returns an explicitly requested retry delay.

func SanitizeError added in v0.2.0 

func SanitizeError(sanitizer FailureSanitizer, err error) error

SanitizeError preserves errors.Is/errors.As through Unwrap while exposing only sanitized text through Error.

func SanitizeFailure added in v0.2.0 

func SanitizeFailure(sanitizer FailureSanitizer, err error) string

SanitizeFailure returns safe failure text. A nil or typed-nil sanitizer uses DefaultFailureSanitizer.

Types

type Builder 

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

Builder declares immutable descriptors, handlers, routes, and managed services. It is not safe for concurrent mutation.

func NewBuilder 

func NewBuilder(options ...Option) *Builder

NewBuilder constructs an empty messenger builder.

func (*Builder) Build 

func (b *Builder) Build() (*Messenger, *Runtime, error)

Build validates and freezes the declared topology.

func (*Builder) HandleCommand 

func (b *Builder) HandleCommand[T any](descriptor Command[T], handlerID string, handler Handler[T])

HandleCommand registers the one local handler for a command descriptor. Validation errors are returned by Build.

func (*Builder) HandleCommandFunc 

func (b *Builder) HandleCommandFunc[T any](
	descriptor Command[T],
	handlerID string,
	handler PayloadHandler[T],
)

HandleCommandFunc registers a payload-only command handler.

func (*Builder) HandleQuery 

func (b *Builder) HandleQuery[Q, R any](
	descriptor Query[Q, R],
	handlerID string,
	handler QueryHandler[Q, R],
)

HandleQuery registers the one local handler for a query descriptor. Validation errors are returned by Build.

func (*Builder) HandleQueryFunc 

func (b *Builder) HandleQueryFunc[Q, R any](
	descriptor Query[Q, R],
	handlerID string,
	handler QueryPayloadHandler[Q, R],
)

HandleQueryFunc registers a payload-only local query handler.

func (*Builder) RouteCommand 

func (b *Builder) RouteCommand[T any](descriptor Command[T], route Route)

RouteCommand sets the command's one primary outbound route.

func (*Builder) RouteEvent 

func (b *Builder) RouteEvent[T any](descriptor Event[T], route Route)

RouteEvent sets the event's one primary outbound route.

func (*Builder) RouteQuery 

func (b *Builder) RouteQuery[Q, R any](descriptor Query[Q, R], route LocalQueryRoute)

RouteQuery sets the query's required local request/reply route.

func (*Builder) Subscribe 

func (b *Builder) Subscribe[T any](descriptor Event[T], subscriptionID string, handler Handler[T])

Subscribe appends a named local event subscription. Validation errors are returned by Build.

func (*Builder) SubscribeFunc 

func (b *Builder) SubscribeFunc[T any](
	descriptor Event[T],
	subscriptionID string,
	handler PayloadHandler[T],
)

SubscribeFunc appends a payload-only event subscription.

func (*Builder) Use 

func (b *Builder) Use(serviceID string, service Service)

Use adds a named managed consumer or worker service to the returned Runtime.

func (*Builder) UseMiddleware 

func (b *Builder) UseMiddleware(middlewares ...Middleware)

UseMiddleware appends global handler middleware. Validation errors are returned by Build.

type Codec 

type Codec[T any] interface {
	Encode(value T) ([]byte, error)
	Decode(data []byte) (T, error)
	ContentType() string
	Encoding() DataEncoding
}

Codec encodes and decodes one descriptor payload type.

func Bytes 

func Bytes() Codec[[]byte]

Bytes returns a binary codec that copies byte slices at the boundary.

func CustomCodec 

func CustomCodec[T any](
	contentType string,
	encoding DataEncoding,
	encode func(T) ([]byte, error),
	decode func([]byte) (T, error),
) (Codec[T], error)

CustomCodec constructs a codec with an explicit content type and wire encoding.

func JSON 

func JSON[T any]() Codec[T]

JSON returns the standard JSON codec for T.

func Text 

func Text() Codec[string]

Text returns a UTF-8 text codec.

type Command 

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

Command is an immutable typed command descriptor.

func MustCommand 

func MustCommand[T any](name string, schemaVersion int, codec Codec[T]) Command[T]

MustCommand constructs a command descriptor and panics when its declaration is invalid.

func NewCommand 

func NewCommand[T any](name string, schemaVersion int, codec Codec[T]) (Command[T], error)

NewCommand constructs a command descriptor.

func (Command[T]) Info 

func (d Command[T]) Info() DescriptorInfo

Info returns a copy of the command's wire identity.

func (Command[T]) WithSchema 

func (d Command[T]) WithSchema(schema string) Command[T]

WithSchema returns a command descriptor with an explicit schema URI.

type ContextPropagator 

type ContextPropagator interface {
	Inject(ctx context.Context, carrier map[string]string)
	Extract(ctx context.Context, carrier map[string]string) context.Context
}

ContextPropagator injects and extracts distributed context through immutable message headers.

func NoopContextPropagator 

func NoopContextPropagator() ContextPropagator

NoopContextPropagator returns a propagator that leaves contexts and carriers unchanged.

type DataEncoding 

type DataEncoding uint8

DataEncoding selects how encoded payload bytes appear in envelope JSON. 

const (
	// DataJSON stores codec output directly in the envelope data field.
	DataJSON DataEncoding = iota + 1
	// DataText stores codec output as a JSON string in the envelope data field.
	DataText
	// DataBinary stores codec output in the envelope dataBase64 field.
	DataBinary
)

func (DataEncoding) MarshalJSON 

func (e DataEncoding) MarshalJSON() ([]byte, error)

MarshalJSON encodes a data encoding as its stable wire name.

func (DataEncoding) String 

func (e DataEncoding) String() string

String returns the stable wire name of the data encoding.

func (*DataEncoding) UnmarshalJSON 

func (e *DataEncoding) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes a stable data-encoding wire name.

type DeepHealthChecker added in v0.2.0 

type DeepHealthChecker interface {
	DeepHealth(ctx context.Context) error
}

DeepHealthChecker optionally performs expensive topology and infrastructure validation outside the normal readiness probe path.

type Delivery 

type Delivery interface {
	Metadata() Metadata
	HandlerCount() int
	MarshalEnvelope() ([]byte, error)
	Fingerprint() ([sha256.Size]byte, error)
	Invoke(ctx context.Context) error
}

Delivery is the transport-neutral route input. MarshalEnvelope is lazy, so local routes do not serialize payloads. Invoke is intended for local and terminal durable adapters.

type DescriptorInfo 

type DescriptorInfo struct {
	Kind          Kind         `json:"kind"`
	Name          string       `json:"name"`
	SchemaVersion int          `json:"schemaVersion"`
	ContentType   string       `json:"contentType"`
	DataEncoding  DataEncoding `json:"dataEncoding"`
	Schema        string       `json:"schema,omitempty"`
}

DescriptorInfo is the transport-neutral public identity of a descriptor.

type Envelope 

type Envelope struct {
	SpecVersion   string            `json:"specVersion"`
	ID            MessageID         `json:"id"`
	Kind          Kind              `json:"kind"`
	Name          string            `json:"name"`
	SchemaVersion int               `json:"schemaVersion"`
	Source        string            `json:"source"`
	Subject       string            `json:"subject,omitempty"`
	Time          time.Time         `json:"time"`
	CorrelationID MessageID         `json:"correlationId"`
	CausationID   MessageID         `json:"causationId,omitzero"`
	Key           string            `json:"key,omitempty"`
	ContentType   string            `json:"contentType"`
	DataEncoding  DataEncoding      `json:"dataEncoding"`
	Schema        string            `json:"schema,omitempty"`
	Headers       map[string]string `json:"headers,omitempty"`
	NotBefore     time.Time         `json:"notBefore,omitzero"`
	ExpiresAt     time.Time         `json:"expiresAt,omitzero"`
	Data          json.RawMessage   `json:"data,omitempty"`
	DataBase64    *string           `json:"dataBase64,omitempty"`
}

Envelope is the canonical native wire representation.

func UnmarshalEnvelope 

func UnmarshalEnvelope(data []byte) (Envelope, error)

UnmarshalEnvelope parses and validates a native envelope.

func (Envelope) Metadata 

func (e Envelope) Metadata() Metadata

Metadata returns a defensive copy of the envelope metadata.

func (Envelope) Payload 

func (e Envelope) Payload() ([]byte, DataEncoding, error)

Payload returns decoded codec bytes and their envelope representation.

func (Envelope) Validate 

func (e Envelope) Validate() error

Validate checks the native envelope invariants and configured default bounds.

type Event 

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

Event is an immutable typed event descriptor.

func MustEvent 

func MustEvent[T any](name string, schemaVersion int, codec Codec[T]) Event[T]

MustEvent constructs an event descriptor and panics when its declaration is invalid.

func NewEvent 

func NewEvent[T any](name string, schemaVersion int, codec Codec[T]) (Event[T], error)

NewEvent constructs an event descriptor.

func (Event[T]) Info 

func (d Event[T]) Info() DescriptorInfo

Info returns a copy of the event's wire identity.

func (Event[T]) WithSchema 

func (d Event[T]) WithSchema(schema string) Event[T]

WithSchema returns an event descriptor with an explicit schema URI.

type FailureSanitizer added in v0.2.0 

type FailureSanitizer interface {
	SanitizeFailure(err error) string
}

FailureSanitizer converts an error to text safe for operational channels such as default logs, traces, and dead-letter records.

func DefaultFailureSanitizer added in v0.2.0 

func DefaultFailureSanitizer() FailureSanitizer

DefaultFailureSanitizer returns the conservative built-in sanitizer. Hosts may opt in to richer text with an explicit FailureSanitizer implementation.

type FailureSanitizerFunc added in v0.2.0 

type FailureSanitizerFunc func(error) string

FailureSanitizerFunc adapts a function to FailureSanitizer.

func (FailureSanitizerFunc) SanitizeFailure added in v0.2.0 

func (f FailureSanitizerFunc) SanitizeFailure(err error) string

SanitizeFailure implements FailureSanitizer.

type Handler 

type Handler[T any] func(context.Context, Message[T]) error

Handler processes one typed message.

func ChainHandler 

func ChainHandler[T any](handler Handler[T], middlewares ...HandlerMiddleware[T]) Handler[T]

ChainHandler applies typed middleware with the first item outermost. It returns nil when handler, a middleware, or a middleware result is nil so the receiving Builder or durable consumer can reject the invalid chain.

func HandlePayload 

func HandlePayload[T any](handler PayloadHandler[T]) Handler[T]

HandlePayload adapts a payload-only handler to the primary Handler contract.

type HandlerFunc 

type HandlerFunc func(context.Context) error

HandlerFunc is the transport-neutral terminal handler shape used by global middleware.

type HandlerMiddleware 

type HandlerMiddleware[T any] func(Handler[T]) Handler[T]

HandlerMiddleware wraps a typed handler.

type HandlerPanicError added in v0.2.0 

type HandlerPanicError interface {
	error
	HandlerPanicID() string
}

HandlerPanicError is the transport-neutral safe view of a recovered handler or middleware panic. Independently versioned adapters implement this interface structurally without exposing the recovered value or stack.

type IDGenerator 

type IDGenerator interface {
	New() (MessageID, error)
}

IDGenerator creates stable message identities.

func UUIDv7Generator 

func UUIDv7Generator() IDGenerator

UUIDv7Generator returns the default cryptographically random UUIDv7 generator.

type Kind 

type Kind string

Kind identifies the semantic message category. 

const (
	// KindCommand identifies a command with one logical handler.
	KindCommand Kind = "command"
	// KindEvent identifies an event with zero or more subscriptions.
	KindEvent Kind = "event"
	// KindQuery identifies a local request/reply query with one handler.
	KindQuery Kind = "query"
)

type LivenessChecker added in v0.2.0 

type LivenessChecker interface {
	Liveness(ctx context.Context) error
}

LivenessChecker optionally separates process liveness from readiness and transient broker or topology failures.

type LocalAsyncConfig 

type LocalAsyncConfig struct {
	Capacity int
	Workers  int
	// DetachExecution is retained for source compatibility. Accepted one-way
	// jobs always detach execution from the caller's cancellation and deadline.
	// Query calls always retain the caller context for execution and waiting.
	//
	// Deprecated: caller context controls admission only.
	DetachExecution bool
}

LocalAsyncConfig bounds local asynchronous admission and execution.

type LocalAsyncRoute 

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

LocalAsyncRoute admits handler calls to a bounded GoBus async runtime.

func NewLocalAsyncRoute 

func NewLocalAsyncRoute(name string, config LocalAsyncConfig) (*LocalAsyncRoute, error)

NewLocalAsyncRoute constructs a named bounded local asynchronous route.

func (*LocalAsyncRoute) BeginDrain 

func (r *LocalAsyncRoute) BeginDrain()

BeginDrain rejects new work and drains accepted jobs.

func (*LocalAsyncRoute) Deliver 

func (r *LocalAsyncRoute) Deliver(ctx context.Context, delivery Delivery) (Receipt, error)

Deliver implements Route and reports admission, not handler completion.

func (*LocalAsyncRoute) ManagedService 

func (r *LocalAsyncRoute) ManagedService() (string, Service)

ManagedService exposes this route to Builder runtime aggregation.

func (*LocalAsyncRoute) Name 

func (r *LocalAsyncRoute) Name() string

Name implements Route.

func (*LocalAsyncRoute) Readiness 

func (r *LocalAsyncRoute) Readiness(context.Context) error

Readiness verifies that this route is accepting work.

func (*LocalAsyncRoute) Run 

func (r *LocalAsyncRoute) Run(ctx context.Context) error

Run starts queue workers and blocks until cancellation or draining.

func (*LocalAsyncRoute) Shutdown 

func (r *LocalAsyncRoute) Shutdown(ctx context.Context) error

Shutdown drains or force-cancels the route within ctx.

type LocalQueryRoute 

type LocalQueryRoute interface {
	Name() string
	// contains filtered or unexported methods
}

LocalQueryRoute is the sealed local request/reply route contract. The built-in LocalSyncRoute and LocalAsyncRoute are its only implementations.

type LocalSyncRoute 

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

LocalSyncRoute executes handlers synchronously through a private GoBus instance.

func NewLocalSyncRoute 

func NewLocalSyncRoute() *LocalSyncRoute

NewLocalSyncRoute constructs a local synchronous route.

func (*LocalSyncRoute) Deliver 

func (r *LocalSyncRoute) Deliver(ctx context.Context, delivery Delivery) (Receipt, error)

Deliver implements Route.

func (*LocalSyncRoute) Name 

func (*LocalSyncRoute) Name() string

Name implements Route.

type LogAttr 

type LogAttr struct {
	Key   string
	Value any
}

LogAttr is one structured logging attribute.

type LogLevel 

type LogLevel uint8

LogLevel is the transport-neutral severity understood by Logger. 

const (
	// LogDebug records diagnostic lifecycle details.
	LogDebug LogLevel = iota
	// LogInfo records ordinary lifecycle transitions.
	LogInfo
	// LogWarn records recoverable infrastructure failures.
	LogWarn
	// LogError records infrastructure failures requiring attention.
	LogError
)

type Logger 

type Logger interface {
	Log(ctx context.Context, level LogLevel, message string, attrs ...LogAttr)
}

Logger is the minimal structured logging contract used by GoMessenger. Implementations must not retain ctx and should return quickly.

func AdaptSlog 

func AdaptSlog(logger *slog.Logger) Logger

AdaptSlog adapts a standard slog logger. A nil logger returns a no-op adapter, which makes optional host wiring safe.

type Manifest 

type Manifest struct {
	SpecVersion string               `json:"specVersion"`
	Source      string               `json:"source"`
	Descriptors []ManifestDescriptor `json:"descriptors"`
	Services    []string             `json:"services,omitempty"`
}

Manifest is a deterministic, secret-free description of runtime topology.

func (Manifest) Validate 

func (m Manifest) Validate() error

Validate checks manifest structure without constructing a runtime.

type ManifestDescriptor 

type ManifestDescriptor struct {
	DescriptorInfo
	Route      string   `json:"route,omitempty"`
	HandlerIDs []string `json:"handlerIds,omitempty"`
}

ManifestDescriptor describes one typed descriptor and its static route.

type Message 

type Message[T any] struct {
	Metadata Metadata
	Payload  T
}

Message is the typed value passed to a handler.

func DecodeCommand 

func DecodeCommand[T any](descriptor Command[T], data []byte) (Message[T], error)

DecodeCommand decodes and verifies a native command envelope.

func DecodeEvent 

func DecodeEvent[T any](descriptor Event[T], data []byte) (Message[T], error)

DecodeEvent decodes and verifies a native event envelope.

type MessageID 

type MessageID [16]byte

MessageID is a UUID-compatible 128-bit message identity.

func ParseMessageID 

func ParseMessageID(value string) (MessageID, error)

ParseMessageID parses the canonical UUID text form.

func (MessageID) IsZero 

func (id MessageID) IsZero() bool

IsZero reports whether no message identity is set.

func (MessageID) MarshalJSON 

func (id MessageID) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler.

func (MessageID) MarshalText 

func (id MessageID) MarshalText() ([]byte, error)

MarshalText implements encoding.TextMarshaler.

func (MessageID) String 

func (id MessageID) String() string

String returns the canonical lowercase UUID text form.

func (*MessageID) UnmarshalJSON 

func (id *MessageID) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler.

func (*MessageID) UnmarshalText 

func (id *MessageID) UnmarshalText(text []byte) error

UnmarshalText implements encoding.TextUnmarshaler.

type Messenger 

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

Messenger sends commands, executes local queries, and publishes events through immutable descriptor bindings.

func (*Messenger) Manifest 

func (m *Messenger) Manifest() Manifest

Manifest returns a defensive copy of the messenger topology.

func (*Messenger) MarshalManifest 

func (m *Messenger) MarshalManifest() ([]byte, error)

MarshalManifest returns deterministic indented JSON suitable for gomessengerctl.

func (*Messenger) Publish 

func (m *Messenger) Publish[T any](ctx context.Context, descriptor Event[T], payload T) (Receipt, error)

Publish publishes an event with generated metadata.

func (*Messenger) PublishMessage 

func (m *Messenger) PublishMessage[T any](
	ctx context.Context,
	descriptor Event[T],
	outgoing Outgoing[T],
) (Receipt, error)

PublishMessage publishes an event with explicit optional metadata.

func (*Messenger) Query 

func (m *Messenger) Query[Q, R any](
	ctx context.Context,
	descriptor Query[Q, R],
	payload Q,
) (R, error)

Query executes a typed local request/reply call through its configured route.

Example 
package main

import (
	"context"
	"fmt"

	messenger "github.com/assurrussa/gomessenger"
)

func main() {
	type findArticle struct{ ID int64 }
	type articleView struct {
		ID    int64
		Title string
	}

	find := messenger.MustQuery[findArticle, articleView]("article.find", 1, messenger.JSON[findArticle]())
	builder := messenger.NewBuilder(messenger.WithSource("urn:service:catalog"))
	builder.HandleQueryFunc(find, "article-reader", func(_ context.Context, query findArticle) (articleView, error) {
		return articleView{ID: query.ID, Title: "CQRS in Go"}, nil
	})
	builder.RouteQuery(find, messenger.NewLocalSyncRoute())
	bus, _, err := builder.Build()
	if err != nil {
		panic(err)
	}

	reader := messenger.BindQuerier(bus, find)
	article, err := reader.Query(context.Background(), findArticle{ID: 42})
	if err != nil {
		panic(err)
	}
	fmt.Println(article.ID, article.Title)

}

Output:
42 CQRS in Go

func (*Messenger) Send 

func (m *Messenger) Send[T any](ctx context.Context, descriptor Command[T], payload T) (Receipt, error)

Send sends a command with generated metadata.

Example 
package main

import (
	"context"
	"fmt"

	messenger "github.com/assurrussa/gomessenger"
)

func main() {
	type resizeMedia struct {
		JobID int64 `json:"jobId"`
	}

	resize := messenger.MustCommand("media.resize", 1, messenger.JSON[resizeMedia]())
	builder := messenger.NewBuilder(messenger.WithSource("urn:service:media-resizer"))
	builder.HandleCommandFunc(resize, "media-worker", func(_ context.Context, payload resizeMedia) error {
		fmt.Println("handler", payload.JobID)
		return nil
	})
	builder.RouteCommand(resize, messenger.NewLocalSyncRoute())
	bus, _, err := builder.Build()
	if err != nil {
		panic(err)
	}
	resizeSender := messenger.BindSender(bus, resize)
	receipt, err := resizeSender.Send(context.Background(), resizeMedia{JobID: 42})
	if err != nil {
		panic(err)
	}
	fmt.Println(receipt.State)

}

Output:
handler 42
completed

func (*Messenger) SendMessage 

func (m *Messenger) SendMessage[T any](
	ctx context.Context,
	descriptor Command[T],
	outgoing Outgoing[T],
) (Receipt, error)

SendMessage sends a command with explicit optional metadata.

type Metadata 

type Metadata struct {
	ID            MessageID
	Kind          Kind
	Name          string
	SchemaVersion int
	Source        string
	Subject       string
	Time          time.Time
	CorrelationID MessageID
	CausationID   MessageID
	Key           string
	ContentType   string
	Schema        string
	Headers       map[string]string
	NotBefore     time.Time
	ExpiresAt     time.Time
}

Metadata is canonical message metadata independent of a transport attempt.

func MetadataFromContext 

func MetadataFromContext(ctx context.Context) (Metadata, bool)

MetadataFromContext returns the currently handled message metadata.

type Middleware 

type Middleware func(
	ctx context.Context,
	metadata Metadata,
	handlerID string,
	next HandlerFunc,
) error

Middleware wraps one local query/command/event or durable handler. The first registered middleware is the outermost wrapper and may short-circuit by not calling next.

type Observation 

type Observation struct {
	Operation     Operation
	MessageID     MessageID
	Kind          Kind
	Name          string
	SchemaVersion int
	Route         string
	HandlerID     string
	ConsumerID    string
	ServiceID     string
	Attempt       uint64
	Duplicate     bool
	RetryDelay    time.Duration
	State         ReceiptState
	StartedAt     time.Time
	Duration      time.Duration
	Err           error
}

Observation contains bounded operational data. Observers decide which fields are safe for low-cardinality metric labels.

type Observer 

type Observer interface {
	Observe(ctx context.Context, observation Observation)
}

Observer receives messaging lifecycle observations. Implementations must not retain ctx and should return quickly.

func NewLoggingObserver 

func NewLoggingObserver(logger Logger) Observer

NewLoggingObserver reports sanitized observations through logger. Successful operations use Debug and failed operations use Error. A nil logger creates a no-op observer.

func NewSanitizedLoggingObserver added in v0.2.0 

func NewSanitizedLoggingObserver(logger Logger, sanitizer FailureSanitizer) Observer

NewSanitizedLoggingObserver reports observations with an explicit failure sanitizer. A nil sanitizer uses DefaultFailureSanitizer.

type Operation 

type Operation string

Operation identifies an observable messaging boundary. 

const (
	// OperationDeliver covers outbound route delivery.
	OperationDeliver Operation = "deliver"
	// OperationHandle covers local or durable handler execution.
	OperationHandle Operation = "handle"
	// OperationQuery covers a complete local request/reply call.
	OperationQuery Operation = "query"
	// OperationService covers managed service completion.
	OperationService Operation = "service"
	// OperationBrokerAck covers broker-confirmed acknowledgement of a consumed message.
	OperationBrokerAck Operation = "broker_ack"
	// OperationOffsetCommit covers transactional Kafka offset finalization.
	OperationOffsetCommit Operation = "offset_commit"
	// OperationRetryHandoff covers durable retry scheduling or broker hand-off.
	OperationRetryHandoff Operation = "retry_handoff"
	// OperationDLQHandoff covers durable terminal hand-off to a dead-letter destination.
	OperationDLQHandoff Operation = "dlq_handoff"
)

type Option 

type Option func(*Builder)

Option configures a Builder.

func WithClock 

func WithClock(clock func() time.Time) Option

WithClock overrides the UTC wall clock used for messages and receipts.

func WithContextPropagator 

func WithContextPropagator(propagator ContextPropagator) Option

WithContextPropagator sets distributed-context injection for outgoing metadata. The default propagator is a no-op.

func WithIDGenerator 

func WithIDGenerator(generator IDGenerator) Option

WithIDGenerator overrides UUIDv7 generation, primarily for deterministic tests.

func WithLogger 

func WithLogger(logger Logger) Option

WithLogger sets the core structured logger. The default logger is a no-op.

func WithObserver 

func WithObserver(observer Observer) Option

WithObserver appends a lifecycle observer.

func WithPanicReporter added in v0.2.0 

func WithPanicReporter(reporter PanicReporter) Option

WithPanicReporter enables explicit handling of sensitive recovered-panic values and stacks. Without it, only a sanitized HandlerPanicError is emitted.

func WithRuntimeShutdownTimeout added in v0.2.0 

func WithRuntimeShutdownTimeout(timeout time.Duration) Option

WithRuntimeShutdownTimeout sets the internal bound used when Run owns service shutdown after cancellation or an unexpected service return.

func WithSource 

func WithSource(source string) Option

WithSource sets the required stable producer identity.

type Outgoing 

type Outgoing[T any] struct {
	Payload  T
	Metadata OutgoingMetadata
}

Outgoing combines a typed payload with optional explicit metadata.

type OutgoingMetadata 

type OutgoingMetadata struct {
	ID            MessageID
	Subject       string
	Time          time.Time
	CorrelationID MessageID
	CausationID   MessageID
	Key           string
	Headers       map[string]string
	NotBefore     time.Time
	ExpiresAt     time.Time
}

OutgoingMetadata customizes metadata generated for a new outgoing message. Source, kind, name, schema version, content type, and schema come from the builder and descriptor and cannot be overridden per call.

type PanicReport added in v0.2.0 

type PanicReport struct {
	HandlerID string
	Value     any
	Stack     []byte
}

PanicReport contains sensitive diagnostics for an explicitly configured PanicReporter. Value and Stack must not be written to untrusted logs or DLQ records without host-side redaction.

type PanicReporter added in v0.2.0 

type PanicReporter interface {
	ReportPanic(ctx context.Context, handlerID string, recovered any, stack []byte)
}

PanicReporter receives sensitive recovered-panic diagnostics. The default is to drop these details and return only HandlerPanicError to application code.

type PanicReporterFunc added in v0.2.0 

type PanicReporterFunc func(context.Context, PanicReport)

PanicReporterFunc adapts a function to PanicReporter.

func (PanicReporterFunc) ReportPanic added in v0.2.0 

func (f PanicReporterFunc) ReportPanic(
	ctx context.Context,
	handlerID string,
	recovered any,
	stack []byte,
)

ReportPanic implements PanicReporter.

type PayloadHandler 

type PayloadHandler[T any] func(context.Context, T) error

PayloadHandler processes only the payload and ignores message metadata.

type Publisher 

type Publisher[T any] interface {
	Publish(ctx context.Context, payload T) (Receipt, error)
	PublishMessage(ctx context.Context, outgoing Outgoing[T]) (Receipt, error)
}

Publisher is the ordinary generic DI interface for a bound event descriptor.

func BindPublisher 

func BindPublisher[T any](messenger *Messenger, descriptor Event[T]) Publisher[T]

BindPublisher returns a narrow DI facade bound to one event descriptor.

type Querier 

type Querier[Q, R any] interface {
	Query(ctx context.Context, payload Q) (R, error)
}

Querier is the ordinary generic DI interface for a bound query descriptor.

func BindQuerier 

func BindQuerier[Q, R any](messenger *Messenger, descriptor Query[Q, R]) Querier[Q, R]

BindQuerier returns a narrow DI facade bound to one query descriptor.

type Query 

type Query[Q, R any] struct {
	// contains filtered or unexported fields
}

Query is an immutable typed local query descriptor. Its codec describes the request Q only; R is a compile-time result identity and is never serialized.

func MustQuery 

func MustQuery[Q, R any](name string, schemaVersion int, codec Codec[Q]) Query[Q, R]

MustQuery constructs a typed local query descriptor and panics when its declaration is invalid.

func NewQuery 

func NewQuery[Q, R any](name string, schemaVersion int, codec Codec[Q]) (Query[Q, R], error)

NewQuery constructs a typed local query descriptor.

func (Query[Q, R]) Info 

func (d Query[Q, R]) Info() DescriptorInfo

Info returns a copy of the query request identity.

func (Query[Q, R]) WithSchema 

func (d Query[Q, R]) WithSchema(schema string) Query[Q, R]

WithSchema returns a query descriptor with an explicit request schema URI.

type QueryHandler 

type QueryHandler[Q, R any] func(context.Context, Message[Q]) (R, error)

QueryHandler processes one typed query and returns its typed result.

func ChainQueryHandler 

func ChainQueryHandler[Q, R any](
	handler QueryHandler[Q, R],
	middlewares ...QueryHandlerMiddleware[Q, R],
) QueryHandler[Q, R]

ChainQueryHandler applies typed query middleware with the first item outermost. It returns nil when the chain is invalid.

func HandleQueryPayload 

func HandleQueryPayload[Q, R any](handler QueryPayloadHandler[Q, R]) QueryHandler[Q, R]

HandleQueryPayload adapts a payload-only query handler to QueryHandler.

type QueryHandlerMiddleware 

type QueryHandlerMiddleware[Q, R any] func(QueryHandler[Q, R]) QueryHandler[Q, R]

QueryHandlerMiddleware wraps a typed query handler and may return a cached or synthetic result without invoking the wrapped handler.

type QueryPayloadHandler 

type QueryPayloadHandler[Q, R any] func(context.Context, Q) (R, error)

QueryPayloadHandler processes only a query payload and returns its typed result.

type Receipt 

type Receipt struct {
	MessageID MessageID    `json:"messageId"`
	Route     string       `json:"route"`
	State     ReceiptState `json:"state"`
	At        time.Time    `json:"at"`
}

Receipt describes the guarantee reached by one primary route.

type ReceiptState 

type ReceiptState string

ReceiptState describes what a successful route call has guaranteed. 

const (
	// ReceiptCompleted means local synchronous handlers completed.
	ReceiptCompleted ReceiptState = "completed"
	// ReceiptAccepted means bounded in-process async admission succeeded.
	ReceiptAccepted ReceiptState = "accepted"
	// ReceiptStaged means an outbox write succeeded in the current transaction.
	ReceiptStaged ReceiptState = "staged"
	// ReceiptBrokerConfirmed means the broker confirmed persistence.
	ReceiptBrokerConfirmed ReceiptState = "broker_confirmed"
	// ReceiptNoop means a local event had no subscribers.
	ReceiptNoop ReceiptState = "noop"
)

type Route 

type Route interface {
	Name() string
	Deliver(ctx context.Context, delivery Delivery) (Receipt, error)
}

Route is one static primary delivery route.

type Runtime 

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

Runtime supervises the services declared on one immutable Builder. It never restarts a service automatically.

func (*Runtime) BeginDrain 

func (r *Runtime) BeginDrain()

BeginDrain marks the runtime unready and asks every service to stop admission.

func (*Runtime) DeepHealth added in v0.2.0 

func (r *Runtime) DeepHealth(ctx context.Context) error

DeepHealth performs explicit, potentially expensive service health and topology checks. It is intended for diagnostics or a low-frequency probe.

func (*Runtime) Liveness added in v0.2.0 

func (r *Runtime) Liveness(ctx context.Context) error

Liveness checks that Runtime has not terminated and invokes optional service liveness checks without requiring readiness or topology access.

func (*Runtime) Readiness 

func (r *Runtime) Readiness(ctx context.Context) error

Readiness checks runtime admission state and each service's lightweight readiness contract. Expensive topology validation belongs in DeepHealth.

func (*Runtime) Run 

func (r *Runtime) Run(ctx context.Context) error

Run starts every service and blocks until cancellation, draining, or the first unexpected service return.

func (*Runtime) Shutdown 

func (r *Runtime) Shutdown(ctx context.Context) error

Shutdown drains all services and waits for a concurrent Run to finish. Coordinate it outside handlers executing on this Runtime.

type Sender 

type Sender[T any] interface {
	Send(ctx context.Context, payload T) (Receipt, error)
	SendMessage(ctx context.Context, outgoing Outgoing[T]) (Receipt, error)
}

Sender is the ordinary generic DI interface for a bound command descriptor.

func BindSender 

func BindSender[T any](messenger *Messenger, descriptor Command[T]) Sender[T]

BindSender returns a narrow DI facade bound to one command descriptor.

type Service 

type Service interface {
	Run(ctx context.Context) error
	Readiness(ctx context.Context) error
	BeginDrain()
	Shutdown(ctx context.Context) error
}

Service is a host-supervised managed consumer or worker lifecycle. BeginDrain must be non-blocking, stop admission, and cause a running Run call to return without requiring Shutdown to be invoked first. Shutdown waits for or force-cancels remaining work within its context.

type ServiceProvider 

type ServiceProvider interface {
	ManagedService() (serviceID string, service Service)
}

ServiceProvider lets a route contribute one managed service to Builder.

Source Files

View all Source files

Directories

Path Synopsis
adapters
inbox module
kafka module
nats module
outbox module
tools
gomessengerctl module

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.