21875627364

Committed 10 Feb 2026 05:34PM UTC coverage: 84.658% (-9.7%) from 94.396%

Build # 21875627364

Build Type

Pull #28

github

Committed by

web-flow

Commit Message

Merge f3844c4de into 828e39903

Pull Request Pull Request #28: TypeScript and event dispatching pipeline refactoring

Run Details

626 of 953 branches covered (65.69%)

832 of 947 new or added lines in 65 files covered. (87.86%)

59 existing lines in 13 files now uncovered.

1225 of 1447 relevant lines covered (84.66%)

28.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.43

/src/EventStore.ts

import {
        type IAggregateSnapshotStorage,
        type IEvent,
        type IEventStorageReader,
        type IEventSet,
        type ILogger,
        type IMessageHandler,
        type IObservable,
        type IEventStream,
        type IEventStore,
        type EventQueryAfter,
        type EventQueryBefore,
        type Identifier,
        type IIdentifierProvider,
        type IEventDispatcher,
        type IEventBus,
        type IContainer,
        type AggregateEventsQueryParams,
        isIdentifierProvider,
        isIEventBus,
        isIEventStorageReader,
        isEventSet,
        isIObservableQueueProvider
} from './interfaces/index.ts';
import {
        getClassName,
        setupOneTimeEmitterSubscription
} from './utils/index.ts';
import { EventDispatcher } from './EventDispatcher.ts';

export class EventStore implements IEventStore {

        #identifierProvider: IIdentifierProvider;
        #eventStorageReader: IEventStorageReader;
        #snapshotStorage: IAggregateSnapshotStorage | undefined;
        eventBus: IEventBus;
        #eventDispatcher: IEventDispatcher;
        #sagaStarters: Set<string> = new Set();
        #logger?: ILogger;

        constructor({
                eventStorageReader,
                identifierProvider = isIdentifierProvider(eventStorageReader) ? eventStorageReader : undefined,
                snapshotStorage,
                eventBus,
                eventDispatcher,
                eventDispatchPipeline,
                eventDispatchPipelines,
                logger
        }: Pick<IContainer,
                'identifierProvider' |
                'eventStorageReader' |
                'snapshotStorage' |
                'eventBus' |
                'eventDispatcher' |
                'logger' |
                'eventDispatchPipeline' |
                'eventDispatchPipelines'
        >) {
                if (!eventStorageReader)
                        throw new TypeError('eventStorageReader argument required');
                if (!identifierProvider)
                        throw new TypeError('identifierProvider argument required');
                if (!isIEventStorageReader(eventStorageReader))
                        throw new TypeError('storage does not implement IEventStorage interface');
                if (eventBus && !isIEventBus(eventBus))
                        throw new TypeError('eventBus does not implement IMessageBus interface');

                this.#eventStorageReader = eventStorageReader;
                this.#identifierProvider = identifierProvider;
                this.#snapshotStorage = snapshotStorage;
                this.#eventDispatcher = eventDispatcher ?? new EventDispatcher({
                        eventBus,
                        eventDispatchPipeline,
                        eventDispatchPipelines
                });
                this.eventBus = eventBus ?? this.#eventDispatcher.eventBus;
                this.#logger = logger && 'child' in logger ?
                        logger.child({ service: getClassName(this) }) :
                        logger;
        }

        /**
         * Generates and returns a new unique identifier using the configured identifier provider.
         *
         * @returns A promise resolving to a unique identifier suitable for aggregates, sagas, and events.
         */
        async getNewId(): Promise<Identifier> {
                return this.#identifierProvider.getNewId();
        }

        async* getEventsByTypes(eventTypes: Readonly<string[]>, options?: EventQueryAfter): IEventStream {
                if (!Array.isArray(eventTypes))
                        throw new TypeError('eventTypes argument must be an Array');

                this.#logger?.debug(`retrieving ${eventTypes.join(', ')} events...`);

                const eventsIterable = await this.#eventStorageReader.getEventsByTypes(eventTypes, options);

                yield* eventsIterable;

                this.#logger?.debug(`${eventTypes.join(', ')} events retrieved`);
        }

        /** Retrieve all events of specific Aggregate */
        async* getAggregateEvents(aggregateId: Identifier, options?: AggregateEventsQueryParams): IEventStream {
                if (!aggregateId)
                        throw new TypeError('aggregateId argument required');

                this.#logger?.debug(`retrieving event stream for aggregate ${aggregateId}...`);

                // Get snapshot from snapshot storage if not provided in options
                let snapshot = options?.snapshot;
                if (!snapshot && this.#snapshotStorage)
                        snapshot = await this.#snapshotStorage.getAggregateSnapshot(aggregateId);

                if (snapshot)
                        yield snapshot;

                const eventsIterable = await this.#eventStorageReader.getAggregateEvents(aggregateId, {
                        ...options,
                        snapshot
                });

                yield* eventsIterable;

                this.#logger?.debug(`all events for aggregate ${aggregateId} retrieved`);
        }

        /** Retrieve events of specific Saga */
        async* getSagaEvents(sagaId: Identifier, filter: EventQueryBefore) {
                if (!sagaId)
                        throw new TypeError('sagaId argument required');
                if (!filter)
                        throw new TypeError('filter argument required');
                if (!filter.beforeEvent)
                        throw new TypeError('filter.beforeEvent argument required');
                if (filter.beforeEvent.sagaVersion === undefined)
                        throw new TypeError('filter.beforeEvent.sagaVersion argument required');

                this.#logger?.debug(`retrieving event stream for saga ${sagaId}, v${filter.beforeEvent.sagaVersion}...`);

                const eventsIterable = await this.#eventStorageReader.getSagaEvents(sagaId, filter);

                yield* eventsIterable;

                this.#logger?.debug(`all events for saga ${sagaId} retrieved`);
        }

        /**
         * Register event types that start sagas.
         * Upon such event commit a new sagaId will be assigned
         */
        registerSagaStarters(eventTypes: string[] = []) {
                for (const eventType of eventTypes)
                        this.#sagaStarters.add(eventType);
        }

        /**
         * Validate events, commit to storage and publish to messageBus, if needed
         *
         * @param events - a set of events to commit
         * @returns Signed and committed events
         */
        async dispatch(events: IEventSet): Promise<IEventSet> {
                if (!isEventSet(events) || events.length === 0)
                        throw new TypeError('dispatch requires a non-empty array of events');

                const augmentedEvents = await this.#attachSagaIdToSagaStarterEvents(events);

                return this.#eventDispatcher.dispatch(augmentedEvents, { origin: 'internal' });
        }

        /**
         * Generate and attach sagaId to events that start new sagas
         */
        async #attachSagaIdToSagaStarterEvents(events: IEventSet): Promise<IEventSet> {
                if (!this.#sagaStarters.size)
                        return events;

                const augmentedEvents: IEvent[] = [];
                for (const event of events) {
                        if (this.#sagaStarters.has(event.type)) {
                                if (event.sagaId)
                                        throw new Error(`Event "${event.type}" already contains sagaId. Multiple sagas with same event type are not supported`);

                                (event as IEvent).sagaId = await this.getNewId();
                                (event as IEvent).sagaVersion = 0;

                                augmentedEvents.push(event);
                        }
                        else {
                                augmentedEvents.push(event);
                        }
                }
                return augmentedEvents;
        }

        on(messageType: string, handler: IMessageHandler) {
                this.eventBus.on(messageType, handler);
        }

        off(messageType: string, handler: IMessageHandler) {
                this.eventBus.off(messageType, handler);
        }

        queue(name: string): IObservable {
                if (!isIObservableQueueProvider(this.eventBus))
                        throw new Error('Injected eventBus does not support named queues');

                return this.eventBus.queue(name);
        }

        /** Creates one-time subscription for one or multiple events that match a filter */
        once(messageTypes: string | string[], handler?: IMessageHandler, filter?: (e: IEvent) => boolean): Promise<IEvent> {
                const subscribeTo = Array.isArray(messageTypes) ? messageTypes : [messageTypes];

                return setupOneTimeEmitterSubscription(this.eventBus, subscribeTo, filter, handler, this.#logger);
        }
}

1	import {	22✔
2	type IAggregateSnapshotStorage,
3	type IEvent,
4	type IEventStorageReader,
5	type IEventSet,
6	type ILogger,
7	type IMessageHandler,
8	type IObservable,
9	type IEventStream,
10	type IEventStore,
11	type EventQueryAfter,
12	type EventQueryBefore,
13	type Identifier,
14	type IIdentifierProvider,
15	type IEventDispatcher,
16	type IEventBus,
17	type IContainer,
18	type AggregateEventsQueryParams,
19	isIdentifierProvider,
20	isIEventBus,
21	isIEventStorageReader,
22	isEventSet,
23	isIObservableQueueProvider
24	} from './interfaces/index.ts';
25	import {	22✔
26	getClassName,
27	setupOneTimeEmitterSubscription
28	} from './utils/index.ts';
29	import { EventDispatcher } from './EventDispatcher.ts';	22✔
30
31	export class EventStore implements IEventStore {	22✔
32
33	#identifierProvider: IIdentifierProvider;
34	#eventStorageReader: IEventStorageReader;
35	#snapshotStorage: IAggregateSnapshotStorage \| undefined;
36	eventBus: IEventBus;
37	#eventDispatcher: IEventDispatcher;
38	#sagaStarters: Set<string> = new Set();	58✔
39	#logger?: ILogger;
40
41	constructor({
42	eventStorageReader,
43	identifierProvider = isIdentifierProvider(eventStorageReader) ? eventStorageReader : undefined,	×
44	snapshotStorage,
45	eventBus,
46	eventDispatcher,
47	eventDispatchPipeline,
48	eventDispatchPipelines,
49	logger
50	}: Pick<IContainer,
51	'identifierProvider' \|
52	'eventStorageReader' \|
53	'snapshotStorage' \|
54	'eventBus' \|
55	'eventDispatcher' \|
56	'logger' \|
57	'eventDispatchPipeline' \|
58	'eventDispatchPipelines'
59	>) {
60	if (!eventStorageReader)	58!
NEW 61	throw new TypeError('eventStorageReader argument required');	×
62	if (!identifierProvider)	58!
NEW 63	throw new TypeError('identifierProvider argument required');	×
64	if (!isIEventStorageReader(eventStorageReader))	58!
UNCOV 65	throw new TypeError('storage does not implement IEventStorage interface');	×
66	if (eventBus && !isIEventBus(eventBus))	58!
NEW 67	throw new TypeError('eventBus does not implement IMessageBus interface');	×
68
69	this.#eventStorageReader = eventStorageReader;	58✔
70	this.#identifierProvider = identifierProvider;	58✔
71	this.#snapshotStorage = snapshotStorage;	58✔
72	this.#eventDispatcher = eventDispatcher ?? new EventDispatcher({	58!
73	eventBus,
74	eventDispatchPipeline,
75	eventDispatchPipelines
76	});
77	this.eventBus = eventBus ?? this.#eventDispatcher.eventBus;	58!
78	this.#logger = logger && 'child' in logger ?	58!
79	logger.child({ service: getClassName(this) }) :
80	logger;
81	}
82
83	/**
84	* Generates and returns a new unique identifier using the configured identifier provider.
85	*
86	* @returns A promise resolving to a unique identifier suitable for aggregates, sagas, and events.
87	*/
88	async getNewId(): Promise<Identifier> {
89	return this.#identifierProvider.getNewId();	24✔
90	}
91
92	async* getEventsByTypes(eventTypes: Readonly<string[]>, options?: EventQueryAfter): IEventStream {
93	if (!Array.isArray(eventTypes))	4!
NEW 94	throw new TypeError('eventTypes argument must be an Array');	×
95
96	this.#logger?.debug(`retrieving ${eventTypes.join(', ')} events...`);	4✔
97
98	const eventsIterable = await this.#eventStorageReader.getEventsByTypes(eventTypes, options);	4✔
99
100	yield* eventsIterable;	4✔
101
102	this.#logger?.debug(`${eventTypes.join(', ')} events retrieved`);	2✔
103	}
104
105	/** Retrieve all events of specific Aggregate */
106	async* getAggregateEvents(aggregateId: Identifier, options?: AggregateEventsQueryParams): IEventStream {
107	if (!aggregateId)	22!
UNCOV 108	throw new TypeError('aggregateId argument required');	×
109
110	this.#logger?.debug(`retrieving event stream for aggregate ${aggregateId}...`);	22✔
111
112	// Get snapshot from snapshot storage if not provided in options
113	let snapshot = options?.snapshot;	22✔
114	if (!snapshot && this.#snapshotStorage)	22✔
115	snapshot = await this.#snapshotStorage.getAggregateSnapshot(aggregateId);	22✔
116
117	if (snapshot)	22✔
118	yield snapshot;	2✔
119
120	const eventsIterable = await this.#eventStorageReader.getAggregateEvents(aggregateId, {	22✔
121	...options,
122	snapshot
123	});
124
125	yield* eventsIterable;	22✔
126
127	this.#logger?.debug(`all events for aggregate ${aggregateId} retrieved`);	22✔
128	}
129
130	/** Retrieve events of specific Saga */
131	async* getSagaEvents(sagaId: Identifier, filter: EventQueryBefore) {
132	if (!sagaId)	4!
UNCOV 133	throw new TypeError('sagaId argument required');	×
134	if (!filter)	4!
UNCOV 135	throw new TypeError('filter argument required');	×
136	if (!filter.beforeEvent)	4!
UNCOV 137	throw new TypeError('filter.beforeEvent argument required');	×
138	if (filter.beforeEvent.sagaVersion === undefined)	4!
UNCOV 139	throw new TypeError('filter.beforeEvent.sagaVersion argument required');	×
140
141	this.#logger?.debug(`retrieving event stream for saga ${sagaId}, v${filter.beforeEvent.sagaVersion}...`);	4✔
142
143	const eventsIterable = await this.#eventStorageReader.getSagaEvents(sagaId, filter);	4✔
144
145	yield* eventsIterable;	4✔
146
147	this.#logger?.debug(`all events for saga ${sagaId} retrieved`);	4✔
148	}
149
150	/**
151	* Register event types that start sagas.
152	* Upon such event commit a new sagaId will be assigned
153	*/
154	registerSagaStarters(eventTypes: string[] = []) {	3✔
155	for (const eventType of eventTypes)	10✔
156	this.#sagaStarters.add(eventType);	4✔
157	}
158
159	/**
160	* Validate events, commit to storage and publish to messageBus, if needed
161	*
162	* @param events - a set of events to commit
163	* @returns Signed and committed events
164	*/
165	async dispatch(events: IEventSet): Promise<IEventSet> {
166	if (!isEventSet(events) \|\| events.length === 0)	48✔
167	throw new TypeError('dispatch requires a non-empty array of events');	2✔
168
169	const augmentedEvents = await this.#attachSagaIdToSagaStarterEvents(events);	46✔
170
171	return this.#eventDispatcher.dispatch(augmentedEvents, { origin: 'internal' });	46✔
172	}
173
174	/**
175	* Generate and attach sagaId to events that start new sagas
176	*/
177	async #attachSagaIdToSagaStarterEvents(events: IEventSet): Promise<IEventSet> {
178	if (!this.#sagaStarters.size)	46✔
179	return events;	42✔
180
181	const augmentedEvents: IEvent[] = [];	4✔
182	for (const event of events) {	4✔
183	if (this.#sagaStarters.has(event.type)) {	4!
184	if (event.sagaId)	4!
UNCOV 185	throw new Error(`Event "${event.type}" already contains sagaId. Multiple sagas with same event type are not supported`);	×
186
187	(event as IEvent).sagaId = await this.getNewId();	4✔
188	(event as IEvent).sagaVersion = 0;	4✔
189
190	augmentedEvents.push(event);	4✔
191	}
192	else {
193	augmentedEvents.push(event);	×
194	}
195	}
196	return augmentedEvents;	4✔
197	}
198
199	on(messageType: string, handler: IMessageHandler) {
200	this.eventBus.on(messageType, handler);	4✔
201	}
202
203	off(messageType: string, handler: IMessageHandler) {
204	this.eventBus.off(messageType, handler);	2✔
205	}
206
207	queue(name: string): IObservable {
208	if (!isIObservableQueueProvider(this.eventBus))	4!
NEW 209	throw new Error('Injected eventBus does not support named queues');	×
210
211	return this.eventBus.queue(name);	4✔
212	}
213
214	/** Creates one-time subscription for one or multiple events that match a filter */
215	once(messageTypes: string \| string[], handler?: IMessageHandler, filter?: (e: IEvent) => boolean): Promise<IEvent> {
216	const subscribeTo = Array.isArray(messageTypes) ? messageTypes : [messageTypes];	4!
217
218	return setupOneTimeEmitterSubscription(this.eventBus, subscribeTo, filter, handler, this.#logger);	4✔
219	}
220	}

snatalenko / node-cqrs / 21875627364

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous