21643388185

Committed 03 Feb 2026 06:52PM UTC coverage: 75.749% (-18.6%) from 94.396%

Build # 21643388185

Build Type

Pull #28

github

Committed by

web-flow

Commit Message

Merge 8c7b95a7f into 828e39903

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

Run Details

522 of 939 branches covered (55.59%)

696 of 932 new or added lines in 65 files covered. (74.68%)

59 existing lines in 13 files now uncovered.

1087 of 1435 relevant lines covered (75.75%)

25.48 hits per line

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

82.81

/src/AbstractProjection.ts

import { describe } from './Event.ts';
import { InMemoryView } from './in-memory/InMemoryView.ts';
import {
        type IViewLocker,
        type IEventLocker,
        type IProjection,
        type ILogger,
        type IExtendableLogger,
        type IEvent,
        type IObservable,
        type IEventStorageReader,
        isViewLocker,
        isEventLocker
} from './interfaces/index.ts';

import {
        getClassName,
        validateHandlers,
        getHandler,
        subscribe,
        getMessageHandlerNames
} from './utils/index.ts';

export type AbstractProjectionParams<T> = {

        /**
         * The default view associated with the projection.
         * Can optionally implement IViewLocker and/or IEventLocker.
         */
        view?: T,

        /**
         * Manages view restoration state to prevent early access to an inconsistent view
         * or conflicts from concurrent restoration by other processes.
         */
        viewLocker?: IViewLocker,

        /**
         * Tracks event processing state to prevent concurrent handling by multiple processes.
         */
        eventLocker?: IEventLocker,

        logger?: ILogger | IExtendableLogger
}

/**
 * Base class for Projection definition
 */
export abstract class AbstractProjection<TView = any> implements IProjection<TView> {

        /**
         * List of event types handled by the projection. Can be overridden in the projection implementation.
         * If not overridden, event types will be inferred from handler methods defined on the Projection class.
         */
        static get handles(): string[] {
                return getMessageHandlerNames(this);
        }

        #view?: TView;
        #viewLocker?: IViewLocker | null;
        #eventLocker?: IEventLocker | null;
        protected _logger?: ILogger;

        /**
         * The default view associated with the projection.
         * Can optionally implement IViewLocker and/or IEventLocker.
         */
        public get view(): TView {
                return this.#view ?? (this.#view = new InMemoryView() as TView);
        }

        protected set view(value: TView) {
                this.#view = value;
        }

        /**
         * Manages view restoration state to prevent early access to an inconsistent view
         * or conflicts from concurrent restoration by other processes.
         */
        protected get _viewLocker(): IViewLocker | null {
                if (this.#viewLocker === undefined)
                        this.#viewLocker = isViewLocker(this.view) ? this.view : null;

                return this.#viewLocker;
        }

        protected set _viewLocker(value: IViewLocker | undefined | null) {
                this.#viewLocker = value;
        }

        /**
         * Tracks event processing state to prevent concurrent handling by multiple processes.
         */
        protected get _eventLocker(): IEventLocker | null {
                if (this.#eventLocker === undefined)
                        this.#eventLocker = isEventLocker(this.view) ? this.view : null;

                return this.#eventLocker;
        }

        protected set _eventLocker(value: IEventLocker | undefined | null) {
                this.#eventLocker = value;
        }

        constructor({
                view,
                viewLocker,
                eventLocker,
                logger
        }: AbstractProjectionParams<TView> = {}) {
                validateHandlers(this);

                this.#view = view;
                this.#viewLocker = viewLocker;
                this.#eventLocker = eventLocker;

                this._logger = logger && 'child' in logger ?
                        logger.child({ service: getClassName(this) }) :
                        logger;
        }

        /**
         * Subscribe to event store
         * and restore view state from not yet projected events
         */
        async subscribe(eventStore: IObservable & IEventStorageReader): Promise<void> {
                subscribe(eventStore, this, {
                        masterHandler: this.project
                });

                await this.restore(eventStore);
        }

        /** Pass event to projection event handler */
        async project(event: IEvent): Promise<void> {
                if (this._viewLocker && !this._viewLocker.ready) {
                        this._logger?.debug(`view is locked, awaiting until it is ready to process ${describe(event)}`);
                        await this._viewLocker.once('ready');
                        this._logger?.debug(`view is ready, processing ${describe(event)}`);
                }

                return this._project(event);
        }

        /** Pass event to projection event handler, without awaiting for restore operation to complete */
        protected async _project(event: IEvent): Promise<void> {
                const handler = getHandler(this, event.type);
                if (!handler)
                        throw new Error(`'${event.type}' handler is not defined or not a function`);

                if (this._eventLocker) {
                        const eventLockObtained = await this._eventLocker.tryMarkAsProjecting(event);
                        if (!eventLockObtained)
                                return;
                }

                await handler.call(this, event);

                if (this._eventLocker)
                        await this._eventLocker.markAsProjected(event);
        }

        /**
         * Restore view state from not-yet-projected events.
         *
         * Lock the view to ensure same restoring procedure
         * won't be performed by another projection instance.
         * */
        async restore(eventStore: IEventStorageReader): Promise<void> {
                if (this._viewLocker)
                        await this._viewLocker.lock();

                await this._restore(eventStore);

                if (this._viewLocker)
                        this._viewLocker.unlock();
        }

        /** Restore view state from not-yet-projected events */
        protected async _restore(eventStore: IEventStorageReader): Promise<void> {
                if (!eventStore)
                        throw new TypeError('eventStore argument required');
                if (typeof eventStore.getEventsByTypes !== 'function')
                        throw new TypeError('eventStore.getEventsByTypes must be a Function');

                let lastEvent: IEvent | undefined;

                if (this._eventLocker) {
                        this._logger?.debug('retrieving last event projected');
                        lastEvent = await this._eventLocker.getLastEvent();
                }

                this._logger?.debug(`retrieving ${lastEvent ? `events after ${describe(lastEvent)}` : 'all events'}...`);

                const messageTypes = (this.constructor as typeof AbstractProjection).handles;
                const eventsIterable = eventStore.getEventsByTypes(messageTypes, { afterEvent: lastEvent });

                let eventsCount = 0;
                const startTs = Date.now();

                for await (const event of eventsIterable) {
                        try {
                                await this._project(event);
                                eventsCount += 1;
                        }
                        catch (err: unknown) {
                                this._onRestoringError(err, event);
                        }
                }

                this._logger?.info(`view restored from ${eventsCount} event(s) in ${Date.now() - startTs} ms`);
        }

        /**
         * Handle error on restoring.
         *
         * Logs and throws error by default
         */
        protected _onRestoringError(error: unknown, event: IEvent) {
                const errorMessage = error instanceof Error ? error.message : String(error);
                this._logger?.error(`view restoring has failed (view remains locked): ${errorMessage}`, {
                        service: getClassName(this),
                        event,
                        error
                });

                throw error;
        }
}

1	import { describe } from './Event.ts';	30✔
2	import { InMemoryView } from './in-memory/InMemoryView.ts';	30✔
3	import {	30✔
4	type IViewLocker,
5	type IEventLocker,
6	type IProjection,
7	type ILogger,
8	type IExtendableLogger,
9	type IEvent,
10	type IObservable,
11	type IEventStorageReader,
12	isViewLocker,
13	isEventLocker
14	} from './interfaces/index.ts';
15
16	import {	30✔
17	getClassName,
18	validateHandlers,
19	getHandler,
20	subscribe,
21	getMessageHandlerNames
22	} from './utils/index.ts';
23
24	export type AbstractProjectionParams<T> = {
25
26	/**
27	* The default view associated with the projection.
28	* Can optionally implement IViewLocker and/or IEventLocker.
29	*/
30	view?: T,
31
32	/**
33	* Manages view restoration state to prevent early access to an inconsistent view
34	* or conflicts from concurrent restoration by other processes.
35	*/
36	viewLocker?: IViewLocker,
37
38	/**
39	* Tracks event processing state to prevent concurrent handling by multiple processes.
40	*/
41	eventLocker?: IEventLocker,
42
43	logger?: ILogger \| IExtendableLogger
44	}
45
46	/**
47	* Base class for Projection definition
48	*/
49	export abstract class AbstractProjection<TView = any> implements IProjection<TView> {	30✔
50
51	/**
52	* List of event types handled by the projection. Can be overridden in the projection implementation.
53	* If not overridden, event types will be inferred from handler methods defined on the Projection class.
54	*/
55	static get handles(): string[] {
56	return getMessageHandlerNames(this);	16✔
57	}
58
59	#view?: TView;
60	#viewLocker?: IViewLocker \| null;
61	#eventLocker?: IEventLocker \| null;
62	protected _logger?: ILogger;
63
64	/**
65	* The default view associated with the projection.
66	* Can optionally implement IViewLocker and/or IEventLocker.
67	*/
68	public get view(): TView {
69	return this.#view ?? (this.#view = new InMemoryView() as TView);	102✔
70	}
71
72	protected set view(value: TView) {
NEW 73	this.#view = value;	×
74	}
75
76	/**
77	* Manages view restoration state to prevent early access to an inconsistent view
78	* or conflicts from concurrent restoration by other processes.
79	*/
80	protected get _viewLocker(): IViewLocker \| null {
81	if (this.#viewLocker === undefined)	82✔
82	this.#viewLocker = isViewLocker(this.view) ? this.view : null;	20!
83
84	return this.#viewLocker;	82✔
85	}
86
87	protected set _viewLocker(value: IViewLocker \| undefined \| null) {
NEW 88	this.#viewLocker = value;	×
89	}
90
91	/**
92	* Tracks event processing state to prevent concurrent handling by multiple processes.
93	*/
94	protected get _eventLocker(): IEventLocker \| null {
95	if (this.#eventLocker === undefined)	80✔
96	this.#eventLocker = isEventLocker(this.view) ? this.view : null;	22!
97
98	return this.#eventLocker;	80✔
99	}
100
101	protected set _eventLocker(value: IEventLocker \| undefined \| null) {
NEW 102	this.#eventLocker = value;	×
103	}
104
105	constructor({	3✔
106	view,
107	viewLocker,
108	eventLocker,
109	logger
110	}: AbstractProjectionParams<TView> = {}) {
111	validateHandlers(this);	30✔
112
113	this.#view = view;	30✔
114	this.#viewLocker = viewLocker;	30✔
115	this.#eventLocker = eventLocker;	30✔
116
117	this._logger = logger && 'child' in logger ?	30!
118	logger.child({ service: getClassName(this) }) :
119	logger;
120	}
121
122	/**
123	* Subscribe to event store
124	* and restore view state from not yet projected events
125	*/
126	async subscribe(eventStore: IObservable & IEventStorageReader): Promise<void> {
127	subscribe(eventStore, this, {	8✔
128	masterHandler: this.project
129	});
130
131	await this.restore(eventStore);	8✔
132	}
133
134	/** Pass event to projection event handler */
135	async project(event: IEvent): Promise<void> {
136	if (this._viewLocker && !this._viewLocker.ready) {	4✔
137	this._logger?.debug(`view is locked, awaiting until it is ready to process ${describe(event)}`);	2✔
138	await this._viewLocker.once('ready');	2✔
139	this._logger?.debug(`view is ready, processing ${describe(event)}`);	2✔
140	}
141
142	return this._project(event);	4✔
143	}
144
145	/** Pass event to projection event handler, without awaiting for restore operation to complete */
146	protected async _project(event: IEvent): Promise<void> {
147	const handler = getHandler(this, event.type);	32✔
148	if (!handler)	32✔
149	throw new Error(`'${event.type}' handler is not defined or not a function`);	2✔
150
151	if (this._eventLocker) {	30!
NEW 152	const eventLockObtained = await this._eventLocker.tryMarkAsProjecting(event);	×
NEW 153	if (!eventLockObtained)	×
NEW 154	return;	×
155	}
156
157	await handler.call(this, event);	30✔
158
159	if (this._eventLocker)	30!
NEW 160	await this._eventLocker.markAsProjected(event);	×
161	}
162
163	/**
164	* Restore view state from not-yet-projected events.
165	*
166	* Lock the view to ensure same restoring procedure
167	* won't be performed by another projection instance.
168	* */
169	async restore(eventStore: IEventStorageReader): Promise<void> {
170	if (this._viewLocker)	20✔
171	await this._viewLocker.lock();	20✔
172
173	await this._restore(eventStore);	20✔
174
175	if (this._viewLocker)	16✔
176	this._viewLocker.unlock();	16✔
177	}
178
179	/** Restore view state from not-yet-projected events */
180	protected async _restore(eventStore: IEventStorageReader): Promise<void> {
181	if (!eventStore)	20!
UNCOV 182	throw new TypeError('eventStore argument required');	×
183	if (typeof eventStore.getEventsByTypes !== 'function')	20!
NEW 184	throw new TypeError('eventStore.getEventsByTypes must be a Function');	×
185
186	let lastEvent: IEvent \| undefined;
187
188	if (this._eventLocker) {	20!
NEW 189	this._logger?.debug('retrieving last event projected');	×
NEW 190	lastEvent = await this._eventLocker.getLastEvent();	×
191	}
192
193	this._logger?.debug(`retrieving ${lastEvent ? `events after ${describe(lastEvent)}` : 'all events'}...`);	20!
194
195	const messageTypes = (this.constructor as typeof AbstractProjection).handles;	20✔
196	const eventsIterable = eventStore.getEventsByTypes(messageTypes, { afterEvent: lastEvent });	20✔
197
198	let eventsCount = 0;	20✔
199	const startTs = Date.now();	20✔
200
201	for await (const event of eventsIterable) {	20✔
202	try {	26✔
203	await this._project(event);	26✔
204	eventsCount += 1;	24✔
205	}
206	catch (err: unknown) {
207	this._onRestoringError(err, event);	2✔
208	}
209	}
210
211	this._logger?.info(`view restored from ${eventsCount} event(s) in ${Date.now() - startTs} ms`);	16✔
212	}
213
214	/**
215	* Handle error on restoring.
216	*
217	* Logs and throws error by default
218	*/
219	protected _onRestoringError(error: unknown, event: IEvent) {
220	const errorMessage = error instanceof Error ? error.message : String(error);	2!
221	this._logger?.error(`view restoring has failed (view remains locked): ${errorMessage}`, {	2✔
222	service: getClassName(this),
223	event,
224	error
225	});
226
227	throw error;	2✔
228	}
229	}

snatalenko / node-cqrs / 21643388185

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