lib-streaming

lib-streaming is Lerian Studio's event publication AND consumption library for CloudEvents-framed domain events. It gives Go services a catalog-driven Emitter API with multi-transport routing across Kafka, SQS, RabbitMQ, and EventBridge, per-target tenant-aware circuit breakers, route-aware outbox fallback, per-route DLQ, and observability contracts designed for financial infrastructure — plus a hardened at-least-once group Consumer that subscribes by producing application, dispatches by event key, verifies ce-source, and owns commit, retry, seek-back, DLQ, tenant propagation, and rebalance safety.

It does not replace github.com/LerianStudio/lib-commons/v7/commons/rabbitmq, which remains the internal command-queue primitive. The two are orthogonal: lib-streaming carries past-tense business facts, lib-commons carries internal commands.

Module	github.com/LerianStudio/lib-streaming/v4
Go	1.26.3
License	Elastic License 2.0. See LICENSE

Why lib-streaming?

Contract-First Events	Services publish catalog-keyed events; resource, event type, schema version, content type, and default delivery policy live in immutable event definitions
CloudEvents Native	Every message uses CloudEvents 1.0 binary-mode metadata with raw JSON payloads
Multi-Transport	One Emit dispatches to Kafka, SQS, RabbitMQ, and EventBridge routes in deterministic route-table order; per-target tenant-aware circuit breakers, all-or-error semantics for required routes, best-effort for optional
Broker Resilience	Per-target circuit breakers prevent hot-looping on broker failures and route through outbox fallback when configured. With lib-commons TenantAwareManager, non-system events get isolated (tenant, target) breakers so one tenant's outage does not reject neighbors. A background recovery goroutine per Producer pokes registered breakers so stuck-OPEN mirrors can transition within CBTimeout + 5s + one probe round-trip after broker recovery, even for emit-only services with no other CB traffic
Reliable Replay	Route-aware outbox envelopes persist target name, transport, destination, policy, and event payload for deterministic replay
Forensic DLQs	Routable failures land in the route's DLQ with structured headers for source topic, error class, retry count, failure time, and producer identity
Testing Ergonomics	streamingtest.MockEmitter gives services a concurrency-safe test double with assertion helpers and wait support

Publication Pipeline

lib-streaming implements a complete producer-side publication pipeline:

1. Define — Build a Catalog of EventDefinition records at service bootstrap.
2. Configure — Load STREAMING_* settings, broker credentials, and optional delivery-policy overrides.
3. Construct — Use streaming.NewBuilder() to wire targets, routes, and lifecycle dependencies. Disabled-feature-flag environments use streaming.NewNoopEmitter().
4. Emit — Service code sends catalog-keyed EmitRequest values with tenant, subject, payload, and optional policy override.
5. Resolve — Definition defaults, config overrides, and call overrides resolve the final direct/outbox/DLQ behavior per route.
6. Publish or Persist — Deterministic per-route dispatch: direct publish to each target, or persist a route-aware OutboxEnvelope when policy or circuit state requires it.
7. Observe — Emit metrics, spans, structured logs, circuit state, health state, and DLQ/outbox routing counters.

Architecture

lib-streaming is a Go library with a small public facade at the repository root and implementation details isolated under internal/.

Package Boundaries

Package	Purpose
streaming	Public facade: NewBuilder, NewNoopEmitter, config aliases, producer wrapper, event/catalog/policy aliases, options, manifest helpers, error sentinels
internal/contract	Core event model, catalog validation, delivery policies, routes, health states, sentinels, caller-error taxonomy
internal/config	STREAMING_* environment parsing, defaults, validation
internal/manifest	Publisher descriptors, manifest DTOs, stdlib HTTP introspection handler
internal/cloudevents	Kafka CloudEvents binary-mode header codec
internal/emitter	No-op emitter implementation
internal/producer	Producer runtime: multi-target dispatch, per-target circuit breakers, publish/outbox/DLQ paths, metrics, tracing, runtime assertions
internal/transport	TransportAdapter port and shared message/header types
internal/transport/kafka	franz-go-backed Kafka adapter
internal/transport/sqs	SQS adapter built on a caller-supplied SQSPublisherClient
internal/transport/rabbitmq	RabbitMQ events adapter built on a caller-supplied RabbitMQPublisher
internal/transport/eventbridge	EventBridge adapter built on a caller-supplied PutEvents client
streamingtest	Public testing helpers, mock emitter, assertion helpers

Technical Highlights

• Three-method emitter contract: Emit(ctx, EmitRequest) error, Close() error, and Healthy(ctx) error.
• Immutable catalog: deterministic definition ordering, duplicate-key rejection, and duplicate contract-tuple rejection.
• Policy precedence: definition default → config override → call override.
• One topic per producing application: lerian.streaming.<source>. The topic carries no resource type, no event type, and no schema version — every business fact a service emits rides it. Its service-to-service commands ride lerian.streaming.<source>.commands, and DLQ is lerian.streaming.<source>.dlq. Kafka ACLs scope a command-emitting application to exactly those three WRITES and everyone else to two — consumers included, since a consumer quarantines into its own DLQ — plus READ on the topics of the applications it consumes.
• Commands are a separate queue with opposite unmatched semantics: a definition marked Class: streaming.ClassCommand publishes to <source>.commands, and a consumer subscribed via Commands(...) quarantines an event key it has no handler for instead of skipping it. The class is not on the wire — the queue is the class.
• Strict ce-source: a single dot-free lowercase segment (^[a-z0-9][a-z0-9_-]*$), at most 223 bytes so the derived .commands name fits Kafka's 249-byte limit, rejected — never rewritten — at config, Builder, and preflight time.
• ce-type carries the app: studio.lerian.<source>.<resource>.<event>, so two services' same-named events never collide.
• Tenant-aware partitioning: tenant ID by default; system events use system:<eventType> and require explicit opt-in. The key buys per-tenant FIFO on the direct-emit path and per-tenant partition affinity — not strict order — for outbox-relayed events, because the lib-commons relay retries per row with no per-aggregate serialization. A consumer needing strict per-aggregate order must reconcile on its own sequence field.
• Caller-error taxonomy: IsCallerError(err) distinguishes correctable validation/auth/serialization failures from broker/runtime faults.
• Lifecycle integration: *Producer implements commons.App for Launcher-owned startup and shutdown.
• Metric cardinality discipline: no tenant_id metric labels; tenant identity belongs on spans.

Getting Started

Prerequisites

• Go 1.26.3+
• Redpanda or Kafka for direct publish and integration tests
• Docker for testcontainers-backed integration tests

1. Install

go get github.com/LerianStudio/lib-streaming/v4@latest

2. Bootstrap a Producer

cfg, warnings, err := streaming.LoadConfig()
if err != nil {
    return err
}
for _, warning := range warnings {
    logger.Log(ctx, log.LevelWarn, warning)
}

runtime.InitPanicMetrics(metricsRecorder)
assert.InitAssertionMetrics(metricsRecorder)
// Scrubs panic value strings and truncates stack traces before they hit
// log fields, span events, and ErrorReporter payloads — guards against
// PII leakage from arbitrary panic arguments and OTel attribute bloat.
// appCfg is your service-level config; streaming.Config intentionally does
// not own runtime environment classification.
runtime.SetProductionMode(appCfg.Env == "production")

catalog, err := streaming.NewCatalog(streaming.EventDefinition{
    Key:          "transaction.created",
    ResourceType: "transaction",
    EventType:    "created",
})
if err != nil {
    return err
}

// Kill switch FIRST. LoadConfig skips validation when disabled, so a disabled
// deployment legitimately carries an empty source — deriving a topic from it
// before this check would fail the one path that is supposed to be inert.
if !cfg.Enabled {
    // Inject streaming.NewNoopEmitter() into services; skip launcher.Add on
    // the no-op path.
    return nil
}
if len(cfg.Brokers) == 0 {
    return errors.New("streaming enabled but STREAMING_BROKERS is empty")
}

// ONE topic per producing application: lerian.streaming.<source>.
// AppTopic VALIDATES the source and returns (string, error): provisioning
// creates this name, an ACL grants it, and a route publishes to it, so a
// malformed source fails here rather than reaching a real broker.
appTopic, err := streaming.AppTopic(cfg.CloudEventsSource)
if err != nil {
    return err
}

emitter, err := streaming.NewBuilder().
    Source(cfg.CloudEventsSource).
    Catalog(catalog).
    Routes(streaming.RouteDefinition{
        // No DefinitionKey: a catch-all route serves the whole catalog.
        // Under one topic per app there is nothing to fan out per event.
        Key:         "primary.kafka",
        Target:      "primary",
        Destination: streaming.KafkaTopic(appTopic),
        Requirement: streaming.RouteRequired,
    }).
    Target(streaming.TargetConfig{
        Name:    "primary",
        Kind:    streaming.TransportKafkaLike,
        Brokers: cfg.Brokers,
    }).
    Logger(logger).
    MetricsRecorder(metricsRecorder).
    Tracer(tracer).
    CircuitBreakerManager(cbManager).
    OutboxRepository(outboxRepo).
    Build(ctx)
if err != nil {
    return err
}

// Cast to the lifecycle-bearing wrapper for outbox replay registration and
// launcher integration. The Builder returns the Emitter interface so tests
// see the same surface as production code.
producer := emitter.(*streaming.Producer)
if err := producer.RegisterOutboxRelay(outboxRegistry); err != nil {
    return err
}
if err := launcher.Add("streaming", producer); err != nil {
    return err
}

3. Emit from Services

Inject the interface, not the concrete producer:

func NewTransactionService(emitter streaming.Emitter) *TransactionService {
    return &TransactionService{emitter: emitter}
}

err := emitter.Emit(ctx, streaming.EmitRequest{
    DefinitionKey: "transaction.created",
    TenantID:      "t-abc",
    Subject:       "tx-123",
    Payload:       payloadBytes,
})

4. Unit-Test with streamingtest

mock := streamingtest.NewMockEmitter()
svc := NewTransactionService(mock)

if err := svc.Create(ctx, input); err != nil {
    t.Fatal(err)
}

streamingtest.AssertEventEmitted(t, mock, "transaction.created")
streamingtest.AssertTenantID(t, mock, "t-abc")

Configuration

All environment variables use the STREAMING_ prefix. The canonical reference lives in .env.reference, and LoadConfig() returns migration warnings alongside the parsed config.

When STREAMING_ENABLED=false, callers should use streaming.NewNoopEmitter() instead of constructing a Builder. Do not treat an empty broker list as an intentional production disablement when streaming is required. Fail startup and fix the deployment secret or config instead. Multi-transport wiring (multiple Kafka clusters, SQS / RabbitMQ / EventBridge dispatch) is programmatic — non-Kafka destinations such as SQS queue URLs, RabbitMQ exchanges, and EventBridge bus names are typically already plumbed through the consuming service's own configuration.

Topics are created for you

Declaring an event is the whole job. At construction, each runtime creates the topics it owns — no new method, no new argument.

The rule: a runtime ensures every topic it writes under its own source namespace.

Construction	Creates
Builder.Build	lerian.streaming.<source> — the app's fact topic, on every Kafka target
Builder.Build	lerian.streaming.<source>.dlq — where it route-DLQs a publish failure
NewConsumer().Build	lerian.streaming.<source>.dlq — the consumer's own DLQ (it is that topic's producer)
NewConsumer().Commands(<own source>).Build	lerian.streaming.<source>.commands — commands taken under its own namespace

The DLQ is the same name family on both sides, so an app that both produces and consumes ensures it twice — fine and expected, since TOPIC_ALREADY_EXISTS is silent success.

Nobody provisions a name outside their own namespace. Subscribing with Apps(...), taking another app's Commands(...), or using the raw Topics(...) escape hatch creates nothing — those names belong to their owners, which create them on their own Build. Kafka only; the SQS, RabbitMQ, and EventBridge adapters are untouched.

One boundary is held deliberately: a commands queue is never provisioned by its emitter. A commands queue lives in the commanding app's namespace — a producer with a Class: ClassCommand definition writes lerian.streaming.<its own source>.commands, and the addressee subscribes with Commands("<commander>"). Even though that name is in the emitter's own namespace, it is not created there. The resulting ordering expectation is intentional: a command emitted before its addressee's first deploy fails visibly, rather than succeeding into a queue nobody reads. Same posture as subscriptions — a command accepted into an unread queue is undelivered money-path work behind a green dashboard, which is the exact failure the commands queue exists to prevent.

This exists because nothing else created them: Lerian brokers run auto_create_topics_enabled=false (correct hardening) and the streaming-hub reconciler is read-only by design. The resulting failures were quiet at boot and loud too late — a producer initialized cleanly and its first publish returned UNKNOWN_TOPIC_OR_PARTITION, and a consumer subscribed to a nonexistent topic is indistinguishable from one on an idle topic: franz-go surfaces no topic-specific fetch error, so the poll loop records a clean cycle, Healthy passes, and the service consumes nothing indefinitely.

Failure posture: WARN, never refuse to start. A creation that fails logs a WARN naming the topic and the missing ACL, and startup continues. The reason is that an authorization failure is the normal state in a hardened environment where topics come from IaC and the runtime credential deliberately has no CreateTopics — refusing to boot there would break exactly the deployments that are configured correctly.

What happens next is not symmetric, and it decides how you alert:

• Publishing to a missing topic fails loudly — the publish returns ClassTopicNotFound to the caller, already fail-closed. This is not the outbox absorbing it: outbox fallback covers circuit-open only, and only when a caller wired one. A producer with no outbox simply gets the error back.
• Consuming from a missing topic fails silently — indistinguishable from an idle subscription, so no error, no metric, and Healthy still passes. The WARN is the only signal. Alert on it.

Remediation: grant the service's principal CREATE on the named topic (or on the CLUSTER), or pre-provision through IaC and set STREAMING_TOPIC_AUTO_PROVISION=false.

Variable	Type	Default	Purpose
STREAMING_TOPIC_AUTO_PROVISION	bool	true	Opt out in environments that pre-provision through IaC
STREAMING_TOPIC_PARTITIONS	int	-1	-1 uses the broker's num.partitions
STREAMING_TOPIC_REPLICATION_FACTOR	int	-1	-1 uses the broker's default.replication.factor

The admin call rides the runtime's own franz-go client, so the broker dial (brokers, TLS, SASL) is the validated one already in use — there is no second connection configuration to drift. The round-trip is bounded at 10s (not configurable) so a broker outage cannot hang startup.

Consuming a Stream

Under one topic per producing application, one subscription delivers that application's entire fact stream. Selection moves from the broker to the consumer: name the producers you consume, register a handler per event, and the library resolves the topics, verifies each event's ce-source, and dispatches by event key.

Commands are the exception, and they get their own subscription — see Commands: the same condition, the opposite verdict.

c, err := streaming.NewConsumer().
    Brokers(cfg.Brokers...).
    Group("my-service").
    Source(cfg.CloudEventsSource).                    // this app's own identity; names its DLQ
    Apps("lender", "matcher").                        // -> lerian.streaming.{lender,matcher}
    OnFrom("lender", "loan.disbursed", onLenderLoan). // "<resourceType>.<eventType>"
    OnFrom("matcher", "loan.disbursed", onMatcherLoan).
    OnFrom("lender", "loan.settled", onLoanSettled).
    RetryBudget(3).
    Classifier(isTransient).
    Build(ctx)
if err != nil {
    return err
}

go func() { // runtime.SafeGo in production
    // Run returns nil on clean shutdown (ctx cancel or Close); any error is
    // an unexpected exit and must reach a log line or the lifecycle
    // supervisor, or processing stops silently.
    if err := c.Run(ctx); err != nil {
        logger.Log(ctx, log.LevelError, "streaming consumer exited", log.Err(err))
    }
}()
defer c.Close()

Those two loan.disbursed registrations are the point. Lender's and matcher's are different facts with different payloads — v3 put the producing app into ce-type precisely so they stop colliding, and OnFrom is how a consumer spends that. A single-producer consumer keeps the terse form: Apps("lender").On("loan.disbursed", h) binds to the sole app. A bare On under several apps fails the build rather than binding to whichever record arrives first.

Concern	Behaviour
Identity	Source(...) (or STREAMING_CLOUDEVENTS_SOURCE) is required. One service, one ce-source: the producer publishes under it and the consumer quarantines into lerian.streaming.<source>.dlq.
Subscription	Apps("lender") → lerian.streaming.lender (facts). Commands("lender") → lerian.streaming.lender.commands. Topics(...) remains the raw escape hatch. All three compose, and naming one app in both Apps and Commands is legal — two subscriptions, one allowlist entry.
Dispatch key	(producing app, "<resourceType>.<eventType>"). OnFrom(app, key, fn) names the app; On(key, fn) binds to the sole app and fails the build when there is more than one. Underscores travel verbatim.
Unmatched events	On a fact stream: ignored (skipped and committed) by default; UnmatchedPolicy(streaming.UnmatchedError) quarantines them instead. On a commands queue: always quarantined with cause kind unhandled_key. Not configurable — UnmatchedPolicy governs fact streams only.
Source verification	A record whose ce-source is not one of the named Apps/Commands is quarantined with ErrUnexpectedSource before any handler runs — in both handler modes. ExpectSources(...) — or STREAMING_CONSUMER_EXPECT_SOURCES — replaces that derived allowlist, must cover every named app, and is the only way to resolve the named-app + Topics refusal from the environment.
Whole-stream handler	Handler(h) receives every record for consumers that select themselves, and gets the same source verification. It still rejects the genuinely dispatch-only knobs: On/OnFrom (ErrHandlerAndDispatchBothSet), UnmatchedPolicy (ErrHandlerAndUnmatchedPolicyBothSet), and Commands (ErrHandlerAndCommandsBothSet — with no handler registry there is nothing to ask whether a command key is handled).
Readiness	Healthy(ctx) fails with ErrConsumerPartitionHalted once a partition has been head-of-line blocked across three consecutive poll cycles. Polling cleanly is not the same as making progress.

UnmatchedIgnore is the default on fact streams because it is the only safe one there: a consumer subscribed to a producer's fact stream receives every fact that producer emits and will legitimately care about a handful. Erroring on the rest would fail-closed the producer's entire sibling stream into the DLQ. Choose UnmatchedError only when this consumer genuinely owns every fact on the stream and an unknown key means the producer's catalog drifted ahead of it.

Commands: the same condition, the opposite verdict

A command is work one named service asks another to do. It rides a separate queue — lerian.streaming.<producer>.commands — and a consumer subscribes to it with Commands(...):

c, err := streaming.NewConsumer().
    Brokers(cfg.Brokers...).
    Group("br-consignado-gw").
    Source(cfg.CloudEventsSource).
    Apps("lender").                                    // lender's FACTS: unmatched keys ignored
    Commands("lender").                                // lender's COMMANDS: unmatched keys quarantined
    OnFrom("lender", "margin.reserve", onMarginReserve).
    Build(ctx)
if err != nil {
    return err
}

Same consumer, same unregistered key, opposite outcomes — decided by the topic the record arrived on. That asymmetry is the only reason the queue exists. A fact you did not register for is noise on someone else's firehose. A command you did not register for is undelivered work addressed to you: under fact semantics, a producer shipping a new command key before this consumer deploys its handler would lose every one of those commands, forever, with green dashboards on both sides. So it quarantines to this consumer's own DLQ with cause kind unhandled_key, and that is deliberately not configurable.

On the producing side nothing changes but one field:

streaming.EventDefinition{
    Key:          "margin.reserve",
    ResourceType: "margin",
    EventType:    "reserve",
    Class:        streaming.ClassCommand, // -> lerian.streaming.<source>.commands
}

The route table is untouched — a catch-all route to AppTopic(source) still serves the whole catalog, and the producer redirects command-class definitions at dispatch. An explicit KafkaTopic(...) you pointed somewhere on purpose is never rewritten — with one exception: you may not point one at lerian.streaming.<source>.commands itself. Build refuses that route with ErrInvalidRouteDefinition, because a command reaching the queue by route skips the DLQ pin and derives the .commands.dlq that deliberately does not exist, and a fact reaching it by route lands on the strict queue where consumers quarantine keys they were entitled to ignore. The class on the definition is the only door. The wire record is byte-identical either way: no ce-* header carries the class, because the queue is the class — which makes it a subscription-time, ACL-visible fact rather than a runtime string every consumer has to trust.

Tenant filtering remains the handler's responsibility: event.TenantID comes from the validated ce-tenantid header, never from the payload, and a handler that skips the tenant check has a cross-tenant leak.

Handler errors must not carry PII. A terminal handler error travels to the DLQ as x-lerian-dlq-error-message, with only broker credentials stripped and the value capped at 4 KiB — nothing else is redacted. A CPF, account number, or name interpolated into a returned error is published onto the DLQ topic verbatim, readable by anything with DLQ read access for the topic's whole retention window. Return an opaque identifier and look the record up out of band.

Quarantine lands in the consumer's own DLQ

A consumer publishes poison to lerian.streaming.<consumer-app>.dlq — its own topic, not the producer's. The rule that falls out of it is worth stating plainly for whoever writes the ACLs:

Every application WRITES only its own names — its topic, its .commands queue if it commands anyone, and its .dlq. Three names for a command-emitting app, two for everyone else, whether it produces, consumes, or both. It READS the topics of the applications it consumes — their fact topics when it watches their facts, their .commands when they command it.

Consuming therefore never widens an application's write grant, and a filling DLQ names the team that owns the fix rather than the team whose events happened to be poison.

There is deliberately no .commands.dlq. A consumer quarantines into its own .dlq, and a producer route-DLQs a failed command publish into its own; both names already exist and are already granted.

Splitting commands out also gives a read grant back. A rail consumer that only takes a producer's commands — a gateway acting on lender's instructions — needs READ on lerian.streaming.lender.commands and nothing on lerian.streaming.lender. Under the collapsed topic it had to read the producer's whole fact stream to receive one command; least-privilege is partially restored.

The Streaming Hub subscribes fact topics only, never .commands. The Hub fans business events out to tenant webhooks and external buses; a service-to-service command is neither public nor idempotent under external redelivery, so it is not in the Hub's grant. Because the DLQ topic no longer implies where a record came from, every consumer quarantine carries x-lerian-dlq-source-topic, -source-partition, and -source-offset — the route back to the original.

Provision .dlq topics with size headroom. A DLQ record is strictly larger than the record it quarantines (same payload, same headers, plus the forensic set), so max.message.bytes on a .dlq topic must be at or above its source topic's. Without headroom a near-cap record cannot be quarantined at all, and on the consume side that is fail-closed: the partition is held back and the record redelivers forever. The library covers what it can from its side — a size-driven DLQ publish is retried once with the payload omitted and marked x-lerian-dlq-payload-omitted: true (plus x-lerian-dlq-payload-bytes), and the payload stays recoverable from the source topic via the origin coordinates — but headroom is the actual fix.

Reading a DLQ

A quarantined record is durable but invisible until something drains it. The forensic x-lerian-dlq-* headers (six on every entry, three more on a consumer quarantine, and two payload markers only when the payload was dropped) do not survive the CloudEvents codec, so a plain Handler cannot see any of them — DiscardHandler is the seam that can:

type desk struct{}

func (desk) HandleDiscard(ctx context.Context, r streaming.DiscardRecord) error {
    // r.Event.TenantID — the tenant that owned the poison record
    // r.CauseKind      — why it died: codec / handler / source_mismatch / unhandled_key
    // r.SourceTopic, r.SourcePartition, r.SourceOffset — the route back to it
    // r.PayloadOmitted — whether r.Payload is genuinely absent or the real bytes
    return nil
}

dlqTopic, err := streaming.AppDLQTopic("lender") // the queue it drains
if err != nil {
    return err
}

c, err := streaming.NewConsumer().
    Brokers(cfg.Brokers...).
    Group("lender-dlq-desk").
    Source("lender-dlq-desk"). // NOT "lender" — see below
    Topics(dlqTopic).
    DiscardHandler(desk{}).
    Build(ctx)
if err != nil {
    return err
}
defer func() { _ = c.Close() }()

return c.Run(ctx)

That snippet is compiled, not transcribed: it is Example_readingADLQ in example_test.go, so it fails the build if the wiring ever drifts.

Give the reader its own ce-source. Source(...) names where a consumer quarantines — lerian.streaming.<source>.dlq — so a reader built with Source("lender") draining lerian.streaming.lender.dlq would quarantine into the topic it is emptying: republish, redeliver, quarantine, forever, while reporting healthy and growing the topic without bound. Both strings are known at construction, so Build refuses a DiscardHandler consumer subscribed to its own quarantine topic (ErrSubscribedToOwnQuarantineTopic). A distinct source gives the reader its own quarantine topic, which the consumer provisions and owns; draining another application's .dlq was never the constraint. Budget one new Kafka topic and one new app identity in the ACL model per DLQ reader.

That refusal is what makes the remaining rule safe: the error your handler returns is classified like any other handler error, and it now lands somewhere you do not read.

A plain Handler subscribed to its own quarantine topic is warned about and still built. That shape is one earlier versions accept and that drains clean in practice — valid envelopes, an allowlisted ce-source, the handler returns nil — so the loop there is latent rather than active, and refusing it would turn a library upgrade into a startup outage. The log line names the topic; the fix is the same distinct ce-source.

DiscardHandler is mutually exclusive with Handler, On/OnFrom, Commands, UnmatchedPolicy, Apps and ExpectSources, enforced at Build in either order. Apps subscribes to fact topics, never a .dlq, and a reader never verifies ce-source, so an allowlist would be ignored. Name the queue with Topics(...).

The origin triple is the stable natural key for deduping a redelivered or replayed quarantine; ce-id is not, because the replay path can quarantine the same event twice. It fails closed as a unit: if any of its three headers is present but unreadable, all three are discarded and HeaderError says so, because topic/0/42 is a plausible-looking coordinate pointing at the wrong record. An absent coordinate is not a failure — a producer-side quarantine legitimately has an origin topic and no partition or offset.

Two library-side terminal verdicts are lifted on this path, because a .dlq topic's normal content would otherwise be treated as poison. A codec fault delivers the record with a zero envelope and the reason in DiscardRecord.EnvelopeError — an unparseable envelope is what a codec entry IS — and ce-source verification is skipped, since a quarantine copy carries the original producer's ce-source, never the reader's.

For tooling that holds the headers itself, streaming.ParseDiscardRecord(headers, payload) does the same decode standalone and never fails; the header keys and cause-kind values are exported as streaming.DLQHeader* / streaming.DLQCause*, and streaming.TruncatedErrorMessageBytes tells a cut error message from a whole one.

Multi-Transport Routing

A single Emit can dispatch to N routes. Route attempts run in deterministic route-table order inside the Emit call. Per-target circuit breakers isolate target failures; when the configured manager supports lib-commons TenantAwareManager, non-system events use tenant-scoped breakers for each target so tenant A's outage does not reject tenant B. Required routes drive the aggregate Emit outcome; optional routes are best-effort.

Concepts

• Target — a named transport runtime (kafka-primary, sqs-shadow, ...). One adapter per target.
• Route — maps one catalog EventDefinition to one (target, destination) pair. A definition can have many routes; each one is evaluated independently per Emit.
• Requirement — RouteRequired (must succeed for the Emit to succeed) or RouteOptional (best-effort; metric outcomes, trace events, and DLQ when configured).
• Outbox — when a route's target circuit breaker is OPEN and an outbox writer is wired, lib-streaming writes a route-aware envelope and replays it through the target's adapter without going through Emit (no breaker re-check).
• DLQ per route — each route can declare a DLQ destination. DLQ topic naming, headers, and routing rules apply per route.

Wiring multiple targets

// AppTopic returns (string, error), so hoist it out of the literal.
appTopic, err := streaming.AppTopic("midaz-ledger")
if err != nil {
    return err
}

emitter, err := streaming.NewBuilder().
    Source("midaz-ledger").
    Catalog(catalog).
    Routes(
        // Catch-all: the whole catalog rides the app topic on the primary target.
        streaming.RouteDefinition{
            Key:         "kafka_primary.all",
            Target:      "kafka-primary",
            Destination: streaming.KafkaTopic(appTopic),
            Requirement: streaming.RouteRequired,
        },
        // Definition-scoped: shadow ONE event to SQS. Resolution is
        // ADDITIVE per target — this route names a DIFFERENT target than
        // the catch-all, so transaction.created goes to BOTH the app topic
        // and the SQS queue. A definition-scoped route only replaces the
        // catch-all when it names the SAME target.
        streaming.RouteDefinition{
            Key:           "transaction_created.sqs.shadow",
            DefinitionKey: "transaction.created",
            Target:        "sqs-shadow",
            Destination:   streaming.SQSQueueURL("https://sqs.us-east-1.amazonaws.com/123/q"),
            Requirement:   streaming.RouteOptional,
        },
    ).
    Target(streaming.TargetConfig{
        Name:    "kafka-primary",
        Kind:    streaming.TransportKafkaLike,
        Brokers: cfg.Brokers,
    }).
    SQSTarget("sqs-shadow", sqsClient, "https://sqs.us-east-1.amazonaws.com/123/q").
    Logger(logger).
    MetricsRecorder(metricsRecorder).
    Tracer(tracer).
    OutboxRepository(outboxRepo).
    Build(ctx)
if err != nil {
    return err
}

All-or-error semantics

For a single Emit dispatched across N routes:

• Every RouteRequired route must succeed (or fall back to outbox) for Emit to return nil.
• A required-route failure aggregates into *MultiEmitError. IsCallerError returns true only when every required failure is itself caller-correctable.
• RouteOptional failures never propagate. They still produce per-route streaming_emitted_total outcomes such as failed, circuit_open, outbox_failed, or dlq, and they add a route.optional_failed span event with route.key, route.target, route.state, and error.type attributes. If the route declares a DLQ, routable failures are sent to that destination.

Production guidance: Do not use optional routes for audit, compliance, or customer-visible obligations unless you also alert on optional-route degradation. Derive that alert from the route.optional_failed span event or from route-specific logs/traces. There is no outcome="optional_failed" metric label in the current code.

Built-in non-Kafka adapters

The library does NOT bundle AWS or AMQP SDKs. The built-in adapters in internal/transport/{sqs,rabbitmq,eventbridge} define small interfaces that callers fulfill with their own SDK clients:

Transport	Caller interface	Public helpers
SQS	streaming.SQSPublisherClient (SendMessage(ctx, queueURL, body, attributes) error)	streaming.SQSAdapter, Builder.SQSTarget
RabbitMQ	streaming.RabbitMQPublisher (Publish(ctx, exchange, routingKey, contentType, body, headers) error)	streaming.RabbitMQAdapter, Builder.RabbitMQTarget
EventBridge	streaming.EventBridgePutEventsClient (PutEvents(ctx, entries) error)	streaming.EventBridgeAdapter, Builder.EventBridgeTarget

Production SQS, RabbitMQ, and EventBridge clients must implement Ping(ctx) error; Adapter.Healthy fails closed when the caller-supplied client has no health probe. The health capabilities are exported as streaming.SQSPingClient, streaming.RabbitMQPingClient, and streaming.EventBridgePingClient so wrappers can assert the contract at compile time without changing the backwards-compatible publish interfaces.

The SQS and EventBridge adapters reject provider wire messages larger than 256 KiB with ErrPayloadTooLarge before issuing any network call. SQS accounting includes the body plus String message attributes. Because SQS allows only 10 message attributes, the SQS adapter forwards up to 10 deterministic attributes directly and packs overflow metadata into x-lerian-streaming-extra-headers as JSON instead of rejecting valid CloudEvents or DLQ metadata; traceparent and tracestate are kept as top-level SQS attributes whenever present so standard trace extraction can see them. EventBridge accounting measures the rendered PutEvents entry contribution. The EventBridge adapter renders a canonical CloudEvents Detail JSON shape (specversion, id, source, type, subject, time, datacontenttype, dataschema, tenantid, traceparent, tracestate, data) so downstream rules can match without hard-coding ce-* headers. EventBridge clients may also implement EventBridgePutEventsResultClient with PutEventsWithResult(ctx, entries) (PutEventsResult, error) to expose per-entry PutEvents failures without changing the backwards-compatible EventBridgePutEventsClient interface.

Operational note: SQS routes resolve DNS at construction

NewRouteDefinition and NewRouteTable validate every SQS Destination against AWS SQS endpoint host patterns, then call ssrf.ResolveAndValidate to reject DNS answers in blocked ranges (loopback, link-local, RFC1918, cloud-metadata ranges). This prevents caller-owned SDK wrappers from signing requests to arbitrary public hosts and fails startup when an SQS hostname resolves unsafely. Operational consequence: service bootstrap waits on resolver validation for each unique SQS queue URL in a route table. DNS lookup failures remain fail-closed and cause NewRouteTable (and therefore Builder.Build) to return an error after the internal bounded retry budget is exhausted. The current internal budget is 3 attempts, a 500 ms timeout per attempt, and 25 ms backoff between attempts. lib-streaming does not expose a public retry or timeout option. Deploy with a healthy DNS resolver in the pod/container network namespace.

RabbitMQ is events-only

The lib-streaming RabbitMQ adapter is for business events aimed at third-party / SaaS subscribers. Internal command queues remain on github.com/LerianStudio/lib-commons/v7/commons/rabbitmq. The two are orthogonal — neither replaces the other.

Custom transports

For SDK shapes lib-streaming does not cover (Kinesis, Pub/Sub, NATS, ...), declare streaming.TransportCustom on the route Destination and register the adapter factory via Builder.RegisterTransport(streaming.TransportCustom, factory). The factory receives a per-target TransportAdapterOptions with the target name and any caller-typed Extra payload from Builder.TargetExtra.

Manifest

streaming.BuildManifest(descriptor, catalog, routes) renders a JSON-serializable view of the catalog plus the active route table for ops and contract introspection. streaming.NewStreamingHandler(descriptor, catalog, opts ...HandlerOption) returns a stdlib http.Handler that serves the manifest. Pass streaming.WithManifestRoutes(routeTable) to advertise the active route table in the manifest's routes section; with no options the handler serves a catalog-only manifest, byte-identical to the prior two-argument form.

doc, err := streaming.BuildManifest(descriptor, catalog, routeTable)
// doc.Routes is populated and JSON-stable when len(routeTable) > 0;
// pass an empty RouteTable for a catalog-only document.

The manifest version is exposed as streaming.ManifestVersion (currently 1.0.0 — the platform is greenfield, so this document is the first shipped manifest contract and consumers discriminate by structure, never by this string). Routes are deterministically ordered (definition key, then route key) so the JSON document is byte-stable across builds.

The document advertises the application's topic and dlqTopic, and — only when the catalog holds at least one ClassCommand definition — its commandsTopic. Its presence is the manifest's answer to "does this application command anyone?", so a fact-only producer never points provisioning or ACL tooling at a topic it will not write. There is no commandsDlqTopic. Every event carries a class of "fact" or "command", always present so a reader can tell "emits only facts" from "predates the field".

(*Producer).Descriptor(base PublisherDescriptor) returns the validated descriptor with the per-process ProducerID populated and Source replaced by the producer's own ce-source (the manifest topic derives from it, so the producer is its only authority — a caller-supplied value is discarded). Use it to feed BuildManifest or to stamp identity into custom DLQ metadata without re-validating the descriptor surface manually.

SECURITY: the manifest exposes event taxonomy, schema versions, service metadata, and producer IDs. Callers MUST wrap the handler in their app's auth middleware before mounting it publicly. The library does not enforce authentication.

Example with an explicit auth wrapper:

manifestHandler, err := streaming.NewStreamingHandler(
    descriptor,
    catalog,
    streaming.WithManifestRoutes(routeTable),
)
if err != nil {
    return err
}

// authenticate is owned by the host service. It must reject unauthenticated
// requests before the manifest handler sees them.
mux.Handle("/streaming", authenticate(manifestHandler))

Every PR that mounts or changes /streaming exposure must include an explicit review note that names the auth middleware, the intended audience, and whether the route is reachable from public networks.

Operational Guidance

Circuit-breaker tuning

The Builder exposes three setters that control per-target circuit-breaker behavior. Zero values fall back to lib-commons HTTP presets:

Setter	Default	Effect
CBFailureRatio(float64)	0.5	Failure ratio that trips OPEN once CBMinRequests is reached
CBMinRequests(int)	10	Minimum request count in the rolling window before the breaker can trip
CBTimeout(time.Duration)	30s	OPEN-state dwell time AND the source for the CB recovery loop's tick interval

The CB recovery goroutine runs at clamped(cbTimeout/4, [500ms, 5s]). With a tenant-aware manager it calls GetState on every no-tenant target breaker and GetStateForTenant for every Producer-owned (tenant, target) breaker key recorded during Emit; otherwise it calls GetState on every target. This makes gobreaker's lazy OPEN→HALF-OPEN transition fire deterministically without scanning unrelated manager inventory. Maximum recovery latency after broker recovery is bounded at CBTimeout + 5s + one probe round-trip. Shorter CBTimeout values tighten that envelope at the cost of more aggressive trip behavior — choose by failure-budget, not by recovery time alone.

Healthy(ctx) checks target adapter readiness and the internal CB recovery goroutine liveness. It returns Healthy when every target ping succeeds and the recovery loop is alive/fresh, Degraded when at least one target ping fails but another target or outbox fallback remains viable, and Down when all targets fail with no outbox or after the producer closes. If the recovery goroutine exits after a recovered panic or stops reporting fresh liveness, Healthy(ctx) reports a Degraded health error while otherwise healthy targets can still publish.

Dashboard-visible recovery liveness comes from streaming_cb_recovery_liveness plus panic/assertion signals. Alert on these conditions:

• streaming_cb_recovery_liveness == 0 while the producer is expected to be running. This means the recovery loop is dead or stale and Healthy(ctx) will no longer report fully healthy.
• panic_recovered_total{component="streaming",goroutine_name="cb_recovery_loop"} > 0 after runtime.InitPanicMetrics(...) is initialized. This means the recovery loop recovered a panic and exited.
• assertion_failed_total{component="streaming",operation="cb_recovery.start"} > 0 after assert.InitAssertionMetrics(...) is initialized. This means a construction invariant disabled the recovery loop at startup.
• Persistent streaming_emitted_total{outcome="circuit_open"} or streaming_outbox_routed_total{reason="circuit_open"} after the broker is healthy again. This catches ineffective recovery even when there is no panic metric.

Per-target observability

streaming_circuit_state is a single-dimension gauge tracking the primary target's no-tenant compatibility breaker only (the first registered target). Tenant-scoped breaker state is intentionally not projected onto this gauge because the metric has no tenant dimension; use lib-commons circuit-breaker metrics/logs, which expose bounded tenant_hash, for per-tenant dashboards. Per-target circuit-state changes are emitted through traces and logs to keep metric label cardinality bounded:

• Span events on the active emit span carry target.name and target.cb_state attributes. Trace-based metrics derived from these attributes give per-target dashboards without exploding the gauge series.
• Structured log fields: every CB-related log line includes target=<name>. Log-based metric extraction (Loki / CloudWatch metric filters / GCP log metrics) is the supported path for per-target alerting.
• Rationale: tenant_id is already off the metric label set for the same cardinality reason. A per-target gauge series is reasonable for small fixed N but creates a foot-gun for services that scale targets dynamically. Operators wanting bounded per-target gauges can derive them from spans or logs under their own cardinality budget.

Alerting recipe:

1. Keep streaming_circuit_state == 2 as the primary-target compatibility alert.
2. Add log- or trace-derived alerts grouped by target for every non-primary target.
3. Add lib-commons tenant-aware circuit-breaker alerts grouped by bounded tenant_hash when the service uses TenantAwareManager.

Do not assume a green streaming_circuit_state means every route is healthy. It only proves that the primary no-tenant breaker is not OPEN.

Per-route metric volume

A single Emit dispatched across N routes increments streaming_emitted_total N times — one per route attempt — even though the caller issued a single Emit call. Dashboards computing "logical Emits per second" should aggregate per-Emit attempts via trace spans, not by summing per-route counters. The topic label distinguishes destinations across routes; the outcome label uses the current closed set from code: produced, outboxed, circuit_open, caller_error, dlq, failed, and outbox_failed.

Capacity-plan accordingly: counter volume scales with route count, not Emit count.

Outbox replay: classify caller errors as non-retryable

Wire streaming.IsCallerError into the lib-commons outbox dispatcher:

dispatcher, err := outbox.NewDispatcher(
    outboxRepo,
    outboxRegistry,
    logger,
    tracer,
    outbox.WithRetryClassifier(outbox.RetryClassifierFunc(streaming.IsCallerError)),
)
if err != nil {
    return err
}

Without it, a row that can never succeed burns its whole retry budget and its whole backoff window before anyone hears about it, and the relay's throughput goes with it. IsCallerError returns true exactly for the synchronous, caller-correctable faults — validation, serialization, auth, and every caller-correctable sentinel — so the dispatcher moves those rows straight to INVALID and keeps retrying only the ones a retry could fix.

This no longer applies to leftover version-1 rows, and it is worth being explicit about why, because earlier versions of this document said the opposite. OutboxEnvelopeVersion is 2 (bumped in v3), but the relay now READS version-1 rows — the ones lib-streaming v2 wrote — and re-derives their destination onto the current application topic. They drain; they are not rejected. See Upgrading from lib-streaming v2.

What the classifier still buys you is every OTHER permanently-unpublishable row: an unknown envelope version (neither 1 nor 2 — corruption, or a row from a future major), a malformed envelope (one that does not decode, or decodes but fails validation), or an event that fails replay preflight — an empty or oversized payload, a payload that fails json.Valid under a JSON content type (a declared non-JSON DataContentType ships opaque and is not scanned), a missing ResourceType or EventType, a missing or invalid ce-source for a version-2 row, a system event this producer does not allow, or an unsafe header field (a control character or an over-length CloudEvents attribute). That holds for version-1 rows too, except the ce-source check: v2 enforced every one of the other checks, so a version-1 row failing one of them is no more publishable than a version-2 row. Those can never succeed, so they should land in INVALID immediately — alertable, countable, replayable after a rewrite — rather than cycling through retries for hours.

One row shape deliberately does NOT go to INVALID even with the classifier wired: a version-1 row whose ce-source cannot be re-derived under the current rules. It fails with ErrLegacyOutboxRowUnroutable, which is intentionally not a caller error, so it keeps its retry budget while streaming_outbox_relay_rejected_total{reason="legacy_unroutable"} and an ERROR log naming the row id call for an operator. Alert on it:

increase(streaming_outbox_relay_rejected_total{reason="legacy_unroutable"}[15m]) > 0

The counter's other reason, version_unsupported, is a different condition with a different cause — an envelope this build cannot read, bound for INVALID — so alert on it separately rather than folding the two together.

DLQ alerting

streaming_dlq_publish_failed_total increments when the DLQ publish itself fails. The original required-route failure may still return to the caller, but the forensic copy was not preserved. Alert on any increase:

increase(streaming_dlq_publish_failed_total[5m]) > 0

Non-Kafka routes need explicit DLQ destinations. Kafka-like routes derive <destination topic>.dlq — for the default route that is lerian.streaming.<source>.dlq, one DLQ per application. A failed command publish quarantines there too, not to a .commands.dlq: that fourth name does not exist and nothing provisions it. SQS, RabbitMQ, EventBridge, and custom routes skip DLQ delivery unless RouteDefinition.DLQ is set. The DLQ destination kind must match the source route destination kind because lib-streaming publishes the DLQ message through the same target adapter.

For production routes where quarantine is mandatory, make DLQ part of the route review checklist. Optional routes that are business-critical should also declare a DLQ and have separate optional-route failure alerts, because optional route failures do not fail the caller's Emit.

Consumer alerting

Three consumer signals are worth a page. All are counters on the streaming_consumer_ prefix.

# A partition is head-of-line blocked. reason="dlq_publish_failed" means a
# record cannot be quarantined at all (check .dlq topic size headroom and ACLs)
# and NOTHING behind it will ever commit; reason="sustained_transient" means a
# downstream is down. Healthy() also fails after three consecutive cycles, so
# this should coincide with the pod leaving the load balancer.
increase(streaming_consumer_partition_halted_total[5m]) > 0

# The quarantine write itself failed. Every increase here is a record that
# could not be set aside — on the consume side that is fail-closed, so it is a
# wedge in progress, not a lost forensic copy.
increase(streaming_consumer_dlq_publish_failed_total[5m]) > 0

# FACT records arriving with no handler registered. A steady non-zero rate on a
# key you expected to own means a typo'd registration or a producer catalog that
# moved ahead of this consumer — it commits and processes nothing, silently.
# event_key="other" means the 64-label cap was reached; the key names are in
# the logs (search "unmatched event key seen past the metric label cap").
sum by (event_key) (increase(streaming_consumer_unmatched_total[15m])) > 0

# An unhandled COMMAND. This one is not "processing nothing" — it is
# undelivered work already set aside, so page rather than ticket: either this
# consumer is behind its producer's catalog, or a handler registration is
# wrong. The quarantined records carry the origin coordinates needed to replay
# them once the handler ships.
increase(streaming_consumer_dlq_total{cause_kind="unhandled_key"}[15m]) > 0

Project Structure

lib-streaming/
├── *.go                    # Public root facade package: streaming
├── internal/               # Private implementation packages
│   ├── cloudevents/        # Kafka CloudEvents binary-mode headers
│   ├── config/             # STREAMING_* config parsing and validation
│   ├── contract/           # Event, catalog, policy, route, health, sentinels
│   ├── emitter/            # No-op emitter
│   ├── manifest/           # Publisher manifest and HTTP handler
│   ├── producer/           # Producer runtime, multi-target dispatch
│   └── transport/          # Transport port + Kafka/SQS/RabbitMQ/EventBridge adapters
├── streamingtest/          # Public mock emitter and test assertions
├── docs/                   # Design notes, plans, and project rules
├── reports/                # Generated local reports and coverage artifacts
├── scripts/                # Makefile support scripts
└── .github/                # CI and release workflows

Development

Common Commands

Command	Purpose
make test	Run unit tests
make test-unit	Run unit tests excluding integration packages
make test-integration	Run testcontainers-backed integration tests
make test-chaos	Run Toxiproxy-backed chaos tests (CHAOS=1 set automatically)
make test-all	Run unit, integration, and chaos suites
make coverage-unit	Generate unit coverage
make coverage-integration	Generate integration coverage
make coverage	Generate combined coverage reports
make lint	Run lint checks
make lint-fix	Run auto-fixing lint checks
make format	Apply gofmt
make tidy	Clean Go module dependencies
make check-tests	Verify repository test expectations
make vet	Run go vet
make sec	Run gosec security checks
make ci	Run local fix + verify pipeline
make goreleaser	Build a local release snapshot

Testing

lib-streaming uses build tags as the authoritative test-tier discriminator:

Tag	Scope	External dependencies
//go:build unit	Unit tests	None
//go:build integration	Integration tests	Docker/testcontainers and Redpanda/Kafka containers
//go:build chaos	Fault-injection tests	Toxiproxy/testcontainers

The default make test target is unit-focused. Integration and chaos coverage is intentionally explicit because it requires local Docker availability and may be slower than pure unit verification.

Test-infrastructure dependencies such as testcontainers, Toxiproxy, kfake, and MongoDB drivers support repository tests. They are not runtime transport dependencies for consuming services, although Go's module graph can still surface them in dependency scanners because Go does not provide a separate dev-dependency section.

Tenant identity is caller-supplied. lib-streaming validates the tenant ID shape, but it does not compare the value against an authenticated request context. A tenant-context validator hook is a deferred service-side hardening option, not current library behavior.

API Documentation

The package-level API documentation is generated from Go doc comments:

go doc github.com/LerianStudio/lib-streaming

Key public API areas:

• Builder — NewBuilder, Source, Catalog, Routes, Target, TargetExtra, RegisterTransport, CBFailureRatio, CBMinRequests, CBTimeout, CloseTimeout, Logger, MetricsRecorder, Tracer, CircuitBreakerManager, OutboxRepository, OutboxWriter, TLSConfig, SASL, AllowPlaintextSASL, AllowSystemEvents, PartitionKey, SQSTarget, RabbitMQTarget, EventBridgeTarget, Build.
• Routes & destinations — TargetConfig, RouteDefinition, RouteTable, Destination, TransportKind, RouteRequirement, KafkaTopic, SQSQueueURL, RabbitMQRoute, EventBridgeBus.
• Topic naming — AppTopic, AppDLQTopic, AppCommandsTopic, ValidateSource, TopicPrefix, DLQTopicSuffix, CommandsTopicSuffix, MaxKafkaTopicNameBytes.
• Consumer dispatch — NewConsumer().Source(...), .Apps(...), .Commands(...), .OnFrom(app, "<resourceType>.<eventType>", handler), .On(...) (single-app shorthand), .UnmatchedPolicy(...), .ExpectSources(...), HandlerFunc, UnmatchedIgnore, UnmatchedError, ErrUnhandledEvent, ErrUnexpectedSource, ErrBareOnWithMultipleApps, ErrUnknownDispatchApp, ErrConsumerMissingSource, ErrConsumerPartitionHalted, ErrHandlerAndCommandsBothSet.
• Transport port — TransportAdapter, TransportMessage, TransportHeader, TransportAdapterOptions, TransportAdapterFactory, PartitionKeyFunc, plus the built-in client interfaces (SQSPublisherClient, RabbitMQPublisher, EventBridgePutEventsClient).
• Emitters — Emitter, Producer (including Descriptor, RegisterOutboxRelay, Run, RunContext, CloseContext), NoopEmitter, and streamingtest.MockEmitter.
• Catalogs — Catalog, EventDefinition, NewCatalog, NewEventDefinition, EventClass, ClassFact, ClassCommand, and duplicate-contract validation.
• Requests — EmitRequest, NewEmitRequest, payload rules, tenant/subject metadata.
• Delivery Policies — DefaultDeliveryPolicy, ResolveDeliveryPolicy, DirectMode, OutboxMode, DLQMode, DeliveryPolicy, DeliveryPolicyOverride.
• Outbox — OutboxEnvelope (with Validate / ValidateShape), OutboxWriter, WithOutboxTx, StreamingOutboxEventType, transactional writer support, relay registration.
• Manifest — BuildManifest, NewStreamingHandler, HandlerOption, WithManifestRoutes, NewPublisherDescriptor, ManifestDocument, ManifestEvent, ManifestRoute, ManifestVersion.
• Errors — sentinels, EmitError, MultiEmitError, RouteError, error classes, HealthError, and IsCallerError.

Contributing

1. Fork the repository.
2. Create a branch from develop such as feat/my-feature or fix/my-bug.
3. Follow conventions: run make format && make tidy && make lint && make test before pushing.
4. Submit a PR to develop with a conventional commit title and a clear description of behavior, tests, and compatibility impact.

PR Requirements

• Public API changes must update README, Go doc comments, and CHANGELOG.md when behavior or compatibility changes.
• New or changed STREAMING_* environment variables must update .env.reference.
• New exported identifiers require accurate Go doc comments.
• New production behavior requires unit tests; broker/outbox/DLQ behavior should include integration coverage when feasible.
• Do not add tenant IDs or other high-cardinality values as metric labels.
• Manifest handler exposure must include an auth review note. Name the middleware and state whether /streaming is public, internal-only, or disabled.
• Optional routes that carry business-critical data must document their alerting path and DLQ posture.

Code Standards

• Preserve the documented public API surface unless a breaking change is explicit and documented.
• Prefer explicit error returns over panic paths.
• Keep nil handling safe for public options, optional dependencies, and lifecycle methods.
• Keep implementation details under internal/; expose only intentional facade types from the root package.
• Reuse lib-commons primitives for observability, UUIDv7 generation, outbox integration, and runtime/assertion instrumentation.

For detailed conventions, see docs/PROJECT_RULES.md.

Community & Support

• Discord: Join our community for discussions, support, and updates.
• GitHub Issues: Bug reports and feature requests.
• GitHub Discussions: Community Q&A.
• Twitter/X: @LerianStudio.

License

Elastic License 2.0. See LICENSE.

About Lerian

Lerian Studio builds open source infrastructure for financial services. Learn more at lerian.studio.