22745197368

Committed 06 Mar 2026 01:43AM UTC coverage: 95.287% (+0.9%) from 94.396%

Build # 22745197368

Build Type

Pull #28

github

Committed by

web-flow

Commit Message

Merge dd3952cc7 into 828e39903

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

Run Details

428 of 528 branches covered (81.06%)

1043 of 1091 new or added lines in 65 files covered. (95.6%)

3 existing lines in 2 files now uncovered.

1294 of 1358 relevant lines covered (95.29%)

31.11 hits per line

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

85.0

/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,
        assertFunction
} 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
         */
        subscribe(eventStore: IObservable): void {
                subscribe(eventStore, this, {
                        masterHandler: this.project
                });
        }

        /** 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> {
                assertFunction(eventStore?.getEventsByTypes, 'eventStore.getEventsByTypes');

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

snatalenko / node-cqrs / 22745197368

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