titan-events

official · @omnitron-dev/titan-events · stable

Typed event bus with wildcard / namespaced patterns, decorator-based listeners, async / serial / reduce emit modes, event history with replay, transactions, schema validation, delayed and cron-scheduled events, and rich metrics.

pnpm add @omnitron-dev/titan-events

When you need it

• In-process domain events. Decouple services that should react to each other (e.g., user.created → send welcome email → update analytics) without direct calls.

• Wildcard fan-out. Subscribe to order.* to capture every order-lifecycle event with one handler.

• Cross-module orchestration. A workflow listens to multiple modules' events without becoming a hard dependency of any of them.

• Pub/sub-style replay. A late-joining consumer pulls the recent event history and catches up.

For cross-process delivery use titan-notifications over Rotif or a real message broker. titan-events is in-process.

Quickstart

import { EventsModule } from '@omnitron-dev/titan-events';

@Module({
  imports: [
    EventsModule.forRoot({
      wildcard:     true,
      delimiter:    '.',
      maxListeners: 100,
      history:      { enabled: true, maxSize: 1_000 },
      metrics:      { enabled: true, slowThreshold: 100 },
      concurrency:  10,
    }),
  ],
})
class AppModule {}

import { OnEvent, EVENTS_SERVICE_TOKEN, EventsService }
  from '@omnitron-dev/titan-events';

@Service({ name: 'users' })
class UsersService {
  constructor(@Inject(EVENTS_SERVICE_TOKEN) private readonly events: EventsService) {}

  @Public()
  async create(input: CreateInput) {
    const user = await this.repo.create(input);
    await this.events.emitAsync('user.created', { id: user.id, email: user.email });
    return user;
  }
}

@Injectable()
class WelcomeMailer {
  @OnEvent({ event: 'user.created', async: true })
  async onCreated(payload: { id: string; email: string }) {
    await this.mailer.send(payload.email, 'Welcome');
  }
}

IEventsModuleOptions

Also exported: forRootAsync({ useFactory, inject? }).

EventsService — the API

Emit modes

// Sync — fire-and-forget; returns true if any listener accepted
this.events.emit('user.created', payload);

// Async — concurrent listeners; resolves when all handlers complete
const results = await this.events.emitAsync('user.created', payload);

// Serial — listeners run one after another; resolves with their results
const results = await this.events.emitSerial('user.created', payload);

// Reduce — pipe payload through listeners, each transforming the previous result
const final = await this.events.emitReduce('user.transform', input, initialValue);

emitAsync propagates failures from regular and wildcard handlers, including synchronous throws. Multiple failed groups produce AggregateError. Handler error policies still apply. Await and handle this promise at the operation boundary; event delivery does not roll back committed domain effects.

Subscribe / unsubscribe

Introspection / patterns

Transactions

const tx = this.events.beginTransaction();

tx.emit('order.created', order);
tx.emit('audit.log',     { kind: 'order.created', orderId: order.id });

if (everythingOk) {
  await tx.commit();   // every emit fires
} else {
  tx.rollback();       // nothing fired
}

Useful for "either all of these events fire or none" semantics within a single request scope.

Decorators

@OnEvent(options)

@OnEvent({
  event:         'user.*',
  async:         true,
  priority:      10,
  timeout:       5_000,
  filter:        (data) => data.tenantId === 'enterprise',
  transform:     (data) => normalizeUser(data),
  errorBoundary: true,
  onError:       (err, data) => log.warn('handler failed', { err, data }),
})
async onUserChange(data: UserEvent) { /* … */ }

@OnceEvent(options) — fire once then unsubscribe

@OnceEvent({ event: 'app.ready' })
async warmCaches() { /* … */ }

@OnAnyEvent() — wildcard catch-all

@EmitEvent(options) — emit when a method runs

@EmitEvent({
  event:     'user.created',
  mapResult: (user) => ({ id: user.id, email: user.email }),
  mapError:  (err)  => ({ reason: err.message }),
})
async create(input: CreateInput) { /* … */ }

The interceptor runs after the method: on success it emits <event>.success (payload = mapResult(result) if provided, else the raw result); on throw it emits <event>.error (payload = mapError(error) if provided, else the error) and re-throws. Options are event, mapResult?, mapError? only.

Two things about it surprise people, and both are silent.

Subscribers must use the suffixed names. event: 'user.created' never emits user.created; subscribing to that bare name receives nothing.

The emitter is found by convention, not injected. The decorator reads this.__eventEmitter__, then this.eventEmitter. On an instance with neither it runs the method and emits nothing — no warning, no error. A service that stores the emitter as events, for example, will not emit through this decorator. Expose it as eventEmitter or __eventEmitter__; the property name is part of the contract.

@ScheduleEvent(options)

@ScheduleEvent({ event: 'user.welcome-followup', delay: 24 * 60 * 60 * 1000 })   // 24h delay
async welcomeFollowup() { /* … */ }

@ScheduleEvent({ event: 'reports.daily', cron: '0 9 * * *' })
async dailyReport() { /* … */ }

Options: event (required), plus one of delay (ms), cron, or at (a Date).

@BatchEvents(options)

Coalesce many emits into one delivery — useful when a tight loop would otherwise fire thousands of identical events. Options: event, maxSize, and maxWait (ms) are all required.

@OnModuleEvent({ event, filter? }) — listen to framework module events

@EventEmitter(namespace?) — class-level emitter binding

Wildcard semantics

With wildcard: true:

@OnEvent({ event: 'user.*' })       // user.created, user.deleted (one segment)
@OnEvent({ event: 'user.**' })      // any nested suffix
@OnEvent({ event: 'order.*.shipped' })  // order.{id}.shipped

Pattern resolution is cached, so wildcards are O(1) per emit at runtime.

Other services exposed

EventValidationService.registerSchema(name, schema) accepts Zod schemas, JSON Schema and custom ISchemaValidator implementations. Pass Zod objects directly; the service adapts their parse results to its validation contract. validateData(name, payload) returns a boolean; validate(name, payload) returns the validation result, including field errors.

Lifecycle

EventsService implements onInit, onStart, onStop, onDestroy. The module honours them via the standard Titan lifecycle.

flowchart LR
  Producer[Service.emit] --> Bus[EventsService]
  Bus --> Pattern{matching?}
  Pattern -- exact --> H1[Handler A]
  Pattern -- wildcard --> H2[Handler B]
  Pattern -- catch-all --> H3[Handler C]
  Bus --> History[EventHistoryService<br/>if history.enabled]
  Bus --> Metrics[Stats]
  Bus --> Sched[EventSchedulerService<br/>for delayed/cron]

Anti-patterns

• Treating in-process events as durable. They aren't. Restart → in-flight handlers gone. For at-least-once delivery, use titan-notifications over Rotif.

• Heavy work in sync emit. Sync emit blocks the publisher. Use emitAsync for I/O-bound handlers; reserve sync for metrics/audit-style side effects.

• Schema-less events at scale. A typo in an event name silently drops the listener. Use the schemas option for production- critical events.

• subscribeAll for analytics. Captures every event in the process — easy to leave running and blow the listener budget. Prefer scoped patterns.

See also

• titan-notifications — for cross-process durable delivery

• titan-scheduler — for full cron/interval semantics (events module has lighter-weight scheduled emission)