signaltree-events

Using @signaltree/events

Use when an app needs typed event contracts with runtime validation, stable IDs/correlation, idempotency, and retry semantics across a process boundary (queue, WebSocket, service bus). Skip for local in-process emitters.

Three layers, composable independently:

1. Schema + validation — createEventSchema, parseEvent (THROWS), safeParseEvent (returns a result), isValidEvent (type guard). Names match Zod's parse/safeParse; before 14.1.1 they were inverted.
2. Registry + factory — EventRegistry, createEventFactory
3. Transport adapters (subpath) — /nestjs (BullMQ/Redis), /angular (WebSocket bridge), /testing (MockEventBus)

Install:

npm install @signaltree/events zod

Zod is required at runtime even without authoring schemas. Peer: zod ^3 || ^4 required. Optional peers (install only what you use): @angular/core ^18+, rxjs ^7, @nestjs/common ^10||^11, bullmq ^5, reflect-metadata ^0.1||^0.2. Angular range is ^18 (broader than rest of SignalTree). ESM-only — use dynamic import() in CommonJS consumers.

Define a schema:

import { createEventSchema, z } from '@signaltree/events';

// createEventSchema: type string + Zod raw shape (field map)
// createEventSchemaFromZod: type string + already-built ZodObject
export const UserCreatedSchema = createEventSchema('user.created', {
  id: z.string().uuid(),
  email: z.string().email(),
  name: z.string().min(1),
});
export type UserCreated = z.infer<typeof UserCreatedSchema>;

Validate:

import { createEventSchema, isValidEvent, parseEvent, safeParseEvent, z } from '@signaltree/events';

declare const incoming: unknown;

const UserCreatedSchema = createEventSchema('user.created', {
  id: z.string().uuid(),
  email: z.string().email(),
  name: z.string().min(1),
});

if (isValidEvent(UserCreatedSchema, incoming)) {
  /* typed */
}
const event = parseEvent(UserCreatedSchema, incoming); // throws on fail
const result = safeParseEvent(UserCreatedSchema, incoming); // never throws; result.success + result.error.issues

Factory (source and environment required; systemActor.type must be 'user'|'system'|'admin'|'webhook'):

import { createEventFactory } from '@signaltree/events';

const factory = createEventFactory<UserCreated>({
  source: 'api-gateway',
  environment: 'production',
  systemActor: { type: 'system', id: 'api-gateway' },
});
const created = factory.create('user.created', { id: crypto.randomUUID(), email: '[email protected]', name: 'Ada Lovelace' });

Retry + idempotency (defaultTtlMs, not ttlMs; check takes (event, consumer, options?)):

import { classifyError, isRetryableError, InMemoryIdempotencyStore, BaseEvent } from '@signaltree/events';

declare const incomingEvent: BaseEvent<string, unknown>;
declare function processEvent(e: BaseEvent<string, unknown>): Promise<void>;

try {
  await processEvent(incomingEvent);
} catch (err) {
  if (isRetryableError(err)) {
    /* enqueue retry */
  }
}

const store = new InMemoryIdempotencyStore({ defaultTtlMs: 60_000 });
const result = await store.check(incomingEvent, 'welcome-email-subscriber');
if (result.isDuplicate) return;

InMemoryIdempotencyStore is process-local — use Redis-backed store for multi-instance deployments.

NestJS (/nestjs subpath — requires @nestjs/common, bullmq, reflect-metadata):

import { EventBusModule } from '@signaltree/events/nestjs';

@Module({
  imports: [EventBusModule.forRoot({ redis: { host: 'localhost', port: 6379 }, preset: 'priority-based' })],
})
export class AppModule {}

Subscribers extend BaseSubscriber<T> with config: SubscriberConfig (name, eventTypes, priority, concurrency) and handle(event): Promise<ProcessingResult>.

Use forRootAsync when Redis config requires async loading.

Angular entityMap bridge (/angular subpath, v13+) — for apps holding entities in a @signaltree/core entityMap, don't hand-write a per-event upsertOne/updateOne/removeOne loop over an event batch; entityEventHandler maps the batch onto entityMap's own batch ops (one upsertMany/updateMany/removeMany call, not one per event):