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

84.52

/src/AbstractAggregate.ts

import {
        IAggregate,
        IMutableAggregateState,
        ICommand,
        Identifier,
        IEvent,
        IEventSet,
        IAggregateConstructorParams
} from './interfaces';

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

/**
 * Deep-clone simple JS object
 */
function clone<T>(obj: T): T {
        return JSON.parse(JSON.stringify(obj));
}

const SNAPSHOT_EVENT_TYPE = 'snapshot';

/**
 * Base class for Aggregate definition
 */
export abstract class AbstractAggregate<TState extends IMutableAggregateState | object | void> implements IAggregate {

        /**
         * List of command names handled by the Aggregate.
         *
         * Can be overridden in the Aggregate implementation to explicitly define supported commands.
         * If not overridden, all public methods will be treated as command handlers by default.
         *
         * @example ['createUser', 'changePassword'];
         */
        static get handles(): string[] {
                return getMessageHandlerNames(this);
        }

        #id: Identifier;
        #changes: IEvent[] = [];
        #version: number = 0;
        #snapshotVersion: number | undefined;

        /** Internal aggregate state */
        protected state: TState | undefined;

        /** Command being handled by aggregate */
        protected command?: ICommand;

        /** Unique aggregate instance identifier */
        get id(): Identifier {
                return this.#id;
        }

        /** Aggregate instance version */
        get version(): number {
                return this.#version;
        }

        /** Restored snapshot version */
        get snapshotVersion(): number | undefined {
                return this.#snapshotVersion;
        }

        /** Events emitted by Aggregate */
        get changes(): IEventSet {
                return [...this.#changes];
        }

        /**
         * Override to define whether an aggregate state snapshot should be taken
         *
         * @example
         *         // create snapshot every 50 events
         *         return this.version % 50 === 0;
         */
        // eslint-disable-next-line class-methods-use-this
        get shouldTakeSnapshot(): boolean {
                return false;
        }

        constructor(options: IAggregateConstructorParams<TState>) {
                const { id, state, events } = options;
                if (!id)
                        throw new TypeError('id argument required');
                if (state && typeof state !== 'object')
                        throw new TypeError('state argument, when provided, must be an Object');
                if (events && !Array.isArray(events))
                        throw new TypeError('events argument, when provided, must be an Array');

                this.#id = id;

                validateHandlers(this);

                if (state)
                        this.state = state;

                if (events)
                        events.forEach(event => this.mutate(event));
        }

        /** Pass command to command handler */
        handle(command: ICommand) {
                if (!command)
                        throw new TypeError('command argument required');
                if (!command.type)
                        throw new TypeError('command.type argument required');

                const handler = getHandler(this, command.type);
                if (!handler)
                        throw new Error(`'${command.type}' handler is not defined or not a function`);

                this.command = command;

                return handler.call(this, command.payload, command.context);
        }

        /** Mutate aggregate state and increment aggregate version */
        mutate(event: IEvent) {
                if (event.aggregateVersion !== undefined)
                        this.#version = event.aggregateVersion;

                if (event.type === SNAPSHOT_EVENT_TYPE) {
                        this.#snapshotVersion = event.aggregateVersion;
                        this.restoreSnapshot(event);
                }
                else if (this.state) {
                        const handler = 'mutate' in this.state ?
                                this.state.mutate :
                                getHandler(this.state, event.type);
                        if (handler)
                                handler.call(this.state, event);
                }

                this.#version += 1;
        }

        /** Format and register aggregate event and mutate aggregate state */
        protected emit<TPayload>(type: string, payload?: TPayload) {
                if (typeof type !== 'string' || !type.length)
                        throw new TypeError('type argument must be a non-empty string');

                const event = this.makeEvent<TPayload>(type, payload, this.command);

                this.emitRaw(event);
        }

        /** Format event based on a current aggregate state and a command being executed */
        protected makeEvent<TPayload>(type: string, payload?: TPayload, sourceCommand?: ICommand): IEvent<TPayload> {
                const event: IEvent<TPayload> = {
                        aggregateId: this.id,
                        aggregateVersion: this.version,
                        type,
                        payload
                };

                if (sourceCommand) {
                        // augment event with command context
                        const { context, sagaId, sagaVersion } = sourceCommand;
                        if (context !== undefined)
                                event.context = context;
                        if (sagaId !== undefined)
                                event.sagaId = sagaId;
                        if (sagaVersion !== undefined)
                                event.sagaVersion = sagaVersion;
                }

                return event;
        }

        /** Register aggregate event and mutate aggregate state */
        protected emitRaw<TPayload>(event: IEvent<TPayload>): void {
                if (!event)
                        throw new TypeError('event argument required');
                if (!event.aggregateId)
                        throw new TypeError('event.aggregateId argument required');
                if (typeof event.aggregateVersion !== 'number')
                        throw new TypeError('event.aggregateVersion argument must be a Number');
                if (typeof event.type !== 'string' || !event.type.length)
                        throw new TypeError('event.type argument must be a non-empty String');

                this.mutate(event);

                this.#changes.push(event);
        }

        /**
         * Take an aggregate state snapshot and add it to the changes queue
         */
        takeSnapshot() {
                this.emit(SNAPSHOT_EVENT_TYPE, this.makeSnapshot());
        }

        /** Create an aggregate state snapshot */
        makeSnapshot(): TState {
                if (!this.state)
                        throw new Error('state property is empty, either define state or override makeSnapshot method');

                return clone(this.state);
        }

        /** Restore aggregate state from a snapshot */
        protected restoreSnapshot(snapshotEvent: IEvent<TState>) {
                if (!snapshotEvent)
                        throw new TypeError('snapshotEvent argument required');
                if (!snapshotEvent.type)
                        throw new TypeError('snapshotEvent.type argument required');
                if (!snapshotEvent.payload)
                        throw new TypeError('snapshotEvent.payload argument required');

                if (snapshotEvent.type !== SNAPSHOT_EVENT_TYPE)
                        throw new Error(`${SNAPSHOT_EVENT_TYPE} event type expected`);
                if (!this.state)
                        throw new Error('state property is empty, either defined state or override restoreSnapshot method');

                Object.assign(this.state, clone(snapshotEvent.payload));
        }

        /** Get human-readable aggregate identifier */
        toString(): string {
                return `${getClassName(this)} ${this.id} (v${this.version})`;
        }
}

1	import {
2	IAggregate,
3	IMutableAggregateState,
4	ICommand,
5	Identifier,
6	IEvent,
7	IEventSet,
8	IAggregateConstructorParams
9	} from './interfaces';
10
11	import { getClassName, validateHandlers, getHandler, getMessageHandlerNames } from './utils';	24✔
12
13	/**
14	* Deep-clone simple JS object
15	*/
16	function clone<T>(obj: T): T {
17	return JSON.parse(JSON.stringify(obj));	18✔
18	}
19
20	const SNAPSHOT_EVENT_TYPE = 'snapshot';	24✔
21
22	/**
23	* Base class for Aggregate definition
24	*/
25	export abstract class AbstractAggregate<TState extends IMutableAggregateState \| object \| void> implements IAggregate {	24✔
26
27	/**
28	* List of command names handled by the Aggregate.
29	*
30	* Can be overridden in the Aggregate implementation to explicitly define supported commands.
31	* If not overridden, all public methods will be treated as command handlers by default.
32	*
33	* @example ['createUser', 'changePassword'];
34	*/
35	static get handles(): string[] {
36	return getMessageHandlerNames(this);	8✔
37	}
38
39	#id: Identifier;
40	#changes: IEvent[] = [];	80✔
41	#version: number = 0;	80✔
42	#snapshotVersion: number \| undefined;
43
44	/** Internal aggregate state */
45	protected state: TState \| undefined;
46
47	/** Command being handled by aggregate */
48	protected command?: ICommand;
49
50	/** Unique aggregate instance identifier */
51	get id(): Identifier {
52	return this.#id;	38✔
53	}
54
55	/** Aggregate instance version */
56	get version(): number {
57	return this.#version;	64✔
58	}
59
60	/** Restored snapshot version */
61	get snapshotVersion(): number \| undefined {
62	return this.#snapshotVersion;	4✔
63	}
64
65	/** Events emitted by Aggregate */
66	get changes(): IEventSet {
67	return [...this.#changes];	48✔
68	}
69
70	/**
71	* Override to define whether an aggregate state snapshot should be taken
72	*
73	* @example
74	* // create snapshot every 50 events
75	* return this.version % 50 === 0;
76	*/
77	// eslint-disable-next-line class-methods-use-this
78	get shouldTakeSnapshot(): boolean {
79	return false;	12✔
80	}
81
82	constructor(options: IAggregateConstructorParams<TState>) {
83	const { id, state, events } = options;	80✔
84	if (!id)	80!
85	throw new TypeError('id argument required');	×
86	if (state && typeof state !== 'object')	80!
87	throw new TypeError('state argument, when provided, must be an Object');	×
88	if (events && !Array.isArray(events))	80✔
89	throw new TypeError('events argument, when provided, must be an Array');	×
90
91	this.#id = id;	80✔
92
93	validateHandlers(this);	80✔
94
95	if (state)	78✔
96	this.state = state;	72✔
97
98	if (events)	78✔
99	events.forEach(event => this.mutate(event));	8✔
100	}
101
102	/** Pass command to command handler */
103	handle(command: ICommand) {
104	if (!command)	28!
105	throw new TypeError('command argument required');	×
106	if (!command.type)	28!
107	throw new TypeError('command.type argument required');	×
108
109	const handler = getHandler(this, command.type);	28✔
110	if (!handler)	28✔
111	throw new Error(`'${command.type}' handler is not defined or not a function`);	2✔
112
113	this.command = command;	26✔
114
115	return handler.call(this, command.payload, command.context);	26✔
116	}
117
118	/** Mutate aggregate state and increment aggregate version */
119	mutate(event: IEvent) {
120	if (event.aggregateVersion !== undefined)	58✔
121	this.#version = event.aggregateVersion;	40✔
122
123	if (event.type === SNAPSHOT_EVENT_TYPE) {	58✔
124	this.#snapshotVersion = event.aggregateVersion;	10✔
125	this.restoreSnapshot(event);	10✔
126	}
127	else if (this.state) {	48✔
128	const handler = 'mutate' in this.state ?	48✔
129	this.state.mutate :
130	getHandler(this.state, event.type);
131	if (handler)	48✔
132	handler.call(this.state, event);	30✔
133	}
134
135	this.#version += 1;	58✔
136	}
137
138	/** Format and register aggregate event and mutate aggregate state */
139	protected emit<TPayload>(type: string, payload?: TPayload) {
140	if (typeof type !== 'string' \|\| !type.length)	36!
141	throw new TypeError('type argument must be a non-empty string');	×
142
143	const event = this.makeEvent<TPayload>(type, payload, this.command);	36✔
144
145	this.emitRaw(event);	36✔
146	}
147
148	/** Format event based on a current aggregate state and a command being executed */
149	protected makeEvent<TPayload>(type: string, payload?: TPayload, sourceCommand?: ICommand): IEvent<TPayload> {
150	const event: IEvent<TPayload> = {	36✔
151	aggregateId: this.id,
152	aggregateVersion: this.version,
153	type,
154	payload
155	};
156
157	if (sourceCommand) {	36✔
158	// augment event with command context
159	const { context, sagaId, sagaVersion } = sourceCommand;	26✔
160	if (context !== undefined)	26✔
161	event.context = context;	2✔
162	if (sagaId !== undefined)	26✔
163	event.sagaId = sagaId;	2✔
164	if (sagaVersion !== undefined)	26✔
165	event.sagaVersion = sagaVersion;	2✔
166	}
167
168	return event;	36✔
169	}
170
171	/** Register aggregate event and mutate aggregate state */
172	protected emitRaw<TPayload>(event: IEvent<TPayload>): void {
173	if (!event)	36!
174	throw new TypeError('event argument required');	×
175	if (!event.aggregateId)	36!
176	throw new TypeError('event.aggregateId argument required');	×
177	if (typeof event.aggregateVersion !== 'number')	36!
178	throw new TypeError('event.aggregateVersion argument must be a Number');	×
179	if (typeof event.type !== 'string' \|\| !event.type.length)	36!
180	throw new TypeError('event.type argument must be a non-empty String');	×
181
182	this.mutate(event);	36✔
183
184	this.#changes.push(event);	36✔
185	}
186
187	/**
188	* Take an aggregate state snapshot and add it to the changes queue
189	*/
190	takeSnapshot() {
191	this.emit(SNAPSHOT_EVENT_TYPE, this.makeSnapshot());	4✔
192	}
193
194	/** Create an aggregate state snapshot */
195	makeSnapshot(): TState {
196	if (!this.state)	4!
197	throw new Error('state property is empty, either define state or override makeSnapshot method');	×
198
199	return clone(this.state);	4✔
200	}
201
202	/** Restore aggregate state from a snapshot */
203	protected restoreSnapshot(snapshotEvent: IEvent<TState>) {
204	if (!snapshotEvent)	22✔
205	throw new TypeError('snapshotEvent argument required');	2✔
206	if (!snapshotEvent.type)	20✔
207	throw new TypeError('snapshotEvent.type argument required');	2✔
208	if (!snapshotEvent.payload)	18✔
209	throw new TypeError('snapshotEvent.payload argument required');	2✔
210
211	if (snapshotEvent.type !== SNAPSHOT_EVENT_TYPE)	16✔
212	throw new Error(`${SNAPSHOT_EVENT_TYPE} event type expected`);	2✔
213	if (!this.state)	14!
214	throw new Error('state property is empty, either defined state or override restoreSnapshot method');	×
215
216	Object.assign(this.state, clone(snapshotEvent.payload));	14✔
217	}
218
219	/** Get human-readable aggregate identifier */
220	toString(): string {
221	return `${getClassName(this)} ${this.id} (v${this.version})`;	×
222	}
223	}

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