Design: enterprise-events

Module layout

packages/universal/events/
├── package.json                  # deps: @suaveplan/error, tslib, zod
├── tsconfig.lib.json
├── vite.config.ts
├── vitest.config.ts              # thresholds 100/100/100/100
├── README.md
├── src/
│   ├── index.ts                  # barrel
│   ├── constants/index.ts
│   ├── errors/
│   │   ├── codes.ts
│   │   ├── events-error.ts
│   │   └── events-error.test.ts
│   ├── envelope/
│   │   ├── envelope.ts           # wrap(type, payload, options)
│   │   ├── envelope.md
│   │   ├── envelope.test.ts
│   │   ├── envelope.types.ts
│   │   └── envelope.schemas.ts   # Zod (opt-in)
│   ├── id/
│   │   ├── id-generator.ts
│   │   ├── id-generator.md
│   │   └── id-generator.test.ts
│   ├── clock/
│   │   ├── default-clock.ts
│   │   ├── default-clock.md
│   │   └── default-clock.test.ts
│   ├── bus/
│   │   ├── event-bus.ts          # EventBus<TMap> class
│   │   ├── event-bus.md
│   │   ├── event-bus.test.ts
│   │   └── event-bus.types.ts
│   ├── interceptor/
│   │   ├── compose.ts            # compose interceptors into a chain
│   │   ├── compose.md
│   │   └── compose.test.ts
│   ├── queue/
│   │   ├── bounded-queue.ts
│   │   ├── bounded-queue.md
│   │   └── bounded-queue.test.ts
│   ├── idempotent/
│   │   ├── idempotent.ts
│   │   ├── idempotent.md
│   │   └── idempotent.test.ts
│   ├── observability/
│   │   ├── with-observability.ts
│   │   ├── with-observability.md
│   │   └── with-observability.test.ts
│   └── __benchmarks__/
│       └── dispatch.bench.ts
├── tests/
│   ├── setup/polyfills.ts
│   ├── dep-graph.test.ts
│   ├── no-disallowed.test.ts
│   └── tree-shake.test.ts
└── CHANGELOG.md

Key design decisions

Typed `EventBus<TMap>`

type EventMap = Record<string, unknown>;

class EventBus<TMap extends EventMap> {
    on<K extends keyof TMap>(type: K, listener: Listener<TMap[K]>): Unsubscribe;
    once<K extends keyof TMap>(type: K, listener: Listener<TMap[K]>): Unsubscribe;
    off<K extends keyof TMap>(type: K, listener: Listener<TMap[K]>): void;
    emit<K extends keyof TMap>(type: K, payload: TMap[K]): EmitResult;
    emitAsync<K extends keyof TMap>(
        type: K,
        payload: TMap[K],
        options?: { signal?: AbortSignal; timeoutMs?: number }
    ): Promise<EmitResult>;
    use(interceptor: Interceptor<TMap>): Unsubscribe;
    clear(): void;
}

EmitResult: { delivered: number; errors: readonly EventsError[] }. Sync emit collects listener errors; async collects + surfaces as errors[] (never throws mid-dispatch; interceptors CAN throw to halt).

Interceptor chain

Interceptors run in registration order; last-registered is innermost. Signature:

type Interceptor<TMap> = <K extends keyof TMap>(
    envelope: EventEnvelope<TMap[K]>,
    next: () => Promise<EmitResult>
) => Promise<EmitResult>;

The bus composes them with compose(interceptors) into a single function. Calling next() advances; not calling it drops the event (no listeners fire). Throwing aborts the emit and propagates the rejection.

BoundedQueue

Independent of the bus — used internally when emitAsync is called with { queue: true } and externally by consumers who want the same backpressure primitive.

interface BoundedQueueOptions {
    capacity: number;
    policy: "drop-oldest" | "drop-newest" | "block" | "throw";
}

class BoundedQueue<T> {
    enqueue(item: T, signal?: AbortSignal): Promise<void>;
    dequeue(): T | undefined;
    drain(): T[];
    get size(): number;
}

block waits for room (Promise resolves when dequeued). drop-oldest evicts the head. drop-newest rejects the new enqueue silently. throw throws EventsError("EVENTS.QUEUE.OVERFLOW"). enqueue with a signal that aborts rejects with EventsError("EVENTS.QUEUE.ABORTED").

`idempotent`

Wraps a listener so a given key only fires once. Default cache is a Map<string, true>; consumers can pass { cache: LRUCache } to bound growth.

function idempotent<T>(
    listener: Listener<T>,
    options: { keyOf: (payload: T) => string; cache?: IdempotencyCache }
): Listener<T>;

`withObservability`

Not an interceptor — a passive adapter that receives hooks. Counters/histograms/spans live in events-metrics / events-otel; this package exposes only the hook shape.

Determinism

Two single-location module references:

// clock/default-clock.ts
export const defaultClock = (): number => Date.now();
// id/id-generator.ts
export const defaultIdGenerator = (): string => crypto.randomUUID();

Every other file accepts an injection. The two defaults are the only places Date.now() / crypto.randomUUID() appear in src/.

Error codes

export const EVENTS_CODES = defineCodes("EVENTS", [
    "LISTENER.THREW",
    "INTERCEPTOR.THREW",
    "QUEUE.OVERFLOW",
    "QUEUE.ABORTED",
    "QUEUE.INVALID_CAPACITY",
    "SCHEMA.REJECTED",
    "EMIT.TIMED_OUT",
    "EMIT.ABORTED",
] as const);

Coverage strategy

Enumeration tests for every branch
Property-based (fast-check) for: ordering invariants, interceptor composition (associativity), queue policy state machines, idempotency
No istanbul ignore pragmas — defensive branches removed or rewritten as as assertions with invariant comments
Module-level determinism enforced by tests/no-disallowed.test.ts (greps for Math.random / Date.now / crypto.randomUUID outside the two approved files)

Non-goals

Not shipping an OTel SDK — withObservability produces the hook shape; events-otel will implement the adapter
Not shipping a transport — events-transport plugs in as an interceptor
Not shipping metrics — events-metrics plugs in via withObservability
Not shipping a schema registry — events-schema plugs in as an interceptor

6.2 KiB Raw Permalink Blame History