/mapping-streaming-events

name: ring:mapping-streaming-events
description: "Mapping the eventable points in a Lerian Go service where lib-streaming should emit past-tense, durable, tenant-scoped business events, producing a PM-validated event catalog and instrumentation-map.json for ring:instrumenting-streaming-events. Three-pass discovery (Survey, Slice, Mark) with a scope fence and delivery postures. Use to inventory eventable points. Skip on non-Go, infra-only, or consumer-only services."

Streaming Event Mapping (lib-streaming, PM-team)

When to use

• User requests an event catalog, eventable-point inventory, or "where should we emit events" map
• PM team prepares a Lerian Go service for client-facing event streaming subscriptions
• Pre-flight to ring:instrumenting-streaming-events
• Task mentions "event mapping", "streaming inventory", "eventable points", "client event subscription"

Skip when

• Service is not a Go project (lib-streaming is Go-only)
• Service has no business logic to emit events about (pure infrastructure, gateway, sidecar, proxy)
• Task is purely documentation, configuration, or non-discovery
• Service is consumer-only with no outbound business event surface

Sequence

**Runs before:** ring:instrumenting-streaming-events

Related

**Complementary:** ring:instrumenting-streaming-events, ring:codebase-explorer, ring:mapping-feature-relationships

Prerequisites

• Go service codebase available for read access
• At least one entry point present (HTTP route, gRPC method, RabbitMQ consumer, scheduled job, webhook)
• Tenant identity resolvable from request context

Orchestrates 3-pass codebase discovery to produce an event catalog and instrumentation map for lib-streaming. You orchestrate. Agents explore. You NEVER read, write, or edit source code directly.

**Announce at start:** "Using ring:mapping-streaming-events through 7 gates (0-7)."

Streaming Architecture

lib-streaming is a producer-only, write-only event-emission library for Lerian Go services. Events are emitted via `emitter.Emit(ctx, EmitRequest{DefinitionKey, TenantID, Subject, Payload})` against the `streaming.Emitter` interface (constructed via `streaming.NewBuilder().Catalog(...).Build(ctx)`). Wire format: CloudEvents 1.0 binary mode. Each `RouteDefinition` selects a transport — Kafka, SQS, RabbitMQ, EventBridge, or Custom. Per-tenant SaaS subscription remains the primary delivery model regardless of transport; Kafka topic naming is `lerian.streaming.<resource>.<event>[.vN]` with tenant carried on the `ce-tenantid` CloudEvents header.

**WebFetch canonical docs:** `https://raw.githubusercontent.com/LerianStudio/lib-streaming/main/doc.go` **WebFetch agent constraints:** `https://raw.githubusercontent.com/LerianStudio/lib-streaming/main/AGENTS.md` **WebFetch changelog:** `https://raw.githubusercontent.com/LerianStudio/lib-streaming/main/CHANGELOG.md`

Scope Fence

A candidate is a **streamable business fact** if and ONLY if ALL of: 1. **Past-tense** — something that already happened (`account.created`, not `account.create`) 2. **Durable** — the underlying state is persisted 3. **Broadcastable** — multiple consumers could legitimately receive it 4. **Tenant-scoped** — the event belongs to a tenant (or marked SystemEvent with explicit justification)

Candidates failing ANY of the four → REJECT with documented reason.

Delivery Postures

| Posture | Direct | Outbox | DLQ | Use when | |---------|--------|--------|-----|----------| | CRITICAL | skip | always | on_routable_failure | Loss is correctness/compliance breach; atomic with DB write | | IMPORTANT | direct | fallback_on_circuit_open | on_routable_failure | Direct in normal ops; survives broker outage | | OBSERVATIONAL | direct | never | never | Analytics-grade; loss acceptable | | CUSTOM | per-event | per-event | per-event | None above fits — requires ≥80 char justification |

Gate Overview

| Gate | Name | Agent | Always? | |------|------|-------|---------| | 0 | Stack Detection | Orchestrator (grep + read) | Yes | | 1 | Pass 1 — Survey | ring:codebase-explorer (single) | Yes | | 2 | Pass 2 — Slice + Scope Fence | ring:codebase-explorer (single) | Yes | | 3 | Pass 3 — Mark | ring:codebase-explorer (parallel, 1/segment) | Yes | | 4 | Catalog Assembly + Validation | Orchestrator (deterministic) | Yes | | 5 | Business Rendering | Orchestrator | Yes | | 6 | PM Team Validation | User (PM team) — NEVER SKIPPABLE | Yes | | 7 | Handoff Package | Orchestrator | Yes |

Gates execute sequentially. Pass 3 (Gate 3) parallelizes internally per segment.

Gate 0: Stack Detection

Orchestrator executes directly. Detect in parallel:

1. Go version:          grep "^go " go.mod | head -1
2. lib-streaming:       grep "lib-streaming" go.mod
3. HTTP framework:      grep -rn "gofiber/fiber\|labstack/echo\|gin-gonic" internal/ go.mod
4. gRPC server:         grep -rn "grpc.NewServer" internal/
5. RabbitMQ command consumers:  grep -rn "lib-commons/v5/commons/rabbitmq" internal/   # command-queue plumbing; eventable publish sites should migrate to lib-streaming
6. Scheduled jobs:      grep -rn "robfig/cron\|time.NewTicker" internal/
7. Webhook receivers:   grep -rn "webhook\|/hooks/" internal/
8. Worker patterns:     grep -rn "commons.Launcher\|commons.App" internal/
9. Tenant source:       grep -rn "tmcore.GetTenantIDContext\|GetTenantID" internal/
10. DDD layout:         ls internal/services/ internal/domain/ 2>/dev/null
11. Database:           grep -rn "jackc/pgx\|database/sql" go.mod

**HARD GATE:** If not Go → STOP. If tenant source undetectable → STOP and ask user.

Gate 1: Pass 1 — Survey

Dispatch `ring:codebase-explorer` to produce `docs/streaming/_pass1-survey.md` with:

• Service Identity (name, purpose, bounded context)
• Entry Point Inventory (HTTP routes, gRPC methods, RabbitMQ consumers, cron jobs, webhooks, CLIs)
• Aggregate Inventory (name, persistence, lifecycle states)
• Tenant Identity Resolution (idiomatic call, where set, reliability