Event Bus API

The event bus API is built around three layers: the AIEventBus interface (the contract), BaseEventBus (the abstract implementation with telemetry), and InMemoryEventBus (the concrete default implementation). Most application code interacts only with AIEventBus or with the global bus singleton.

Key Concepts

• Global bus — getGlobalEventBus() returns a lazily-created InMemoryEventBus singleton. All @AIEventHandler-decorated classes register onto this bus automatically.
• Custom bus — construct new InMemoryEventBus(options) and pass it wherever you need an isolated bus (e.g., in tests).
• Synchronous vs asynchronous dispatch — by default, handlers are awaited inline. Pass { async: true } in PublishOptions to queue the handler for background execution.

AIEventBus Interface

interface AIEventBus {
  publish<TPayload = any>(
    eventName: string,
    payload: TPayload,
    options?: PublishOptions
  ): Promise<void>;

  publishWithDefinition<TPayload = any>(
    definition: EventDefinition<TPayload>,
    payload: TPayload,
    options?: PublishOptions
  ): Promise<void>;

  registerHandler(definition: EventHandlerDefinition): void;

  unregisterHandler(eventName: string, handler: EventHandlerDefinition): void;

  getHandlers(eventName: string): EventHandlerDefinition[];

  getDiagnostics(): EventBusDiagnostics;
}

BaseEventBus — Abstract Implementation

BaseEventBus implements all interface methods except dispatch. Subclasses implement dispatch to decide how handlers are invoked (synchronously, queued, over a message broker, etc.).

What BaseEventBus provides:

• Handler registration with priority sorting (higher priority number runs first).
• Payload validation against the event definition's schema via SchemaValidator.
• Header generation (traceId, correlationId, emittedAt) via ensureHeaders.
• An events.publish OpenTelemetry span wrapping every publish call.
• Metric emission: events.published counter.

What subclasses must implement:

protected abstract dispatch<TPayload>(
  event: AIEvent<TPayload>,
  options: PublishOptions
): Promise<void>;

InMemoryEventBus

The default implementation. Dispatches events to handlers inline (synchronous by default) or queues them for background execution when options.async === true or a retry delay is set. An internal worker interval (default 25 ms) flushes the async queue.

import { InMemoryEventBus } from '@tagna/udiot';

const bus = new InMemoryEventBus({
  workerIntervalMs?: number;   // How often the async queue is flushed; default 25 ms
  logger?: Pick<Console, 'debug' | 'error' | 'warn'>;  // Default: silent
});

stop()

Clears the internal worker interval and discards queued async jobs. Call this in test teardown or application shutdown:

bus.stop();

publish

await bus.publish(eventName: string, payload: TPayload, options?: PublishOptions): Promise<void>

1. Looks up the EventDefinition in the registry — throws EventValidationError if not found.
2. Opens an events.publish OTel span.
3. Delegates to publishWithDefinition.

publishWithDefinition

await bus.publishWithDefinition(
  definition: EventDefinition<TPayload>,
  payload: TPayload,
  options?: PublishOptions
): Promise<void>

Use this method when you already have the EventDefinition object (e.g., from a constant exported by a definitions module). It skips the registry lookup and provides compile-time type safety for the payload.

1. Validates payload against definition.schema (if present).
2. Builds an AIEvent with auto-generated headers.
3. Emits events.published metric.
4. Opens an events.dispatch OTel span and calls dispatch.

registerHandler

bus.registerHandler(definition: EventHandlerDefinition): void

Adds a handler for the given event name. Handlers for the same event are sorted by options.priority (higher number = called first). Calling registerHandler multiple times with the same handler object is idempotent in practice but will result in duplicate registrations — use unregisterHandler first if you need to update options.

interface EventHandlerDefinition<TPayload = any> {
  eventName: string;
  handler: EventHandler<TPayload>;
  options: EventHandlerOptions;
  target?: any;
}

EventHandlerOptions

interface EventHandlerOptions {
  priority?: number;        // Higher runs first; default 0
  maxAttempts?: number;     // Retry count for async handlers; default 3
  retryDelayMs?: number;    // Delay between retries in async mode; default 1000 ms
  filters?: EventFilterOptions; // Optional category/tag filters
  description?: string;
  aiGenerated?: boolean;
}

EventFilterOptions

interface EventFilterOptions {
  categories?: EventCategory[];  // Handler only runs if event.category matches
  tags?: string[];               // Handler only runs if event has at least one matching tag
}

unregisterHandler

bus.unregisterHandler(eventName: string, handler: EventHandlerDefinition): void

Removes the given handler definition from the list for eventName. Uses reference equality, so pass the same EventHandlerDefinition object that was passed to registerHandler.

getDiagnostics

bus.getDiagnostics(): EventBusDiagnostics

interface EventBusDiagnostics {
  inFlight: number;      // Handlers currently executing
  queued: number;        // Items waiting in the async queue
  lastDispatch?: number; // Unix ms of the most recent dispatch
}

Each dispatch call also records a diagnostics snapshot into the registry (accessible via getDiagnosticsHistory()), keeping the last 50 snapshots.

PublishOptions

interface PublishOptions {
  async?: boolean;           // Queue handlers for background execution; default false
  delayMs?: number;          // Delay before async handlers execute
  correlationId?: string;    // Custom correlation ID; auto-generated if omitted
  traceId?: string;          // Custom trace ID; auto-generated if omitted
  causationId?: string;      // ID of the event that caused this one
  source?: string;           // Identifier for the publishing module
  context?: Record<string, any>; // Arbitrary context forwarded to handlers via AIEvent.context
}

AIEvent<TPayload>

The envelope passed to every handler:

interface AIEvent<TPayload = any> {
  metadata: EventDefinition<TPayload>;   // The event definition
  payload: TPayload;                     // Validated and cast payload
  headers: EventHeaders;                 // Auto-generated trace/correlation IDs
  context?: Record<string, any>;         // Context from PublishOptions.context
}

interface EventHeaders {
  traceId: string;
  correlationId: string;
  causationId?: string;
  emittedAt: number;         // Unix ms when the event was published
  source?: string;
}

Global Bus

import { getGlobalEventBus, setGlobalEventBus } from '@tagna/udiot';

// Get the default InMemoryEventBus singleton
const bus = getGlobalEventBus();

// Replace the global bus (e.g., for testing or a different provider)
setGlobalEventBus(myCustomBus);

The global bus is created lazily on the first call to getGlobalEventBus() and registered in the provider registry under the name 'in-memory'.

Basic Usage Example

import 'reflect-metadata';
import {
  getGlobalEventBus,
  registerEventDefinition,
  EventDefinition,
  AIEvent,
  EventHandlerContext,
} from '@tagna/udiot';

// 1. Define the event (typically done at module load time)
interface UserRegisteredPayload {
  userId: string;
  email: string;
  plan: 'free' | 'pro';
}

const UserRegisteredEvent: EventDefinition<UserRegisteredPayload> = {
  name: 'user.registered',
  category: 'domain',
  version: 1,
  description: 'Emitted when a new user completes registration',
};

registerEventDefinition(UserRegisteredEvent);

// 2. Register handlers
const bus = getGlobalEventBus();

bus.registerHandler({
  eventName: 'user.registered',
  handler: async (event: AIEvent<UserRegisteredPayload>, ctx: EventHandlerContext) => {
    await sendWelcomeEmail(event.payload.email);
  },
  options: { priority: 10, description: 'Send welcome email' },
});

bus.registerHandler({
  eventName: 'user.registered',
  handler: async (event: AIEvent<UserRegisteredPayload>) => {
    await analytics.track('user_registered', { userId: event.payload.userId });
  },
  options: { priority: 5, description: 'Track in analytics' },
});

// 3. Publish from AuthService
await bus.publishWithDefinition(UserRegisteredEvent, {
  userId: 'usr_001',
  email: 'alice@example.com',
  plan: 'pro',
});

The email handler runs before the analytics handler because its priority (10) is higher.

Common Patterns

Async dispatch with retry

await bus.publish(
  'payment.failed',
  { orderId: 'ord_123', reason: 'insufficient_funds' },
  {
    async: true,
    source: 'payment-service',
  }
);

Register the handler with retryDelayMs to enable automatic retries on failure:

bus.registerHandler({
  eventName: 'payment.failed',
  handler: notifyCustomerOfFailure,
  options: {
    maxAttempts: 3,
    retryDelayMs: 2000,
  },
});

Trace propagation between events

// When publishing a downstream event from within a handler
bus.registerHandler({
  eventName: 'order.placed',
  handler: async (event: AIEvent) => {
    await bus.publish('inventory.reserved', { /* … */ }, {
      causationId: event.headers.traceId,
      correlationId: event.headers.correlationId,
    });
  },
  options: {},
});

Filtered handler

bus.registerHandler({
  eventName: 'product.updated',
  handler: reindexSearchCatalog,
  options: {
    filters: {
      categories: ['domain'],
      tags: ['catalog'],
    },
  },
});

The handler only runs if the event's category is 'domain' and it has at least one tag matching 'catalog'.

Integration Notes

• BaseEventBus uses the shared UDIOT telemetry (startSpan, getMetricsRegistry). If your application has not initialised the telemetry SDK, spans are no-ops and metrics are dropped silently.
• The InMemoryEventBus worker interval (workerIntervalMs, default 25 ms) is unref'd so it does not prevent Node.js from exiting. Always call bus.stop() in test teardown to avoid spurious post-test handler executions.

Pitfalls

• publish throws EventValidationError synchronously if the event is not registered. Wrap in try/catch or ensure definitions are registered at module load time.
• Handlers that run synchronously (the default) share the same call stack as the publisher. An unhandled handler error will propagate up through publish unless the handler is async-queued.
• unregisterHandler uses reference equality. If you create a new handler object literal each time, you cannot unregister it. Store handler definitions in a variable.