# Add @suaveplan/api-plugin-event-bus — typed event-bus plugin wrapping @suaveplan/events on the @suaveplan/api foundation **Package**: `@suaveplan/api-plugin-event-bus` (NEW, server-tier) **Status**: PROPOSED **Type**: new-package - **Spec delta**: ./specs/api-plugin-event-bus/spec.md ## Why Genesis ships `@suaveplan/events` (typed event-bus core — pub/sub, async dispatch, interceptors, bounded-queue backpressure, idempotent listener wrapper) and `@suaveplan/api` (orchestrator) plus `@suaveplan/api-plugin` (`definePlugin` foundation). Multi-tenant apps composing `@suaveplan/api` repeatedly hand-roll the wiring that turns the raw `EventBus` into a per-tenant-scoped, request-aware service: construct the bus, register interceptors and the bounded queue, install middleware that scopes topics by `c.get("tenant").id`, expose a typed `eventBusService` slot for downstream plugins, and optionally publish/manage subscriptions over HTTP. Every consumer rewrites the same six steps and gets the per-tenant isolation slightly wrong each time. This package wraps `@suaveplan/events` as a single `definePlugin` consumer so apps get a working, instrumented, per-tenant-scoped event bus by adding one entry to their plugins array. It is the second genesis-native plugin (after `api-plugin-auth`) that follows the canonical 3-way plug-and-play contract (Zod schemas at the boundary + OpenAPI route registration + canonical service-slot registration — no `.surql` DDL because the bus is in-memory). > **Naming note.** `enterprise/api-plugin-events.md` documents a *different* package (social calendar / RSVP / meetup feed) that is deferred to a later wave. This `event-bus` plugin is intentionally singular in noun and scoped to the `@suaveplan/events` library wrapper. ## What changes Net-new package at `0.1.0`. ADDED: ### Plugin factory + descriptor surface - `createEventBusPlugin(options: EventBusPluginOptions): PluginDescriptor` — Zod-validated factory returning a `PluginDescriptor` from `@suaveplan/api-plugin`. The descriptor's lifecycle hooks implement the canonical 6-step onInit flow. - `EventBusPluginOptions` (Zod-validated via `EventBusPluginOptionsSchema`): `{ interceptors?: readonly Interceptor[]; backpressure?: BackpressureOptions; idempotency?: IdempotencyOptions; routePrefix?: string (default "/event-bus"); mountRoutes?: boolean (default true); requireAuthForRoutes?: boolean (default true); tenantIsolation?: boolean (default true); maxTopicLength?: number (default 256); logger?: LoggerLike }`. - `EventBusPluginConfig` — the descriptor's `configSchema`-inferred shape (resolved options). - `BackpressureOptions` — `{ capacity?: number (default 1024); policy?: QueuePolicy (default "drop-oldest") }`. - `IdempotencyOptions` — `{ enabled?: boolean (default false); cacheCapacity?: number (default 4096) }`. ### Topic naming + per-tenant scoping - `EVENT_TOPIC_PATTERN` — `/^[a-z0-9_\-:.]+$/` regex; all `publish`/`subscribe` topics MUST match. - `scopeTopic(topic, tenantId?)` — returns `tenant::` when `tenantId` is provided and `tenantIsolation` is enabled, else returns `topic` unchanged. Pure function; exported so test code and downstream plugins can compute the canonical scoped form. - `unscopeTopic(scoped, tenantId?)` — inverse of `scopeTopic`; returns the original topic, or `null` when the scoped string does not match the tenant prefix. ### Service-registry slot — `eventBusService` - `eventBusService` registered via `ctx.registerService("eventBusService", ...)`. Surface: - `publish(topic: string, payload: unknown, opts?: PublishOptions): Promise` - `subscribe(topic: string, handler: Listener, opts?: SubscribeOptions): SubscriptionHandle` - `unsubscribe(handleId: string): boolean` - `listSubscriptions(filter?: SubscriptionFilter): readonly SubscriptionRecord[]` - `getRawBus(): EventBus` — escape hatch for downstream consumers that need direct access (e.g. `use(interceptor)` at runtime). - `SubscriptionHandle` — `{ id: string; topic: string; tenantId?: string; unsubscribe(): boolean }`. - `PublishOptions` — `{ tenantId?: string; idempotencyKey?: string; traceId?: string; signal?: AbortSignal; timeoutMs?: number }`. - `SubscribeOptions` — `{ tenantId?: string; idempotencyKey?: (payload: unknown) => string; once?: boolean }`. - `PublishResult` — re-export of `EmitResult` from `@suaveplan/events` (`{ delivered: number; errors: readonly EventsError[] }`) plus `{ dropped: boolean }` whose value reflects whether the backpressure queue overflowed. - ServiceRegistry module augmentation: `declare module "@suaveplan/api-plugin/service-registry" { interface ServiceRegistry { eventBusService: EventBusService } }`. ### Middleware - `requestBusMiddleware()` — sets `c.set("eventBus", scopedHandle)` on every request. `scopedHandle` is a thin façade over `eventBusService` that auto-fills `tenantId` from `c.get("tenant")?.id`. Hono `ContextVariableMap` augmentation: `interface ContextVariableMap { eventBus: ScopedEventBusHandle }`. - `ScopedEventBusHandle` — same `publish`/`subscribe`/`unsubscribe`/`listSubscriptions` surface as `EventBusService` minus the explicit `tenantId` parameter (it is bound from the request context). ### Optional HTTP routes Mounted under `routePrefix` (default `/event-bus`). Each route is gated by `requireAuth()` from `@suaveplan/api-plugin-auth` when `requireAuthForRoutes: true` (the default); otherwise they are public. When `mountRoutes: false`, no routes are mounted. - `POST /event-bus/publish` — body: `{ topic, payload, idempotencyKey?, timeoutMs? }`; response: `{ delivered, dropped, errors: ErrorReportSchema[] }`. Tenant is read from `c.get("tenant")`. - `GET /event-bus/subscriptions` — response: `{ subscriptions: SubscriptionRecordSchema[] }`. Tenant-scoped — only subscriptions owned by the current tenant are returned. - `DELETE /event-bus/subscriptions/:id` — params: `{ id }`; response: `{ unsubscribed: boolean }`. Returns 404 when the id is unknown to the current tenant (404 to avoid leaking cross-tenant ids). ### Backpressure + idempotency wiring - `BoundedQueue` from `@suaveplan/events` wraps every `publish` so a slow listener cannot block faster ones. When the queue overflows under `drop-oldest` / `drop-newest`, the returned `PublishResult.dropped` is `true` and the `event_bus.dropped_total{policy}` counter increments by 1. Under `policy: "throw"` the publish promise rejects with `EventBusError(EVENT_BUS.BACKPRESSURE_DROPPED)`. - The `idempotent` wrapper from `@suaveplan/events` is composed at subscribe time when `opts.idempotencyKey` (or the plugin-level `idempotency.enabled`) is configured. The keyOf function defaults to `(envelope) => envelope.idempotencyKey ?? envelope.id`; duplicates are silently dropped and counted via `event_bus.duplicates_total`. ### Errors + observability - `EventBusError extends SuaveplanError` via `defineCodes("EVENT_BUS", [...])`. Codes: - `INVALID_INPUT` — Zod parse of `EventBusPluginOptions` or a route body failed. - `INVALID_TOPIC` — topic failed `EVENT_TOPIC_PATTERN` or exceeded `maxTopicLength`. - `SERVICE_NOT_AVAILABLE` — a required `ServiceRegistry` slot was absent at init. - `ROUTE_PREFIX_COLLISION` — `routePrefix` collides with previously registered routes. - `SUBSCRIPTION_NOT_FOUND` — `unsubscribe(id)` called with an unknown id. - `BACKPRESSURE_DROPPED` — queue overflowed under `policy: "throw"`. - `TENANT_MISMATCH` — handler tried to operate on another tenant's subscription id. - `PUBLISH_FAILED` — underlying `EventBus.emitAsync` rejected (wraps the `EventsError` as `cause`). - OTel: counters `event_bus.published_total{outcome,topic_root}`, `event_bus.dropped_total{policy}`, `event_bus.duplicates_total`, `event_bus.subscriptions_active`, histogram `event_bus.publish_duration_ms`. ### Subpath exports `.`, `./plugin/plugin`, `./middleware/request-bus`, `./middleware/context-augmentation`, `./services/event-bus-service`, `./services/registry-augmentation`, `./schemas/options`, `./schemas/routes`, `./routes/build-routes`, `./topic/topic`, `./error-codes/error-codes`, `./otel/instruments`, `./types/types`. ## Out of scope - **Cross-process / cross-node delivery.** This plugin wraps `@suaveplan/events` which is intentionally in-process. A future `api-plugin-event-bus-redis` (or similar transport) would extend it. - **Persistence.** Subscriptions are in-memory and per-process — restarting the API server clears them. A persistent registry belongs in a sibling plugin. - **WebSocket / SSE fan-out.** Operators wanting a push surface compose `api-plugin-sse` separately and call `eventBusService.subscribe` from inside that plugin. - **Per-event RBAC beyond the `requireAuth` route gate.** Fine-grained authorisation (e.g. "tenant admins can publish, tenant members cannot") composes `requireRole` at the application layer. - **Schema validation of event payloads.** Payloads are `unknown` at the bus boundary — callers Zod-validate at the producer/consumer if they want a contract. - **Dead-letter queue.** Listener errors are surfaced via `PublishResult.errors` and counted; a DLQ persistence plugin would be a follow-up. ## Source mapping This package is **genesis-native** — there is NO upstream `enterprise/api-plugin-event-bus.md` (the unrelated `enterprise/api-plugin-events.md` documents a social-calendar package shipping in a later wave). Design intent draws from: - The shipped `@suaveplan/events` library (`EventBus`, `BoundedQueue`, `idempotent`, `compose` interceptor chain). - The shipped `@suaveplan/api-plugin` `PluginContext` + `ServiceRegistry` shape. - The shipped `@suaveplan/api-openapi` `definePluginRoutes` + `RouteBuilder` pattern. - The shipped `@suaveplan/api-plugin-auth` reference implementation of the 6-step `onInit` lifecycle (verbatim layout for tests, dirs, and the spec-conformance file). - The shipped `@suaveplan/api-tenant` `c.get("tenant")` augmentation. ### Requirements with no source line | Requirement | Justification | |---|---| | createEventBusPlugin factory | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | 6-step onInit lifecycle | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | Per-tenant topic scoping | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | Tenant isolation enforcement | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | eventBusService registry slot | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | ScopedEventBusHandle on c.get("eventBus") | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | Topic name validation | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | Bounded-queue backpressure wiring | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | Idempotency wrapper composition | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | Optional HTTP route surface (publish / list / delete) | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | Zod boundary validation for routes + options | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | OTel instruments (5 metrics) | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | 8-code EventBusError registry | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | | Subpath exports per feature | Genesis-native event-bus integration plugin — no upstream parity; design driven by `@suaveplan/events` library contract | ## Impact ### Affected dependents Run `bunx turbo ls --filter=...@suaveplan/api-plugin-event-bus` to enumerate. As of authoring time the package is net-new with no dependents. No dependents (verified by `bunx turbo ls --filter=...@suaveplan/api-plugin-event-bus` returning only the package itself). ### Catalog additions required None — every catalog key used (`catalog:server`, `catalog:observability`, `catalog:`, `catalog:forms`, `catalog:build`, `catalog:testing`) already exists in the root `package.json`.