/using-lib-streaming

name: ring:using-lib-streaming
description: "Using lib-streaming, Lerian's producer-only event publication library (Kafka/SQS/RabbitMQ/EventBridge), in two modes. Sweep Mode detects DIY publishers (franz-go, sarama, amqp091, watermill, raw AWS SDK) and re-rolled manifests/breakers. Reference Mode catalogs the Builder/Emitter/Catalog facade. Companion to ring:instrumenting-streaming-events. Go-only. Skip for non-Go or consumer-only."

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

| Request shape | Mode | |---|---| | "Sweep / audit / find opportunities / migrate publishers to lib-streaming" | **Sweep** | | "Replace our DIY franz-go producer with lib-streaming" | **Sweep** | | "What does lib-streaming provide for X?" | **Reference** | | "How do I initialize Y from lib-streaming?" | **Reference** | | "Which `Destination` helper for EventBridge?" | **Reference** | | "Show me the Builder chain end-to-end" | **Reference** |

---

SWEEP MODE

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

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 (exclus