Ring:Using Lib Streaming

ring:using-lib-streaming

When to use

Sweep mode:

• "Sweep this service for lib-streaming adoption opportunities"
• "Find every DIY Kafka / SQS / RabbitMQ publisher that should use lib-streaming"
• "Audit our event publishing layer against lib-streaming"
• "We have raw franz-go / sarama / watermill — what should move to lib-streaming?"
• "Identify ad-hoc manifest or per-service circuit breakers around event publishing"

Reference mode:

• Need to understand what lib-streaming provides at the public facade
• Looking for the right Builder setter, Destination helper, or sentinel
• Setting up a new service that emits CloudEvents-framed domain events
• Wiring multi-transport fan-out (Kafka primary + SQS shadow, etc.)
• Writing unit tests that depend on streamingtest.MockEmitter
• Need the canonical CloudEvents binary-mode Kafka header bytes for an interop layer
• Need to map a runtime *EmitError / *MultiEmitError to an operator response

Skip when

• Working on non-Go services (lib-streaming is Go-only)
• Working on frontend code
• Service is consumer-only with no outbound business event surface — there is nothing for lib-streaming to instrument
• Service publishes only internal command queue messages — those stay on github.com/LerianStudio/lib-commons/v5/commons/rabbitmq; lib-streaming is for past-tense business events to external subscribers

Related

• ring:instrumenting-streaming-events — end-to-end 13-gate orchestration that implements lib-streaming in a target service. Use that after ring:mapping-streaming-events has produced a validated docs/streaming/instrumentation-map.json. This skill is the adoption/reference counterpart — it does not own the implementation cycle.
• ring:mapping-streaming-events — PM-side identification of eventable points; produces the catalog and instrumentation map this skill's REFERENCE MODE consumes.
• ring:using-lib-commons — lib-streaming depends on lib-commons for circuit breaker, outbox repository, App lifecycle, runtime panic instrumentation, and assertions. The CB / outbox / runtime / assert API surface lives there.
• ring:using-outbox — OutboxWriter, TransactionalOutboxWriter, WithOutboxTx, and route-aware envelope replay live in the outbox skill. This skill points at the boundary but does NOT duplicate the writer / dispatcher API.
• ring:using-lib-observability — log.Logger, metrics.MetricsFactory, and trace.Tracer are owned there. Builder setters consume those types; this skill links rather than re-documents.

Distinction: adoption/reference vs end-to-end implementation

ring:using-lib-streaming is the adoption and reference skill. It answers two questions:

1. Where in this codebase are we doing event publication the wrong way? (Sweep Mode)
2. What is the right lib-streaming API for the thing I am building right now? (Reference Mode)

ring:instrumenting-streaming-events is the end-to-end implementation orchestrator — it consumes a validated instrumentation map, walks a 13-gate cycle (catalog, producer bootstrap, emit instrumentation, outbox wiring, HTTP manifest, NoopEmitter fallback, integration + chaos tests, 9 default reviewers plus triggered specialists), and never lets the caller skip TDD. The two are complementary, not overlapping:

• Sweep finds the work. Outputs are file:line replacement candidates and a task backlog.
• Implementation does the work. Consumes the catalog + map, drives gates, owns the agent dispatch.

If the user asks for a sweep, use this skill. If the user already has the map and wants emission wired into a service, hand off to ring:instrumenting-streaming-events.

Mode Selection

SWEEP MODE

Orchestrate a 4-phase sweep. Each phase has a hard gate — do not proceed until the current phase produces its artifact.

Copy code
Phase 1: Version Reconnaissance    -> version-report.json
Phase 2: CHANGELOG Delta Analysis  -> delta-report.json
Phase 3: Multi-Angle DIY Sweep     -> 8 x libstreaming-sweep-{N}-{angle}.json
Phase 4: Consolidated Report       -> libstreaming-sweep-report.md + tasks.json

Phase 1: Version Reconnaissance

1. Read go.mod — extract pinned version of github.com/LerianStudio/lib-streaming (if absent, flag as not-adopted).
2. WebFetch https://api.github.com/repos/LerianStudio/lib-streaming/releases/latest — extract tag_name.
3. Classify drift: not-adopted / up-to-date / minor-drift / moderate-drift / pre-release-only.
4. Cross-check lib-commons pin: lib-streaming requires github.com/LerianStudio/lib-commons/v5 v5.2.0-beta.11 (or newer compatible). If lib-commons is v4.x or absent, add a major upgrade advisory flag — adoption is blocked until lib-commons is on v5.
5. Emit version-report.json: {pinned_version, latest_version, drift_classification, lib_commons_version, lib_commons_compatible, blocked_by_lib_commons, module_path}.

Phase 2: CHANGELOG Delta Analysis

1. WebFetch https://raw.githubusercontent.com/LerianStudio/lib-streaming/main/CHANGELOG.md.
2. Extract entries between pinned_version (exclusive) and latest_version (inclusive). When not-adopted, summarise the current public facade instead of a delta.
3. Classify each entry: new-transport / new-api / breaking-change / security-fix / performance / bugfix.
4. Emit delta-report.json with classified entries plus a producer_only_reminder flag (true if any entry touches the consumer surface — lib-streaming has no consumer surface, so any such entry is wrong CHANGELOG content, not a migration item).

Phase 3: Multi-Angle DIY Sweep

Dispatch all 8 explorer angles in 2 batches (4+4). Wait for each batch before next.

⛔ STOP-CHECK BEFORE DISPATCH (each batch)

Before emitting any Task call in a batch, count the explorers you intend to launch in this turn.

• Count MUST equal 4 (each batch is 4 explorers).
• If count < 4 → STOP. Do not partial-dispatch. Reconcile against the batch row above and try again.
• No substitutions, no omissions within a batch.

⛔ MUST NOT trickle-dispatch within a batch

All 4 explorers in a batch leave in the SAME TURN, before reading any explorer output.

Forbidden sequences:

• Dispatch explorer 1 → read result → dispatch explorer 2
• Dispatch a subset of the batch → wait → dispatch the rest
• Dispatch follow-up explorers conditioned on partial output
• Loop sequentially over the batch's angle list

If you find yourself about to dispatch an explorer in a turn AFTER any explorer in the SAME batch has already returned a result → STOP. You violated parallel dispatch. Report the violation and mark the batch INCOMPLETE rather than completing the trickle. (Sequential batch ordering is intentional; trickle within a batch is not.)

Self-verify after dispatch

After each batch's dispatch turn, verify all 4 Task calls were emitted in that single turn. If fewer than 4 went out, the batch did NOT execute correctly. Mark INCOMPLETE and surface the dispatch failure — do NOT silently continue with a partial batch.

Parallel dispatch — atomic batch (within this batch)

Emit all 4 Task calls for THIS BATCH in a SINGLE TURN, as one atomic batch. (Batches themselves remain sequential — do not dispatch batch N+1 until batch N has fully returned.)

If your runtime exposes a multi_tool_use.parallel wrapper, use it to dispatch the complete batch in one wrapped invocation. This is the canonical fan-out mechanism on OpenAI-style tool envelopes and on certain Anthropic SDK consumers — naming it explicitly activates parallel emission on runtimes where trickle-dispatch is the default behavior.

If your runtime emits parallel tool_use blocks natively (Claude Code with Claude models), multi_tool_use.parallel may not be needed — but naming it is harmless and serves as an enforcement anchor.

The STOP-CHECK, anti-trickle, and self-verify guards above remain binding regardless of which mechanism your runtime uses.

Per-explorer dispatch (subagent_type: ring:codebase-explorer):

Copy code
## Target
<absolute path to target repo root>

## Your Angle
<angle number + name>

## Severity Calibration / DIY Patterns / Replacement / Migration Complexity
<verbatim from the angle table below>

## Output
Write findings to: /tmp/libstreaming-sweep-{N}-{angle-slug}.json
Schema: {
  angle_number, angle_name, severity, migration_complexity,
  findings: [{file, line, diy_pattern, replacement, evidence_snippet, notes}],
  summary, requires_lib_commons_v5
}
If no findings: write file with empty findings array and summary "No DIY patterns detected for this angle".

Angle Catalogue

Severity calibration rules

• HIGH = correctness or operability risk if left in place. A DIY franz-go path bypasses circuit breakers, outbox fallback, tenant-aware partitioning, DLQ routing, and manifest exposure; a sarama path additionally pins the service to a maintained-by-others client.
• MEDIUM = behaviour is roughly correct but the service ships a non-uniform operator experience (custom manifest URL, hand-rolled CB metrics, raw CE headers that drift from the canonical codec).
• LOW = strictly internal command-queue usage that should NOT migrate; record as out_of_scope with a one-line rationale.

Out-of-scope reminders for the explorer

• Internal command queues on lib-commons/v5/commons/rabbitmq are NOT in scope. lib-streaming and the commons RabbitMQ primitive are orthogonal; neither deprecates the other.
• Pure consumers (cloudevents-sdk-go + franz-go consumer groups) are NOT in scope — lib-streaming is producer-only.
• Outbox writer / WithOutboxTx / RegisterOutboxRelay findings belong in the ring:using-outbox sweep, not here. Flag them as cross-skill: using-outbox rather than including them in the angle output.

Phase 4: Consolidated Report

Dispatch synthesizer (subagent_type: ring:codebase-explorer):

Copy code
Read /tmp/version-report.json, /tmp/delta-report.json, /tmp/libstreaming-sweep-*.json (8 files).
Emit:
1. /tmp/libstreaming-sweep-report.md  — aggregate findings by severity, group by file
2. /tmp/libstreaming-sweep-tasks.json — one task per DIY pattern cluster (same file/package = one task)

MUST NOT invent findings. MUST NOT omit explorer findings. MUST NOT reclassify severity without justification.
MUST NOT merge cross-skill outbox findings into this report — surface them as a separate "Cross-skill handoff" section pointing to `ring:using-outbox`.

Surface report path + task count to the user; if adoption is feasible (lib-commons v5 already pinned, target service has a real business-event surface), offer handoff to ring:instrumenting-streaming-events. If the service has no validated instrumentation map yet, the handoff is to ring:mapping-streaming-events first.

REFERENCE MODE

The public facade lives at the root package github.com/LerianStudio/lib-streaming. Everything else is internal/ — do not import it. This section catalogs the facade organised by the producer lifecycle: Bootstrap -> Build -> Emit -> Test -> Manifest.

Quick Navigation

Bootstrap

LoadConfig() reads STREAMING_* environment variables and returns (Config, warnings []string, error). Print warnings — they carry forward-compat migration notes that the validator did not classify as failures.

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

Multi-transport (multiple Kafka clusters, SQS / RabbitMQ / EventBridge fan-out) is wired programmatically through Builder, not via STREAMING_* — non-Kafka destinations such as SQS queue URLs and EventBridge bus names typically live in the consuming service's own config layer.

Catalog construction

goCopy code
catalog, err := streaming.NewCatalog(
    streaming.EventDefinition{
        Key:           "transaction.created",
        ResourceType:  "transaction",
        EventType:     "created",
        SchemaVersion: "1.0.0",
        Description:   "A transaction was durably committed.",
    },
    streaming.EventDefinition{
        Key:           "transaction.posted",
        ResourceType:  "transaction",
        EventType:     "posted",
        SchemaVersion: "1.0.0",
    },
)
if err != nil { return err }

Catalog is immutable. Duplicate keys, malformed schema versions, and missing required fields fail at construction. Topics are derived as lerian.streaming.<resource>.<event> with a .v<major> suffix for SchemaVersion >= 2.

Panic + assertion metric wiring

lib-streaming uses lib-commons/v5/commons/runtime and lib-observability/assert internally. Without these init calls the metric counters stay at zero (logs + span events still fire). Wire after telemetry is initialised and before Builder construction:

goCopy code
runtime.InitPanicMetrics(metricsFactory)
assert.InitAssertionMetrics(metricsFactory)
runtime.SetProductionMode(cfg.Env == "production")

SetProductionMode(true) scrubs panic value strings and truncates stack traces before they hit log fields, span events, and ErrorReporter payloads. Without it, arbitrary panic arguments flow verbatim into telemetry — a PII risk in financial workloads.

Disabled feature-flag fallback

goCopy code
if !cfg.Enabled || len(cfg.Brokers) == 0 {
    emitter := streaming.NewNoopEmitter()
    // inject emitter into services; skip launcher.Add for the no-op path
    return inject(emitter)
}

NewNoopEmitter() returns a concurrency-safe Emitter whose three methods (Emit, Close, Healthy) are unconditional no-ops. Service constructors depend on the Emitter interface, so the rest of the code is unchanged.

Build

streaming.NewBuilder() is the only constructor. All chainable setters return *Builder and are nil-receiver safe. Call Build(ctx) last — it returns (Emitter, error) and is the only step that performs I/O (target adapter construction, SSRF resolution for SQS routes).

Required setters

Observability setters

Resilience setters

Security setters

Outbox setters (boundary only)

The outbox semantics, transactional writer, envelope schema, and replay path are documented in ring:using-outbox — that skill is the canonical source. This skill only flags the Builder boundary.

Partition key override

goCopy code
b.PartitionKey(func(e streaming.Event) string {
    return e.TenantID + ":" + e.ResourceID
})

Nil fn is a no-op — the producer falls back to Event.PartitionKey(). Default tenant-scoped partitioning is appropriate for >95% of services; override only when you have a measured cardinality or ordering problem.

Multi-transport target helpers

The library does not bundle AWS or AMQP SDKs. Each helper takes a caller-supplied client and registers both the target and its transport adapter factory in one call.

Each interface optionally implements Ping(ctx) error — if present, Adapter.Healthy delegates to it. SQS and EventBridge adapters reject payloads larger than 256 KiB with ErrPayloadTooLarge before issuing any network call.

Operational note (SQS): NewRouteDefinition and NewRouteTable validate every SQS Destination via ssrf.ResolveAndValidate, which performs a synchronous DNS lookup at construction. A DNS outage at boot fails Builder.Build. Deploy with a healthy resolver in the pod/container network namespace.

Custom transports

For SDK shapes not covered (Kinesis, Pub/Sub, NATS, ...), declare streaming.TransportCustom on the route Destination and register the adapter via Builder.RegisterTransport(streaming.TransportCustom, factory). The factory receives TransportAdapterOptions{Name, Brokers, Logger, Extra} — Extra is the caller-typed payload from Builder.TargetExtra(name, value).

Destination helpers

Route requirements

• streaming.RouteRequired — must succeed (or fall back to outbox) for Emit to return nil.
• streaming.RouteOptional — best-effort. Failures never propagate; surfaced via metrics and (when the route declares DLQ) the DLQ destination.

Emit

The Emitter interface (alias for contract.Emitter) is three methods:

goCopy code
type Emitter interface {
    Emit(ctx context.Context, request EmitRequest) error
    Close() error
    Healthy(ctx context.Context) error
}

Service code MUST depend on this interface. The Builder returns it from Build(ctx). Three implementations satisfy it:

EmitRequest shape

goCopy code
type EmitRequest struct {
    DefinitionKey  string                 // catalog key, e.g. "transaction.created"
    TenantID       string                 // becomes ce-tenantid; required for non-system events
    Subject        string                 // becomes ce-subject; aggregate ID is conventional
    EventID        string                 // optional; auto-generated UUIDv7 when empty
    Timestamp      time.Time              // optional; defaults to time.Now() at emit
    Payload        json.RawMessage        // body. Must be valid JSON; max 1 MiB
    PolicyOverride DeliveryPolicyOverride // optional per-call override
}

Emit example:

goCopy code
err := emitter.Emit(ctx, streaming.EmitRequest{
    DefinitionKey: "transaction.created",
    TenantID:      tmcore.GetTenantIDContext(ctx),
    Subject:       tx.ID,
    Payload:       payloadBytes,
})

Service handlers MUST pull TenantID from the request context (e.g. tmcore.GetTenantIDContext), never hardcode.

*Producer lifecycle

*Producer implements lib-commons/v5/commons.App. The consuming service's main.go wires it via launcher.Add / launcher.RunApp; the Launcher owns startup and shutdown. Service methods receive the Emitter interface and MUST NOT call Close — the Launcher does on shutdown.

Methods reachable only via type-assertion to *streaming.Producer:

Close is idempotent: first call drains every registered target adapter under STREAMING_CLOSE_TIMEOUT_S; subsequent calls return nil. After Close, Emit returns ErrEmitterClosed synchronously before any I/O.

Delivery policy precedence

Copy code
definition default -> config override (STREAMING_EVENT_POLICIES) -> per-call PolicyOverride

Resolution lives in streaming.ResolveDeliveryPolicy(definition, configOverride, callOverride). Three delivery modes compose:

Three idiomatic postures match ring:mapping-streaming-events taxonomy:

Multi-target fan-out semantics

A single Emit fanned out across N routes:

• Every RouteRequired route must succeed (or fall back to outbox) for Emit to return nil.
• Required-route failures aggregate into *MultiEmitError. errors.Is walks each RouteError.Cause; errors.As captures the first *EmitError in the chain.
• IsCallerError on *MultiEmitError returns true only when every required failure is itself caller-correctable. A single infrastructure-class failure flips the answer to false.
• RouteOptional failures never propagate. They surface via the metric counters and (when the route declares DLQ) the route's DLQ destination.

streaming_emitted_total increments N times per Emit fanned across N routes — one per route attempt. Dashboards computing "logical Emits per second" must aggregate per-Emit attempts via trace spans, not by summing per-route counters.

Test

github.com/LerianStudio/lib-streaming/streamingtest is the public test-only helper package. MockEmitter is the only test double — concurrency-safe, deep-copying captured events on Emit so post-hoc caller mutation does not change the captured slice.

goCopy code
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")
streamingtest.AssertEventCount(t, mock, "transaction.created", 1)
streamingtest.AssertNoEvents(t, mock) // when expecting silence

MockEmitter surface

Assertion helpers

WaitForEvent polls at 1ms — fast wall-clock convergence and fully deterministic under testing/synctest. Nil matcher fails the test cleanly instead of panicking mid-loop.

Mocks and real *Producer and NoopEmitter all satisfy the same three-method Emitter interface — write services against the interface, swap implementations in tests.

Manifest

goCopy code
descriptor, err := streaming.NewPublisherDescriptor(streaming.PublisherDescriptor{
    Service:   "ledger",
    Version:   buildinfo.Version,
    Repo:      "github.com/LerianStudio/ledger",
})
if err != nil { return err }

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.

handler, err := streaming.NewStreamingHandler(
    descriptor,
    catalog,
    streaming.WithManifestRoutes(routeTable),
)
if err != nil { return err }
mux.Handle("/streaming/manifest", authMiddleware(handler))

SECURITY: the manifest exposes event taxonomy, schema versions, service metadata, producer IDs, and (when routes are advertised) target names, transport kinds, sanitized broker URLs, and DLQ destinations. Callers MUST wrap the handler in their app's auth middleware before mounting it publicly. The library does not enforce authentication.

Handler-shipped hardening (independent of the caller's auth chain):

• Cache-Control: no-store
• Content-Type: application/json
• X-Content-Type-Options: nosniff
• X-Frame-Options: DENY
• HTTP method allowlist (GET, HEAD only)

CloudEvents codec

For interop layers that consume from non-lib-streaming producers or need to roundtrip an Event through Kafka headers:

goCopy code
headers := streaming.BuildCloudEventsHeaders(event)
// headers = []kgo.RecordHeader, 8-13 entries depending on optional fields

event, err := streaming.ParseCloudEventsHeaders(headers)
// errors.Is(err, streaming.ErrMissingRequiredHeader) for missing context attrs
// errors.Is(err, streaming.ErrUnsupportedSpecVersion) for non-1.0 ce-specversion

The producer path gets canonical bytes automatically through Emit. Use these helpers only when you have a direct CloudEvents Kafka interop need outside the lib-streaming hot path. ce-resourcetype and ce-eventtype are accepted as optional extensions on parse so non-lib-streaming producers can still be parsed (they are populated from the ce-type breakdown when absent).

Errors

Sentinels (29 total)

Use errors.Is(err, sentinel) for branch logic and IsCallerError(err) for the broad caller-vs-infrastructure split.

Caller-side validation (synchronous, no I/O — IsCallerError returns true):

ErrMissingTenantID, ErrSystemEventsNotAllowed, ErrMissingSource, ErrMissingResourceType, ErrMissingEventType, ErrInvalidTenantID, ErrInvalidResourceType, ErrInvalidEventType, ErrInvalidSource, ErrInvalidSubject, ErrInvalidEventID, ErrInvalidSchemaVersion, ErrInvalidDataContentType, ErrInvalidDataSchema, ErrPayloadTooLarge, ErrNotJSON, ErrEventDisabled, ErrInvalidEventDefinition, ErrInvalidOutboxEnvelope, ErrDuplicateEventDefinition, ErrUnknownEventDefinition, ErrInvalidDeliveryPolicy, ErrInvalidPublisherDescriptor, ErrInvalidRouteDefinition, ErrInvalidDestination, ErrDuplicateRouteDefinition, ErrNoRoutesConfigured, ErrMissingTarget, ErrMultiTransportRuntimeNotConfigured, ErrInvalidTLSConfig, ErrPlaintextSASLNotAllowed.

Config validation (LoadConfig):

ErrMissingBrokers, ErrMissingSource, ErrInvalidCompression, ErrInvalidAcks, ErrInvalidConfigField.

Lifecycle / wiring (NOT caller errors — IsCallerError returns false):

ErrEmitterClosed, ErrNilProducer, ErrCircuitOpen, ErrOutboxNotConfigured, ErrOutboxTxUnsupported, ErrNilOutboxRegistry.

CloudEvents codec:

ErrMissingRequiredHeader, ErrUnsupportedSpecVersion.

Error classes

Runtime publish failures surface as *EmitError with one of eight ErrorClass values:

IsCallerError(err) bool

The single decision function. Returns true for caller-correctable faults — empty tenant ID, malformed payload, missing source, etc. Returns false for broker unavailability, circuit open, network timeout, lifecycle violations.

For *MultiEmitError: returns true only when every Required-route failure is itself caller-correctable. A single infrastructure-class failure in the Required set flips the answer to false.

Recommended response shape in handlers:

goCopy code
if err := emitter.Emit(ctx, request); err != nil {
    if streaming.IsCallerError(err) {
        return badRequest(err) // 4xx
    }
    return internalError(err)  // 5xx, alertable
}

End-to-end Builder example

The canonical bootstrap reference. Mirror the order; the only optional setters are the resilience / security / outbox lines — the rest are required for a non-trivial Build.

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

// 1. Panic + assertion metrics, production-mode scrubbing.
runtime.InitPanicMetrics(metricsFactory)
assert.InitAssertionMetrics(metricsFactory)
runtime.SetProductionMode(cfg.Env == "production")

// 2. Immutable catalog.
catalog, err := streaming.NewCatalog(
    streaming.EventDefinition{
        Key: "transaction.created", ResourceType: "transaction", EventType: "created",
        SchemaVersion: "1.0.0",
    },
    streaming.EventDefinition{
        Key: "transaction.posted", ResourceType: "transaction", EventType: "posted",
        SchemaVersion: "1.0.0",
    },
)
if err != nil { return err }

// 3. Disabled-feature-flag fallback BEFORE Builder construction.
if !cfg.Enabled || len(cfg.Brokers) == 0 {
    return inject(streaming.NewNoopEmitter())
}

// 4. Builder chain. Setters are nil-receiver safe and chainable.
emitter, err := streaming.NewBuilder().
    Source(cfg.CloudEventsSource).
    Catalog(catalog).
    Routes(
        streaming.RouteDefinition{
            Key:           "transaction.created.kafka.primary",
            DefinitionKey: "transaction.created",
            Target:        "kafka-primary",
            Destination:   streaming.KafkaTopic("lerian.streaming.transaction.created"),
            Requirement:   streaming.RouteRequired,
        },
        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).
    MetricsFactory(metricsFactory).
    Tracer(tracer).
    CircuitBreakerManager(cbManager).
    CBFailureRatio(0.5).
    CBMinRequests(10).
    CBTimeout(30 * time.Second).
    TLSConfig(tlsCfg).
    SASL(saslMechanism).
    OutboxRepository(outboxRepo).
    CloseTimeout(30 * time.Second).
    Build(ctx)
if err != nil { return err }

// 5. Lifecycle hand-off. Type-assert ONLY here, not in service code.
producer := emitter.(*streaming.Producer)
if err := producer.RegisterOutboxRelay(outboxRegistry); err != nil { return err }
if err := launcher.Add("streaming", producer); err != nil { return err }

// 6. Manifest HTTP mount, wrapped in the app's auth middleware.
descriptor, err := producer.Descriptor(streaming.PublisherDescriptor{
    Service: "ledger", Version: buildinfo.Version, Repo: "github.com/LerianStudio/ledger",
})
if err != nil { return err }

handler, err := streaming.NewStreamingHandler(
    descriptor,
    catalog,
    streaming.WithManifestRoutes(routeTable),
)
if err != nil { return err }
mux.Handle("/streaming/manifest", authMiddleware(handler))

// 7. Inject the interface, NOT the concrete *Producer, into services.
return inject(emitter)

Service code stays unaware of the wiring:

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

func (s *TransactionService) Create(ctx context.Context, input CreateTransactionInput) error {
    // ... domain logic, durable commit ...
    payload, err := json.Marshal(transactionEvent(tx))
    if err != nil { return err }

    return s.emitter.Emit(ctx, streaming.EmitRequest{
        DefinitionKey: "transaction.created",
        TenantID:      tmcore.GetTenantIDContext(ctx),
        Subject:       tx.ID,
        Payload:       payload,
    })
}