21845544207

Committed 09 Feb 2026 11:50PM UTC coverage: 85.047%. First build

Build # 21845544207

Build Type

Pull #31

github

Committed by

web-flow

Commit Message

Merge 788523d43 into 025edb883

Pull Request Pull Request #31: Multi-saga correlation via `message.sagaOrigins`

Run Details

650 of 986 branches covered (65.92%)

140 of 155 new or added lines in 17 files covered. (90.32%)

1257 of 1478 relevant lines covered (85.05%)

31.91 hits per line

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

77.59

/src/AbstractSaga.ts

import {
        type ICommand,
        type ICommandBus,
        type Identifier,
        type IEvent,
        type IEventStore,
        type IMutableState,
        type ISaga,
        type ISagaConstructor,
        type ISagaConstructorParams,
        isEvent
} from './interfaces/index.ts';
import { SagaEventHandler } from './SagaEventHandler.ts';
import {
        validateHandlers,
        getHandler,
        promiseOrSync,
        getClassName
} from './utils/index.ts';

/**
 * Base class for Saga definition
 */
export abstract class AbstractSaga implements ISaga {

        /** List of events that start new saga, must be overridden in Saga implementation */
        static get startsWith(): string[] {
                throw new Error('startsWith must be overridden to return a list of event types that start saga');
        }

        /** List of event types being handled by Saga, can be overridden in Saga implementation */
        static get handles(): string[] {
                return [];
        }

        /**
         * Convenience helper to create a `SagaEventHandler` for this saga type and subscribe it to
         * the provided `eventStore`.
         */
        static register<T extends AbstractSaga>(
                this: ISagaConstructor & (new (options: ISagaConstructorParams) => T),
                eventStore: IEventStore,
                commandBus: ICommandBus
        ): SagaEventHandler {
                const handler = new SagaEventHandler({
                        sagaType: this,
                        eventStore,
                        commandBus
                });
                handler.subscribe(eventStore);
                return handler;
        }

        /** Saga ID */
        get id(): Identifier {
                return this.#id;
        }

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

        protected state?: IMutableState | object;

        #id: Identifier;
        #version = 0;
        #messages: ICommand[] = [];
        #handling = false;

        /**
         * Creates an instance of AbstractSaga
         */
        constructor(options: ISagaConstructorParams) {
                if (!options)
                        throw new TypeError('options argument required');
                if (!options.id)
                        throw new TypeError('options.id argument required');
                if (options.events)
                        throw new TypeError('options.events argument is deprecated');

                this.#id = options.id;

                validateHandlers(this, 'startsWith');
                validateHandlers(this, 'handles');
        }

        /** Modify saga state by applying an event */
        mutate(event: IEvent): void {
                if (!isEvent(event))
                        throw new TypeError('event argument must be a valid IEvent');

                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;
        }

        /** Process saga event and return produced commands */
        handle(event: IEvent): ReadonlyArray<ICommand> | Promise<ReadonlyArray<ICommand>> {
                if (!isEvent(event))
                        throw new TypeError('event argument must be a valid IEvent');
                if (this.#handling)
                        throw new Error('Another event is being processed, concurrent handling is not allowed');

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

                this.#handling = true;
                this.#messages.length = 0;

                try {
                        const r = handler.call(this, event);

                        return promiseOrSync(r, () => {
                                this.mutate(event);
                                return this.#messages.splice(0);
                        }, () => {
                                this.#handling = false;
                        });
                }
                catch (err) {
                        this.#handling = false;
                        throw err;
                }
        }

        /** Format a command and put it to the execution queue */
        protected enqueue(commandType: string): void;
        protected enqueue(commandType: string, aggregateId: Identifier): void;
        protected enqueue<T>(commandType: string, aggregateId: Identifier | undefined, payload: T): void;
        protected enqueue<T>(commandType: string, aggregateId?: Identifier, payload?: T) {
                if (typeof commandType !== 'string' || !commandType.length)
                        throw new TypeError('commandType argument must be a non-empty String');
                if (!['string', 'number', 'undefined'].includes(typeof aggregateId))
                        throw new TypeError('aggregateId argument must be either string, number or undefined');

                this.enqueueRaw({
                        aggregateId,
                        type: commandType,
                        payload
                });
        }

        /** Put a command to the execution queue */
        protected enqueueRaw(command: ICommand) {
                if (typeof command !== 'object' || !command)
                        throw new TypeError('command argument must be an Object');
                if (typeof command.type !== 'string' || !command.type.length)
                        throw new TypeError('command.type argument must be a non-empty String');

                this.#messages.push(command);
        }

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

1	import {	28✔
2	type ICommand,
3	type ICommandBus,
4	type Identifier,
5	type IEvent,
6	type IEventStore,
7	type IMutableState,
8	type ISaga,
9	type ISagaConstructor,
10	type ISagaConstructorParams,
11	isEvent
12	} from './interfaces/index.ts';
13	import { SagaEventHandler } from './SagaEventHandler.ts';	28✔
14	import {	28✔
15	validateHandlers,
16	getHandler,
17	promiseOrSync,
18	getClassName
19	} from './utils/index.ts';
20
21	/**
22	* Base class for Saga definition
23	*/
24	export abstract class AbstractSaga implements ISaga {	28✔
25
26	/** List of events that start new saga, must be overridden in Saga implementation */
27	static get startsWith(): string[] {
28	throw new Error('startsWith must be overridden to return a list of event types that start saga');	2✔
29	}
30
31	/** List of event types being handled by Saga, can be overridden in Saga implementation */
32	static get handles(): string[] {
33	return [];	44✔
34	}
35
36	/**
37	* Convenience helper to create a `SagaEventHandler` for this saga type and subscribe it to
38	* the provided `eventStore`.
39	*/
40	static register<T extends AbstractSaga>(
41	this: ISagaConstructor & (new (options: ISagaConstructorParams) => T),
42	eventStore: IEventStore,
43	commandBus: ICommandBus
44	): SagaEventHandler {
45	const handler = new SagaEventHandler({	2✔
46	sagaType: this,
47	eventStore,
48	commandBus
49	});
50	handler.subscribe(eventStore);	2✔
51	return handler;	2✔
52	}
53
54	/** Saga ID */
55	get id(): Identifier {
56	return this.#id;	2✔
57	}
58
59	/** Saga version */
60	get version(): number {
61	return this.#version;	6✔
62	}
63
64	protected state?: IMutableState \| object;
65
66	#id: Identifier;
67	#version = 0;	52✔
68	#messages: ICommand[] = [];	52✔
69	#handling = false;	52✔
70
71	/**
72	* Creates an instance of AbstractSaga
73	*/
74	constructor(options: ISagaConstructorParams) {
75	if (!options)	52!
76	throw new TypeError('options argument required');	×
77	if (!options.id)	52!
78	throw new TypeError('options.id argument required');	×
79	if (options.events)	52!
NEW 80	throw new TypeError('options.events argument is deprecated');	×
81
82	this.#id = options.id;	52✔
83
84	validateHandlers(this, 'startsWith');	52✔
85	validateHandlers(this, 'handles');	48✔
86	}
87
88	/** Modify saga state by applying an event */
89	mutate(event: IEvent): void {
90	if (!isEvent(event))	44!
NEW 91	throw new TypeError('event argument must be a valid IEvent');	×
92
93	if (this.state) {	44✔
94	const handler = 'mutate' in this.state ?	6✔
95	this.state.mutate :
96	getHandler(this.state, event.type);
97	if (handler)	6✔
98	handler.call(this.state, event);	6✔
99	}
100
101	this.#version += 1;	44✔
102	}
103
104	/** Process saga event and return produced commands */
105	handle(event: IEvent): ReadonlyArray<ICommand> \| Promise<ReadonlyArray<ICommand>> {
106	if (!isEvent(event))	32!
NEW 107	throw new TypeError('event argument must be a valid IEvent');	×
108	if (this.#handling)	32✔
109	throw new Error('Another event is being processed, concurrent handling is not allowed');	2✔
110
111	const handler = getHandler(this, event.type);	30✔
112	if (!handler)	30!
113	throw new Error(`'${event.type}' handler is not defined or not a function`);	×
114
115	this.#handling = true;	30✔
116	this.#messages.length = 0;	30✔
117
118	try {	30✔
119	const r = handler.call(this, event);	30✔
120
121	return promiseOrSync(r, () => {	30✔
122	this.mutate(event);	30✔
123	return this.#messages.splice(0);	30✔
124	}, () => {
125	this.#handling = false;	30✔
126	});
127	}
128	catch (err) {
NEW 129	this.#handling = false;	×
NEW 130	throw err;	×
131	}
132	}
133
134	/** Format a command and put it to the execution queue */
135	protected enqueue(commandType: string): void;
136	protected enqueue(commandType: string, aggregateId: Identifier): void;
137	protected enqueue<T>(commandType: string, aggregateId: Identifier \| undefined, payload: T): void;
138	protected enqueue<T>(commandType: string, aggregateId?: Identifier, payload?: T) {
139	if (typeof commandType !== 'string' \|\| !commandType.length)	26!
140	throw new TypeError('commandType argument must be a non-empty String');	×
141	if (!['string', 'number', 'undefined'].includes(typeof aggregateId))	26!
142	throw new TypeError('aggregateId argument must be either string, number or undefined');	×
143
144	this.enqueueRaw({	26✔
145	aggregateId,
146	type: commandType,
147	payload
148	});
149	}
150
151	/** Put a command to the execution queue */
152	protected enqueueRaw(command: ICommand) {
153	if (typeof command !== 'object' \|\| !command)	26✔
154	throw new TypeError('command argument must be an Object');	×
155	if (typeof command.type !== 'string' \|\| !command.type.length)	26!
156	throw new TypeError('command.type argument must be a non-empty String');	×
157
158	this.#messages.push(command);	26✔
159	}
160
161	/** Get human-readable Saga name */
162	toString(): string {
163	return `${getClassName(this)} ${this.id} (v${this.version})`;	×
164	}
165	}

snatalenko / node-cqrs / 21845544207

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