React to a saved note

Start with one event

Complete Installation and copy the example. It needs no database or worker. The trace represents reactions; it does not actually save or index a note.

In an app, register listeners during boot on app().events. Dispatch only once the operation the event describes has succeeded.

• Installation

mkdir -p examples
cp -R node_modules/@db3.ai/app/src/events/examples/. examples/

Run, fail and repair a dispatch

Expect the trace index:one, first:one, index:two, after:four, with rejected: true and hasListeners: false.

The failed third dispatch never reaches the later listener. Removing the failing registration allows the fourth dispatch to complete. All remaining listeners are cleared in finally.

npx tsx examples/runNoteEvents.ts

Use a class as the event identity

Name the event after something that happened and carry stable IDs. Event instances are not cloned. Readonly fields help communicate intent, but do not deeply freeze nested values.

The exact class constructor is the subscription key. A subclass is a different event type. Plain objects are rejected rather than sharing one ambiguous Object key.

/** Immutable notification that one application note has already been saved. */
export class NoteSaved {
	/**
	 * Carries identifiers rather than a request, model connection or mutable service.
	 * @param noteId - Saved note identity.
	 * @param ownerId - Authorized owner identity determined by the producer.
	 */
	constructor(readonly noteId: string, readonly ownerId: string) {}
}

Subscribe and await delivery

listen(EventClass, listener) returns an idempotent unsubscribe function. once() removes its registration before invocation so overlapping dispatches cannot run that listener twice.

dispatch(new Event(...)) snapshots current listeners and awaits them sequentially in registration order. Removing a registration does not cancel a callback already captured by an in-progress dispatch.

import { pathToFileURL } from 'node:url';
import { Events } from '@db3.ai/app/events';
import { NoteSaved } from './NoteSaved';

/**
 * Exercises ordered listeners, one-shot observation, failure and explicit repair.
 * @returns The dispatch trace and teardown state; no durable side effect is made.
 */
export async function runNoteEvents() {
	const events = new Events();
	const trace: string[] = [];
	try {
		const unsubscribe = events.listen(NoteSaved, async event => { await Promise.resolve(); trace.push(`index:${event.noteId}`); });
		events.once(NoteSaved, event => { trace.push(`first:${event.noteId}`); });
		await events.dispatch(new NoteSaved('one', 'ada'));
		await events.dispatch(new NoteSaved('two', 'ada'));
		unsubscribe();
		const removeFailure = events.listen(NoteSaved, () => { throw new Error('Reaction failed'); });
		events.listen(NoteSaved, event => { trace.push(`after:${event.noteId}`); });
		let rejected = false;
		try { await events.dispatch(new NoteSaved('three', 'ada')); } catch { rejected = true; }
		removeFailure();
		await events.dispatch(new NoteSaved('four', 'ada'));
		events.forget(NoteSaved);
		return { trace, rejected, hasListeners: events.hasListeners(NoteSaved) };
	} finally {
		events.clear();
	}
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) console.log(JSON.stringify(await runNoteEvents(), null, 2));

Decide whether a reaction must succeed

The first thrown/rejected listener stops dispatch and propagates to the caller. Earlier listeners may already have changed state; there is no rollback of their effects.

Catch only when you intentionally make a reaction non-fatal. Do not replay a whole event blindly: earlier successful side effects might run twice. This lab repairs registration and dispatches a new event; it does not claim automatic retry or duplicate protection.

Use Queue for durable work

Events are in memory, process-local and not persisted. They do not cross API, worker or scheduler processes, survive a crash or retry failed listeners.

A listener can enqueue a job, but a database commit followed by dispatch is not an atomic outbox. If losing the handoff matters, design that transaction/reconciliation explicitly.

Queue lifecycle observers are a different stream: their errors are isolated from already-completed queue transitions. Application Events intentionally propagates listener failures. Logs are observability, not a way to trigger product behavior.

• Durable queued reports
• Structured logging

Remove listeners at their owner boundary

Use the returned unsubscribe function for one registration, forget(EventClass) for one event type, and clear() for the whole dispatcher. Do not clear a shared app dispatcher after each request.

App.close() clears its dispatcher. A standalone dispatcher owns no network resources, but you should still remove listeners when its owner finishes.

Testing

Create a tests/events directory and save the test below as runNoteEvents.test.ts. It imports the example you copied into examples/. Keep the same folder layout so the relative import resolves.

Run from the application root with the development dependencies from Installation. These are consumer tests, not commands that assume a framework checkout.

import { expect, it } from 'vitest';
import { Events } from '@db3.ai/app/events';
import { NoteSaved } from '../../examples/NoteSaved';
import { runNoteEvents } from '../../examples/runNoteEvents';

it('awaits ordered listeners, invokes once once, propagates failure and recovers after removal', async () => {
	expect(await runNoteEvents()).toEqual({ trace: ['index:one', 'first:one', 'index:two', 'after:four'], rejected: true, hasListeners: false });
});

it('dispatches exact classes only and rejects anonymous payloads', async () => {
	const events = new Events();
	/** Distinct event identity, not an implicit subscription to the base event. */
	class ImportedNoteSaved extends NoteSaved {}
	let calls = 0;
	const stop = events.listen(NoteSaved, () => { calls++; });
	try {
		await events.dispatch(new ImportedNoteSaved('one', 'ada'));
		expect(calls).toBe(0);
		await expect(events.dispatch({ noteId: 'one' })).rejects.toThrow('class instances');
		await events.dispatch(new NoteSaved('one', 'ada'));
		expect(calls).toBe(1);
		stop();
		stop();
		expect(events.hasListeners(NoteSaved)).toBe(false);
	} finally { events.clear(); }
});

Test ordering and failure

The copied tests use the real dispatcher and assert the complete trace, failure propagation, one-shot observation, exact-class matching, plain-object rejection and idempotent unsubscribe.

Add a second application listener and prove its position in the trace. If it creates durable work, test the Queue handoff separately.

npx vitest run tests/events/runNoteEvents.test.ts
npx tsc --noEmit --target ES2022 --module ESNext --moduleResolution Bundler --types node --skipLibCheck examples/*.ts

Coverage and limits

Taught/tested: every Events method, async order, listener failure/recovery, exact-class identity and cleanup. App shutdown and overlapping one-shot dispatch also have existing service tests.

Cross-process delivery, replay, retries, cancellation and transaction coupling are not Events features. The reference contains the service and every public event contract.

• Full Events API