Ring:Instrumenting Streaming Events

Streaming Instrumentation (lib-streaming)

When to use

• User requests streaming instrumentation for a Go service with a validated docs/streaming/instrumentation-map.json from ring:mapping-streaming-events
• Task mentions "wire lib-streaming", "instrument streaming events", "implement event emission", "add streaming.NewBuilder", "Emit business events", "lib-streaming bootstrap"

Skip when

• Service is not a Go project
• No instrumentation-map.json present (run ring:mapping-streaming-events first)

You orchestrate. Agents implement. NEVER use Edit/Write/Bash on Go source files. All code changes go through Task(subagent_type="ring:backend-go"). TDD mandatory for all implementation gates (RED → GREEN → REFACTOR).

Streaming Architecture

lib-streaming: producer-only event-emission library. Three-step lifecycle:

1. streaming.NewCatalog(definitions ...EventDefinition) (Catalog, error) — declare every event up-front (immutable)
2. streaming.NewBuilder().Source(...).Catalog(catalog).Routes(...).Target(...).Logger(...).MetricsFactory(...).Tracer(...).CircuitBreakerManager(...).OutboxRepository(...).Build(ctx) — Builder pattern returns (Emitter, error). There is NO NewProducer constructor; *streaming.Producer is reachable only by type-asserting the Emitter returned from Build(ctx), and only when lifecycle methods (Run, RunContext, RegisterOutboxRelay) are needed.
3. emitter.Emit(ctx, EmitRequest{DefinitionKey, TenantID, Subject, Payload}) from handlers/workers

The Emitter interface has THREE methods — Emit(ctx, EmitRequest) error, Close() error, Healthy(ctx) error. Mocks and adapters MUST implement all three.

Wire format: CloudEvents 1.0 binary mode. Each RouteDefinition picks a transport: Kafka (topic lerian.streaming.<resource>.<event>[.vN]), SQS (queue URL), RabbitMQ (exchange + routing key), EventBridge (bus name), or Custom. Tenant carried on ce-tenantid header for CloudEvents-binary transports.

WebFetch URLs (include in every gate dispatch):

• https://raw.githubusercontent.com/LerianStudio/lib-streaming/main/doc.go
• https://raw.githubusercontent.com/LerianStudio/lib-streaming/main/AGENTS.md
• https://raw.githubusercontent.com/LerianStudio/lib-streaming/main/CHANGELOG.md

Three delivery postures:

Canonical import paths:

Emitter implementations:

Service code depends on streaming.Emitter INTERFACE. MUST NOT type-assert to *Producer except in bootstrap to wire Run(launcher) / RunContext(ctx, launcher) / RegisterOutboxRelay(registry). All three implementations satisfy the full three-method interface (Emit, Close, Healthy).

Mandatory agent instruction (include in EVERY dispatch):

WebFetch https://raw.githubusercontent.com/LerianStudio/lib-streaming/main/doc.go and AGENTS.md. docs/streaming/instrumentation-map.json is the canonical contract — every EventDefinition, Emit site, DeliveryPolicy MUST match exactly. Tenant from tmcore.GetTenantIDContext(ctx) — NEVER hardcode. TDD: RED → GREEN → REFACTOR for every gate.

Gate Overview

Gates execute sequentially. Gate 5 skip: only if zero events have outbox != "never". Gate 6 skip: only if service has zero HTTP surface (justify in report).

Gate 0: Stack Detection

Orchestrator executes directly. Runs 3 phases:

Phase 1: Stack Detection

bashCopy code
grep "lib-streaming" go.mod
grep "lib-commons" go.mod
grep -rn "postgresql\|pgx" internal/ go.mod
grep -rn "outbox" go.mod
grep -rn "fiber\|gin\|echo\|net/http" internal/
# Existing lib-streaming code:
grep -rn "streaming.NewBuilder\|streaming.NewCatalog\|streaming.NewNoopEmitter\|streaming.Emitter\|streaming.Producer" internal/
# Non-canonical (must remove):
grep -rn "sarama\|watermill\|segmentio/kafka-go\|amqp091.Publish\|franz-go" internal/

Phase 2: instrumentation-map.json Validation

Copy code
Read docs/streaming/instrumentation-map.json
Validate: JSON well-formed, required fields present (service_name, events[])
Each event must have: definition_key, resource, event_type, delivery_policy
delivery_policy must have: direct (bool), outbox (enum), dlq (string)
CRITICAL events must have outbox = "always"

Phase 3: Existing Compliance Audit (if lib-streaming code detected)

• Construction uses streaming.NewBuilder()...Build(ctx) — NOT a hand-rolled NewProducer shim
• .Catalog(catalog) builder method invoked before .Build(ctx)
• .Source(...), .Routes(...), .Target(...) configured (Builder fails fast on missing required wiring)
• Target names contain no control characters and are ≤256 bytes (Builder validates; service must construct safe names)
• STREAMING_ENABLED feature flag present
• commons.Launcher.Add (or Run/RunContext) lifecycle wiring; Close() on shutdown
• Healthy(ctx) wired into readiness probe
• No non-canonical alternatives

State Management

State: docs/ring:instrumenting-streaming-events/current-cycle.json

Write state after EVERY gate. If write fails → STOP.

jsonCopy code
{
  "service_name": "",
  "started_at": "",
  "gates_completed": [],
  "detected": { "lib_streaming_pinned": false, "has_http": false, "outbox_required": false },
  "instrumentation_map_path": "docs/streaming/instrumentation-map.json",
  "tasks": []
}

Severity Reference