@dokebi/event-bus

An optional plugin for local and NATS Core event delivery. Node.js >=24.0.0, @dokebi/core ^0.4.0, version 0.4.0 (unpublished). Only network buses need @dokebi/nats. AI agents start with AI.md.

Install and configure

dokebi add @dokebi/event-bus calls planInstall of the install module dist/install.js. The plugin returns the src/infra/events/ folder, the src/infra/events/types.ts template, a default local bus and the requires list for the current options. Existing user values are kept.

planInstall and check share one decision function. With any NATS bus it is requires: ['dokebi-nats'], otherwise []. The plugin no longer throws NATS_REQUIRED; the CLI decides and rejects a missing plugin (MISSING_REQUIRED_PLUGIN). check (re-check by doctor, add, remove) does not throw on invalid options but returns diagnostics: [{ code: 'INVALID_OPTIONS', severity: 'error' }]. The manifest (schemaVersion 2) declares dokebi-nats as optional: true, and package.json lists @dokebi/nats as an optional peer and a devDependency.

schemaVersion: 1
options:
  dokebi-event-bus:
    buses:
      local:
        driver: local
      shared:
        driver: nats
        connection: primary
        subjectPrefix: myapp.events

| Key (per bus) | Type | Default | Description | |---|---|---|---| | driver | local | nats | local | Delivery (0.4.0 renamed transport → driver; the old key works with an OPTION_DEPRECATED warning, D-0071) | | connection | string | — | NATS connection name (required for nats) | | subjectPrefix | string | dokebi.events.<bus> | Subject prefix of a nats bus | | maxPending (reloadable) | integer 1–65536 | 4096 | Deliveries in progress (and, separately, network publishes awaiting their flush) | | maxWaiting (reloadable) | integer 0–65536 | 4096 | nats only: received events waiting for a delivery slot | | maxWaitingBytes (reloadable) | integer 0–1073741824 | 16777216 | nats only: envelope bytes of the waiting events | | retryAfterMs (reloadable) | integer 1–60000 | 1000 | data.retryAfterMs of OVERLOADED and BUS_DRAINING (Retry-After of the 503) | | maxListeners (reloadable) | integer | 256 | Subscriptions per bus | | maxPayloadBytes (reloadable) | integer | 1048576 | Payload JSON size | | maxPayloadNodes (reloadable) | integer 1–1000000 | 10000 | Values in one payload | | maxPayloadDepth (reloadable) | integer 0–64 | 32 | Nesting of one payload | | shutdownTimeoutMs (reloadable) | integer | 5000 | Cleanup deadline |

NATS addresses and authentication go in options.dokebi-nats.connections. connection names the connection to use; a missing or unknown name is an error. A network bus prepares the NATS provider automatically and shares one already registered. There is no need to edit the generated registration file or register NATS again. The conditional dependency is decided by dependencies({ options }) of definePlugin, written in this package as providerDependency({ ..., load: () => import('@dokebi/nats') }). Without the package, start fails with DEPENDENCY_CREATE_FAILED (cause DEPENDENCY_NOT_INSTALLED).

const events = createEvents({ app: app.view });
app.register(events);
await app.start();

const bus = app.get(events).bus('shared');
const subscription = bus.subscribe<{ orderId: string }>(
  'order.created',
  (payload, context) => {
    app.logger.log('info', 'Order event', { id: context.id, orderId: payload.orderId });
  },
);
await subscription.ready;
await bus.publish('order.created', { orderId: 'A1' });
await subscription.unsubscribe();
await app.stop();

createEvents is createPlugin of @dokebi/event-bus, defined with definePlugin of plugin contract v2 (D-0028). The factory input is { app, options? }; app is required. Given options are used, otherwise app.config['dokebi-event-bus']. An option error is core INVALID_PLUGIN_OPTIONS (cause: the core INVALID_OPTIONS, with a list of paths and problems in details.issues and no values; since 0.4.0 one defineOptions declaration in src/options.ts builds both the schema and the parser, D-0063). The dependencies input is gone (INVALID_PLUGIN_INPUT). To use a specific NATS instance, register it under the same name first, e.g. app.register(createNats({ app: app.view })). A local-only app never loads the NATS package.

Delivery contract

• Event names: dot-separated tokens starting with a letter; letters, digits, underscore and hyphen; at most 128 characters. Wildcard subscriptions are not public.
• JSON payload: finite numbers, strings, booleans, null, arrays and plain objects. Proxies, getters, cycles, BigInt, undefined and sparse arrays are rejected. Depth (maxPayloadDepth, 32), nodes (maxPayloadNodes, 10000) and the byte limit are checked, and an immutable snapshot is shared. The generic T is the caller's type; runtime validation of the business schema is the consumer's job.
• A local publish runs every matching handler and waits for them. When a handler fails the rest still run, and the publisher gets HANDLER_FAILED with an AggregateError cause. Calls follow registration order; async completion order is not guaranteed.
• A NATS publish means sent, not processed remotely. Events can be lost during a reconnect and events before a subscription is ready are not replayed. There is no persistence, retry or exactly-once. Work that must happen uses a job.
• Apps using the same subjectPrefix share events, and a publisher that also subscribes receives its event once. Local buses are never propagated automatically.
• once removes the registration before the call. unsubscribe and an outside signal block later calls and pass a cooperative cancellation signal to running handlers. unsubscribe does not wait for handlers already running.

Limits, observation, shutdown

At most 32 buses. maxPending (default 4096, max 65536; 0.4.2 raised it from 256) limits deliveries in progress (network publishes have the same limit separately). Size it like Redis maxPending (D-0102): the requests one process handles at once × the events each publishes at once; a pending delivery takes about 5 KB of heap plus its payload (4096 ≈ 20 MB with small payloads, measured). maxListeners (default 256, max 4096) is the subscription count of the whole bus. maxPayloadBytes (default and max 1048576) is the payload JSON size; leave about 512 bytes of NATS envelope room under the provider's send limit; an envelope over that limit is refused with INVALID_PAYLOAD before anything is sent. shutdownTimeoutMs defaults to 5000, max 300000.

Publish overload returns OVERLOADED to the caller: a retryable error with status 503 and data.retryAfterMs (retryAfterMs), which dokebi-web answers as 503 with Retry-After (D-0108). Network receive overload never fails the bus (0.4.2): an event received while maxPending deliveries run waits, in order, and is delivered when a slot frees; past maxWaiting events or maxWaitingBytes bytes waiting it is dropped and logged at error (OVERLOADED, counted per 10 s window). Receiving never pauses, because a paused reader would fill the NATS connection's shared subscription buffer and fail other subscriptions. When the NATS subscription itself ends because a burst overflowed that buffer (its events are lost) or a message over the connection's maxPayloadBytes arrived, that is logged at error (OVERLOADED, INVALID_PAYLOAD) and the bus subscribes again, still ready; any other end of it fails the bus. Drain delivers the waiting events before it waits for handlers; those still waiting at the drain deadline or at cleanup are dropped with one warn EVENT_BUS_WAITING_DROPPED (count, bytes). A failed network handler or an invalid envelope is logged at error with the full error, and receiving goes on (repeats of the same bus and code within 10 seconds are counted, not logged each time). States are starting/ready/draining/closing/closed/failed. During a provider reconnect the bus shows starting; a closed subscription is never recreated automatically.

Shutdown has two phases, drain and cleanup. In drain each bus becomes draining, rejects new publishes and subscribes with BUS_DRAINING and waits for running handlers. A NATS bus first stops receiving (later arrivals are not taken) and waits for running handlers and publishes already received. When core's drainSignal aborts it stops waiting and aborts the running handlers' signals. Cleanup then blocks new work, unsubscribes, cancels what is running and cleans up within a bounded time. Past the deadline it stays failed with SHUTDOWN_TIMEOUT. JavaScript handlers are never killed. The shared NATS connection is owned by the NATS plugin and never closed by the event bus. The public API exposes no start, close or raw provider.

Errors are core DokebiErrors with code, plugin (dokebi-event-bus), and fix/hint where a fix is known; the full code list is in AI.md.

Development and example

See the development document and example 19. Local, controlled-transport and real Docker NATS tests run with npm test --workspace @dokebi/event-bus. The CLI's own install plan checks run separately in the CLI tests.