21717407497

Committed 05 Feb 2026 03:26PM UTC coverage: 84.53% (-9.9%) from 94.396%

Build # 21717407497

Build Type

Pull #28

github

Committed by

web-flow

Commit Message

Merge 025edb883 into 828e39903

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

Run Details

611 of 939 branches covered (65.07%)

819 of 934 new or added lines in 65 files covered. (87.69%)

59 existing lines in 13 files now uncovered.

1213 of 1435 relevant lines covered (84.53%)

28.39 hits per line

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

90.63

/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';	32✔
2	import { InMemoryView } from './in-memory/InMemoryView.ts';	32✔
3	import {	32✔
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 {	32✔
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> {	32✔
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);	46✔
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);	114✔
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)	130✔
82	this.#viewLocker = isViewLocker(this.view) ? this.view : null;	32!
83
84	return this.#viewLocker;	130✔
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)	112✔
96	this.#eventLocker = isEventLocker(this.view) ? this.view : null;	34✔
97
98	return this.#eventLocker;	112✔
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);	48✔
112
113	this.#view = view;	48✔
114	this.#viewLocker = viewLocker;	48✔
115	this.#eventLocker = eventLocker;	48✔
116
117	this._logger = logger && 'child' in logger ?	48!
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) {	6✔
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);	6✔
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);	34✔
148	if (!handler)	34✔
149	throw new Error(`'${event.type}' handler is not defined or not a function`);	2✔
150
151	if (this._eventLocker) {	32✔
152	const eventLockObtained = await this._eventLocker.tryMarkAsProjecting(event);	2✔
153	if (!eventLockObtained)	2!
NEW 154	return;	×
155	}
156
157	await handler.call(this, event);	32✔
158
159	if (this._eventLocker)	32✔
160	await this._eventLocker.markAsProjected(event);	2✔
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)	32✔
171	await this._viewLocker.lock();	32✔
172
173	await this._restore(eventStore);	32✔
174
175	if (this._viewLocker)	26✔
176	this._viewLocker.unlock();	26✔
177	}
178
179	/** Restore view state from not-yet-projected events */
180	protected async _restore(eventStore: IEventStorageReader): Promise<void> {
181	if (!eventStore)	32!
UNCOV 182	throw new TypeError('eventStore argument required');	×
183	if (typeof eventStore.getEventsByTypes !== 'function')	32!
NEW 184	throw new TypeError('eventStore.getEventsByTypes must be a Function');	×
185
186	let lastEvent: IEvent \| undefined;
187
188	if (this._eventLocker) {	32✔
189	this._logger?.debug('retrieving last event projected');	12✔
190	lastEvent = await this._eventLocker.getLastEvent();	12✔
191	}
192
193	this._logger?.debug(`retrieving ${lastEvent ? `events after ${describe(lastEvent)}` : 'all events'}...`);	32!
194
195	const messageTypes = (this.constructor as typeof AbstractProjection).handles;	32✔
196	const eventsIterable = eventStore.getEventsByTypes(messageTypes, { afterEvent: lastEvent });	32✔
197
198	let eventsCount = 0;	32✔
199	const startTs = Date.now();	32✔
200
201	for await (const event of eventsIterable) {	32✔
202	try {	46✔
203	await this._project(event);	46✔
204	eventsCount += 1;	42✔
205	}
206	catch (err: unknown) {
207	this._onRestoringError(err, event);	4✔
208	}
209	}
210
211	this._logger?.info(`view restored from ${eventsCount} event(s) in ${Date.now() - startTs} ms`);	26✔
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);	4!
221	this._logger?.error(`view restoring has failed (view remains locked): ${errorMessage}`, {	4✔
222	service: getClassName(this),
223	event,
224	error
225	});
226
227	throw error;	4✔
228	}
229	}

snatalenko / node-cqrs / 21717407497

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