215 lines
14 KiB
Markdown
215 lines
14 KiB
Markdown
<!-- markdownlint-disable MD024 -->
|
|
|
|
# Capability Spec Delta: api-plugin-event-bus (add-api-plugin-event-bus)
|
|
|
|
**Package**: `@suaveplan/api-plugin-event-bus`
|
|
**Status**: PROPOSED
|
|
|
|
> Every requirement below carries a `**Verified by:**` line at authoring time. No exceptions.
|
|
|
|
## ADDED 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<EventBusPluginConfig>` 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:<id>:` 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:<id>:` where `<id>` 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.<code>` 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/<feature>/<feature>"`. 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"`
|
|
|
|
---
|
|
<!-- markdownlint-enable MD024 -->
|