14431053423

Committed 13 Apr 2025 03:59PM UTC coverage: 83.069% (+0.7%) from 82.347%

Build # 14431053423

Build Type

push

github

Committed by

snatalenko

Commit Message

1.0.0-rc.8

Run Details

490 of 782 branches covered (62.66%)

996 of 1199 relevant lines covered (83.07%)

21.49 hits per line

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

79.66

/src/AbstractProjection.ts

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

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

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;
        #eventLocker?: IEventLocker;
        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 | undefined {
                return this.#viewLocker ?? (isViewLocker(this.view) ? this.view : undefined);
        }

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

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

        protected set _eventLocker(value: IEventLocker | undefined) {
                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 */
        async subscribe(eventStore: IEventStore): Promise<void> {
                subscribe(eventStore, this, {
                        masterHandler: (e: IEvent) => this.project(e)
                });

                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');
                        await this._viewLocker.once('ready');
                }

                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 projection view from event store */
        async restore(eventStore: IEventStore): Promise<void> {
                // lock the view to ensure same restoring procedure
                // won't be performed by another projection instance
                if (this._viewLocker)
                        await this._viewLocker.lock();

                await this._restore(eventStore);

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

        /** Restore projection view from event store */
        protected async _restore(eventStore: IEventStore): 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: any) {
                                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: Error, event: IEvent) {
                this._logger?.error(`view restoring has failed (view will remain locked): ${error.message}`, {
                        service: getClassName(this),
                        event,
                        stack: error.stack
                });
                throw error;
        }
}

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

snatalenko / node-cqrs / 14431053423

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