# Capability Spec: api-plugin-event-bus **Package**: `@suaveplan/api-plugin-event-bus` **Status**: SHIPPED (0.1.0, 2026-05-19, folded from add-api-plugin-event-bus) > Every requirement carries a `**Verified by:**` line pointing at the concrete test that pins it. ## Requirements ### Requirement: createEventBusPlugin factory validates options and returns a typed PluginDescriptor The package SHALL export `createEventBusPlugin(options)` which Zod-validates `options` against `EventBusPluginOptionsSchema` and returns a `PluginDescriptor` produced by `definePlugin` from `@suaveplan/api-plugin`. Invalid input SHALL throw `EventBusError(EVENT_BUS.INVALID_INPUT)` carrying the underlying `ZodError` as `cause`. #### Scenario: valid options resolve defaults and return a descriptor - **WHEN** `createEventBusPlugin({})` is called with no overrides - **THEN** the returned descriptor has `name: "event-bus"`, `version: "0.1.0"`, `defaults.routePrefix === "/event-bus"`, `defaults.mountRoutes === true`, `defaults.requireAuthForRoutes === true`, `defaults.tenantIsolation === true`, `defaults.backpressure.policy === "drop-oldest"`, and `defaults.backpressure.capacity === 1024` #### Scenario: invalid options reject at the factory boundary - **WHEN** `createEventBusPlugin({ routePrefix: "no-leading-slash" })` is called - **THEN** the call throws `EventBusError` whose `code === "EVENT_BUS.INVALID_INPUT"` and `cause` is a `ZodError` **Verified by:** `src/plugin/plugin.test.ts::"createEventBusPlugin resolves defaults and returns a typed PluginDescriptor"` AND `src/plugin/plugin.test.ts::"createEventBusPlugin throws EVENT_BUS.INVALID_INPUT for invalid options"` --- ### Requirement: onInit executes the canonical 6-step lifecycle The package's plugin descriptor SHALL execute, in order, an `onInit(ctx)` flow that (1) resolves `app` and (optionally) `openAPIRegistry` from the host `ServiceRegistry`, (2) constructs an `EventBus` from `@suaveplan/events` and applies configured interceptors / backpressure / idempotency, (3) registers `eventBusService` in the `ServiceRegistry`, (4) auto-prefixes topics with `tenant::` whenever publishes/subscribes happen through the scoped handle, (5) installs middleware that puts a `ScopedEventBusHandle` under `c.set("eventBus", ...)`, and (6) mounts the optional HTTP routes under `routePrefix` when `mountRoutes !== false`. #### Scenario: onInit registers the service slot - **WHEN** `onInit(ctx)` runs against a context carrying the `app` slot - **THEN** `ctx.getService("eventBusService")` returns an `EventBusService` object exposing `publish`, `subscribe`, `unsubscribe`, `listSubscriptions`, and `getRawBus` #### Scenario: onInit installs the per-request middleware - **WHEN** a request hits any path on the host app after `onInit` completes - **THEN** the handler observes `c.get("eventBus")` typed as `ScopedEventBusHandle` with `publish`/`subscribe`/`unsubscribe`/`listSubscriptions` methods bound to the request's tenant (if any) #### Scenario: onInit fails when the `app` slot is missing - **WHEN** `onInit(ctx)` runs against a context with no `app` slot in `ServiceRegistry` - **THEN** the call throws `EventBusError(EVENT_BUS.SERVICE_NOT_AVAILABLE)` with `context.missing` containing `"app"` **Verified by:** `src/plugin/plugin.test.ts::"onInit registers eventBusService in the ServiceRegistry"` AND `src/plugin/plugin.test.ts::"onInit installs request-bus middleware that exposes ScopedEventBusHandle"` AND `src/plugin/plugin.test.ts::"onInit throws SERVICE_NOT_AVAILABLE when the app slot is missing"` --- ### Requirement: Per-tenant topic scoping isolates publishes and subscribes When `tenantIsolation` is enabled (the default), every `publish` and `subscribe` flowing through the scoped middleware SHALL prefix the topic with `tenant::` where `` comes from `c.get("tenant")?.id`. A subscriber for tenant A SHALL NOT receive payloads published by tenant B for the same logical topic, and vice versa. #### Scenario: tenant A publish does not reach tenant B subscriber - **WHEN** tenant A's scoped handle publishes `"order.created"` and tenant B's scoped handle subscribes to `"order.created"` - **THEN** tenant B's handler is invoked zero times and tenant A's handler (if subscribed) is invoked exactly once #### Scenario: tenant-less request does not prefix topics - **WHEN** a request without a `tenant` context publishes `"system.heartbeat"` through the scoped handle - **THEN** the underlying bus emits `"system.heartbeat"` verbatim (no `tenant:` prefix) #### Scenario: scopeTopic / unscopeTopic round-trip - **WHEN** `scopeTopic("order.created", "t1")` is computed and then passed to `unscopeTopic(scoped, "t1")` - **THEN** the result is `"order.created"`; calling `unscopeTopic(scoped, "t2")` returns `null` **Verified by:** `src/topic/topic.test.ts::"scopeTopic prefixes with tenant id when tenantId is provided"` AND `src/topic/topic.test.ts::"unscopeTopic returns null for foreign tenant prefix"` AND `tests/integration/tenant-isolation.test.ts::"tenant A publish does not reach tenant B subscriber"` --- ### Requirement: Topic names are validated against EVENT_TOPIC_PATTERN The package SHALL reject any `publish`/`subscribe` topic that fails the regex `/^[a-z0-9_\-:.]+$/` or exceeds the configured `maxTopicLength` (default 256). Validation SHALL apply both at the service layer (`eventBusService.publish`/`subscribe`) and at the HTTP route layer (`POST /event-bus/publish` body schema). #### Scenario: invalid topic at the service layer - **WHEN** `eventBusService.publish("Order Created", {})` is called - **THEN** the call rejects with `EventBusError(EVENT_BUS.INVALID_TOPIC)` whose `context.topic === "Order Created"` #### Scenario: topic exceeding maxTopicLength - **WHEN** the plugin is configured with `maxTopicLength: 8` and `publish("a".repeat(9), {})` is called - **THEN** the call rejects with `EventBusError(EVENT_BUS.INVALID_TOPIC)` whose `context.reason === "exceeds-max-length"` **Verified by:** `src/services/event-bus-service.test.ts::"publish rejects with INVALID_TOPIC when topic violates pattern"` AND `src/services/event-bus-service.test.ts::"publish rejects with INVALID_TOPIC when topic exceeds maxTopicLength"` --- ### Requirement: Bounded-queue backpressure drops on overflow and increments the dropped counter Every `publish` SHALL flow through a `BoundedQueue` from `@suaveplan/events` configured with `backpressure.capacity` and `backpressure.policy`. Under `drop-oldest` / `drop-newest`, an overflowing publish SHALL resolve with `PublishResult.dropped === true` and increment `event_bus.dropped_total{policy}` by 1. Under `policy: "throw"`, the publish SHALL reject with `EventBusError(EVENT_BUS.BACKPRESSURE_DROPPED)`. #### Scenario: drop-oldest overflow surfaces dropped flag and counter - **WHEN** the plugin is configured with `backpressure: { capacity: 1, policy: "drop-oldest" }` and three sequential publishes occur before any drain - **THEN** at least one published result carries `dropped === true` and `event_bus.dropped_total{policy="drop-oldest"}` is incremented at least once #### Scenario: throw policy rejects on overflow - **WHEN** the plugin is configured with `backpressure: { capacity: 1, policy: "throw" }` and a second concurrent publish overflows the queue - **THEN** the second `publish` promise rejects with `EventBusError(EVENT_BUS.BACKPRESSURE_DROPPED)` **Verified by:** `src/services/event-bus-service.test.ts::"drop-oldest overflow returns dropped=true and increments the counter"` AND `src/services/event-bus-service.test.ts::"throw policy rejects with BACKPRESSURE_DROPPED on overflow"` --- ### Requirement: Idempotent subscriptions deliver duplicates exactly once When a subscription is registered with `idempotencyKey` (or the plugin-level `idempotency.enabled` flag is true), repeated publishes carrying the same key SHALL invoke the handler exactly once. The `event_bus.duplicates_total` counter SHALL increment once per duplicate that was suppressed. #### Scenario: duplicate publish with same idempotency key fires handler once - **WHEN** the same payload is published twice with `idempotencyKey: "k1"` to a subscriber whose `idempotencyKey: (env) => env.idempotencyKey ?? env.id` was provided - **THEN** the handler is invoked exactly once and `event_bus.duplicates_total` is incremented exactly once **Verified by:** `src/services/event-bus-service.test.ts::"idempotent subscription fires handler once for duplicate publish"` --- ### Requirement: HTTP routes mount under routePrefix and are auth-gated by default When `mountRoutes !== false`, the plugin SHALL register `POST /event-bus/publish`, `GET /event-bus/subscriptions`, and `DELETE /event-bus/subscriptions/:id` on the host Hono app and on the host's `OpenAPIRegistry` via `definePluginRoutes`. When `requireAuthForRoutes` is `true` (the default), the routes SHALL be gated by `requireAuth()` middleware imported from `@suaveplan/api-plugin-auth`. When a registered route already lives under `routePrefix`, the plugin SHALL throw `EventBusError(EVENT_BUS.ROUTE_PREFIX_COLLISION)`. #### Scenario: mountRoutes false skips route registration - **WHEN** `createEventBusPlugin({ mountRoutes: false })` runs through `onInit` - **THEN** the host's `OpenAPIRegistry` has zero routes tagged `"event-bus"` #### Scenario: route prefix collision throws - **WHEN** the OpenAPI registry already has a `GET /event-bus/foo` route registered and `onInit` runs with the default `routePrefix` - **THEN** the call throws `EventBusError(EVENT_BUS.ROUTE_PREFIX_COLLISION)` whose `context.prefix === "/event-bus"` #### Scenario: publish route returns delivered + dropped on a real request - **WHEN** a `POST /event-bus/publish` request arrives with body `{ topic: "order.created", payload: { id: "o1" } }` and one subscriber is registered for that topic - **THEN** the response body satisfies `PublishResultSchema` with `delivered === 1` and `dropped === false` **Verified by:** `src/routes/build-routes.test.ts::"mountRoutes=false skips route registration"` AND `src/plugin/plugin.test.ts::"onInit throws ROUTE_PREFIX_COLLISION when routes already occupy the prefix"` AND `tests/integration/http-routes.test.ts::"POST /event-bus/publish returns delivered + dropped"` --- ### Requirement: unsubscribe + listSubscriptions enforce per-tenant ownership `eventBusService.unsubscribe(id)` SHALL return `false` (and `DELETE /event-bus/subscriptions/:id` SHALL return HTTP 404) when the supplied id is not owned by the calling tenant. `listSubscriptions(filter)` SHALL filter by the calling tenant when `tenantIsolation` is enabled and a tenant scope is supplied. #### Scenario: tenant B cannot unsubscribe tenant A's subscription - **WHEN** tenant A subscribes, obtains handle `id1`, and tenant B's scoped handle calls `unsubscribe(id1)` - **THEN** the call returns `false` and tenant A's subscription remains active (a subsequent publish from tenant A still delivers) #### Scenario: listSubscriptions filters to the calling tenant - **WHEN** tenant A has 2 subscriptions and tenant B has 1, and tenant A's scoped handle calls `listSubscriptions()` - **THEN** the returned array has length 2 and every record's `tenantId === "tenant-A"` **Verified by:** `tests/integration/tenant-isolation.test.ts::"tenant B cannot unsubscribe tenant A's subscription"` AND `tests/integration/tenant-isolation.test.ts::"listSubscriptions filters by calling tenant"` --- ### Requirement: Every EventBusError code is reachable from production paths The package SHALL declare the 8-code `EVENT_BUS_CODES` registry via `defineCodes("EVENT_BUS", [...])`. Every declared code SHALL be raised from at least one production code path and SHALL be exercised by at least one test in `src/`. #### Scenario: registry exposes the expected 8 codes - **WHEN** `EVENT_BUS_CODES` is enumerated - **THEN** the object owns keys `INVALID_INPUT`, `INVALID_TOPIC`, `SERVICE_NOT_AVAILABLE`, `ROUTE_PREFIX_COLLISION`, `SUBSCRIPTION_NOT_FOUND`, `BACKPRESSURE_DROPPED`, `TENANT_MISMATCH`, and `PUBLISH_FAILED` #### Scenario: every code is raised somewhere in src/ - **WHEN** the spec-conformance test greps `src/` for `EVENT_BUS_CODES.` references - **THEN** every code is referenced by at least one non-test file **Verified by:** `src/error-codes/error-codes.test.ts::"registry exposes the 8 expected codes"` AND `src/api-plugin-event-bus.spec-conformance.test.ts::"every EVENT_BUS_CODES entry is raised from a production path"` --- ### Requirement: OTel instruments are emitted and assertable from production paths The package SHALL emit, via `@opentelemetry/api`'s global `MeterProvider`, the counters `event_bus.published_total{outcome,topic_root}`, `event_bus.dropped_total{policy}`, `event_bus.duplicates_total`, and the up/down counter `event_bus.subscriptions_active`, plus the histogram `event_bus.publish_duration_ms`. All instruments SHALL be created lazily via a `getEventBusInstruments()` singleton that is reset between tests through `resetEventBusInstruments()`. #### Scenario: published_total increments after a publish - **WHEN** a publish flows through `eventBusService.publish("order.created", {})` - **THEN** the in-memory `MeterProvider` records `event_bus.published_total{outcome="success",topic_root="order"}` incremented by 1 #### Scenario: subscriptions_active reflects active subscription count - **WHEN** two subscriptions are added then one is removed - **THEN** `event_bus.subscriptions_active` is at value 1 **Verified by:** `src/otel/instruments.test.ts::"published_total increments after a publish"` AND `src/otel/instruments.test.ts::"subscriptions_active tracks the live subscription count"` --- ### Requirement: Subpath exports enable feature-grained tree-shaking The package's `exports` map SHALL list one entry per feature module so consumers can `import { ... } from "@suaveplan/api-plugin-event-bus//"`. The exports list SHALL match the set in the proposal's "Subpath exports" section verbatim. #### Scenario: every advertised subpath resolves to a real built artefact - **WHEN** the `exports` map keys are enumerated and each `default` path is checked for existence after `bunx turbo build` completes - **THEN** every default path resolves to a real file under `./dist/` **Verified by:** `src/api-plugin-event-bus.spec-conformance.test.ts::"package.json exports list every feature subpath"` ---