<commit_before>'use strict';

const InMemoryBus = require('./infrastructure/InMemoryMessageBus');
const debug = require('debug')('cqrs:debug:EventStore');
const info = require('debug')('cqrs:info:EventStore');
const coWrap = require('./utils/coWrap');

const _storage = Symbol('storage');
const _messageBus = Symbol('messageBus');
const _validator = Symbol('validator');

function eventsToString(events) {
	if (!events) events = this;
	return !Array.isArray(events) ? events :
		events.length === 1 ? '\'' + events[0].type + '\'' :
			events.length + ' events';
}

function validateEvent(event) {
	if (typeof event !== 'object' || !event) throw new TypeError('event must be an Object');
	if (typeof event.type !== 'string' || !event.type.length) throw new TypeError('event.type must be a non-empty String');
	if (!event.aggregateId && !event.sagaId) throw new TypeError('either event.aggregateId or event.sagaId is required');
	if (event.sagaId && typeof event.sagaVersion === 'undefined') throw new TypeError('event.sagaVersion is required, when event.sagaId is defined');
}


module.exports = class EventStore {

	get storage() {
		return this[_storage];
	}

	set storage(storage) {
		if (storage) {
			if (typeof storage.commitEvents !== 'function') throw new TypeError('storage.commitEvents must be a Function');
			if (typeof storage.getEvents !== 'function') throw new TypeError('storage.getEvents must be a Function');
			if (typeof storage.getAggregateEvents !== 'function') throw new TypeError('storage.getAggregateEvents must be a Function');
			if (typeof storage.getSagaEvents !== 'function') throw new TypeError('storage.getSagaEvents must be a Function');
			if (typeof storage.getNewId !== 'function') throw new TypeError('storage.getNewId must be a Function');
		}
		this[_storage] = storage || undefined;
	}

	get bus() {
		return this[_messageBus];
	}

	set bus(bus) {
		if (bus && typeof bus.publish !== 'function') throw new TypeError('bus.publish must be a Function');

		this[_messageBus] = bus || undefined;
	}

	get validator() {
		return this[_validator];
	}

	set validator(validator) {
		if (validator && typeof validator !== 'function') throw new TypeError('validator must be a Function');

		this[_validator] = validator || undefined;
	}

	constructor(options) {
		if (!options) throw new TypeError('options argument required');
		if (!options.storage) throw new TypeError('options.storage argument required');

		this.storage = options.storage;
		this.bus = options.messageBus || new InMemoryBus();
		this.validator = options.validator || validateEvent;
		this.publishAsync = 'publishAsync' in options ? !!options.publishAsync : true;

		coWrap(this, [
			'getAllEvents',
			'getAggregateEvents',
			'getSagaEvents',
			'commit'
		]);
	}

	getNewId() {
		return Promise.resolve(this.storage.getNewId());
	}

	*getAllEvents(eventTypes) {
		if (eventTypes && !Array.isArray(eventTypes)) throw new TypeError('eventTypes, if specified, must be an Array');

		debug(`retrieving ${eventTypes ? eventTypes.join(', ') : 'all'} events...`);

		const events = yield this.storage.getEvents(eventTypes) || [];
		debug(`${eventsToString(events)} retreieved`);

		return events;
	}

	*getAggregateEvents(aggregateId) {
		if (!aggregateId) throw new TypeError('aggregateId argument required');

		debug(`retrieving event stream for aggregate ${aggregateId}...`);

		const events = yield this.storage.getAggregateEvents(aggregateId) || [];
		debug(`${eventsToString(events)} retreieved`);

		return events;
	}

	/**
	 * Retrieves events, by sagaId
	 *
	 * @param {string} sagaId
	 * @param {{before:number}} options
	 * @returns {PromiseLike<object[]>}
	 */
	*getSagaEvents(sagaId, options) {
		if (!sagaId) throw new TypeError('sagaId argument required');

		debug(`retrieving event stream for saga ${sagaId}...`);

		const events = yield this.storage.getSagaEvents(sagaId, options) || [];
		debug(`${eventsToString(events)} retreieved`);

		if (options && Object.keys(options).length) {
			const {after, before, except} = options;
			const filteredEvents = events.filter(e =>
				(typeof before === 'undefined' || e.sagaVersion < before) &&
				(typeof after === 'undefined' || e.sagaVersion > after) &&
				(typeof except === 'undefined' || e.id != except));

			if(filteredEvents.length !== events.length){
				debug(`${events.length - filteredEvents.length} events excluded by filter: %o`, options);
			}

			return filteredEvents;
		}
		else {
			return events;
		}
	}

	/**
	 * Sign, validate, and commit events to storage
	 *
	 * @param {object[]} events - a set of events to commit
	 * @returns {PromiseLike<object[]>} - resolves to signed and committed events
	 */
	*commit(events) {
		if (!Array.isArray(events)) throw new TypeError('events argument must be an Array');

		debug(`validating ${eventsToString(events)}...`);
		yield events.map(this.validator);

		debug(`committing ${eventsToString(events)}...`);
		yield this.storage.commitEvents(events);

		if (this.publishAsync) {
			debug(`publishing ${eventsToString(events)} asynchronously...`);
			setImmediate(() => Promise.all(events.map(event => this.bus.publish(event))).then(result => {
				info(`${eventsToString(events)} published`);
			}, err => {
				info(`${eventsToString(events)} publishing failed: ${err.stack || err}`);
			}));
		}
		else {
			debug(`publishing ${eventsToString(events)} synchronously...`);
			try{
				yield events.map(event => this.bus.publish(event));
			}
			catch(err) {
				info(`${eventsToString(events)} publishing failed: ${err.stack || err}`);
				throw err;
			}<commit_msg>* changed getSagaEvents options format to keep the saga event stream retrieval logic within the EventStore<commit_after>'use strict';

const InMemoryBus = require('./infrastructure/InMemoryMessageBus');
const debug = require('debug')('cqrs:debug:EventStore');
const info = require('debug')('cqrs:info:EventStore');
const coWrap = require('./utils/coWrap');

const _storage = Symbol('storage');
const _messageBus = Symbol('messageBus');
const _validator = Symbol('validator');

function eventsToString(events) {
	if (!events) events = this;
	return !Array.isArray(events) ? events :
		events.length === 1 ? '\'' + events[0].type + '\'' :
			events.length + ' events';
}

function validateEvent(event) {
	if (typeof event !== 'object' || !event) throw new TypeError('event must be an Object');
	if (typeof event.type !== 'string' || !event.type.length) throw new TypeError('event.type must be a non-empty String');
	if (!event.aggregateId && !event.sagaId) throw new TypeError('either event.aggregateId or event.sagaId is required');
	if (event.sagaId && typeof event.sagaVersion === 'undefined') throw new TypeError('event.sagaVersion is required, when event.sagaId is defined');
}


module.exports = class EventStore {

	get storage() {
		return this[_storage];
	}

	set storage(storage) {
		if (storage) {
			if (typeof storage.commitEvents !== 'function') throw new TypeError('storage.commitEvents must be a Function');
			if (typeof storage.getEvents !== 'function') throw new TypeError('storage.getEvents must be a Function');
			if (typeof storage.getAggregateEvents !== 'function') throw new TypeError('storage.getAggregateEvents must be a Function');
			if (typeof storage.getSagaEvents !== 'function') throw new TypeError('storage.getSagaEvents must be a Function');
			if (typeof storage.getNewId !== 'function') throw new TypeError('storage.getNewId must be a Function');
		}
		this[_storage] = storage || undefined;
	}

	get bus() {
		return this[_messageBus];
	}

	set bus(bus) {
		if (bus && typeof bus.publish !== 'function') throw new TypeError('bus.publish must be a Function');

		this[_messageBus] = bus || undefined;
	}

	get validator() {
		return this[_validator];
	}

	set validator(validator) {
		if (validator && typeof validator !== 'function') throw new TypeError('validator must be a Function');

		this[_validator] = validator || undefined;
	}

	constructor(options) {
		if (!options) throw new TypeError('options argument required');
		if (!options.storage) throw new TypeError('options.storage argument required');

		this.storage = options.storage;
		this.bus = options.messageBus || new InMemoryBus();
		this.validator = options.validator || validateEvent;
		this.publishAsync = 'publishAsync' in options ? !!options.publishAsync : true;

		coWrap(this, [
			'getAllEvents',
			'getAggregateEvents',
			'getSagaEvents',
			'commit'
		]);
	}

	getNewId() {
		return Promise.resolve(this.storage.getNewId());
	}

	*getAllEvents(eventTypes) {
		if (eventTypes && !Array.isArray(eventTypes)) throw new TypeError('eventTypes, if specified, must be an Array');

		debug(`retrieving ${eventTypes ? eventTypes.join(', ') : 'all'} events...`);

		const events = yield this.storage.getEvents(eventTypes) || [];
		debug(`${eventsToString(events)} retreieved`);

		return events;
	}

	*getAggregateEvents(aggregateId) {
		if (!aggregateId) throw new TypeError('aggregateId argument required');

		debug(`retrieving event stream for aggregate ${aggregateId}...`);

		const events = yield this.storage.getAggregateEvents(aggregateId) || [];
		debug(`${eventsToString(events)} retreieved`);

		return events;
	}

	/**
	 * Retrieves events, by sagaId
	 *
	 * @param {string} sagaId
	 * @param {{eventId: string, sagaVersion: number}} options
	 * @returns {PromiseLike<object[]>}
	 */
	*getSagaEvents(sagaId, options) {
		if (!sagaId) throw new TypeError('sagaId argument required');

		// 'except' and 'before' are deprecated, left here for backward compatibility.
		// options argument should contain sagaVersion and eventId, so the logic of
		// event stream retrieval will be handled by the EventStore
		const sagaVersion = options && (options.sagaVersion || options.before);
		const eventId = options && (options.eventId || options.except);

		debug(`retrieving event stream for saga ${sagaId}, v${sagaVersion}, except ${eventId}...`);

		const events = yield this.storage.getSagaEvents(sagaId, { except: eventId }) || [];
		debug(`${eventsToString(events)} retreieved`);

		if (options && Object.keys(options).length) {
			const {after, before, except} = options;
			const filteredEvents = events.filter(e =>
				(typeof before === 'undefined' || e.sagaVersion < before) &&
				(typeof after === 'undefined' || e.sagaVersion > after) &&
				(typeof except === 'undefined' || e.id != except));

			if (filteredEvents.length !== events.length) {
				debug(`${events.length - filteredEvents.length} events excluded by filter: %o`, options);
			}

			return filteredEvents;
		}
		else {
			return events;
		}
	}

	/**
	 * Sign, validate, and commit events to storage
	 *
	 * @param {object[]} events - a set of events to commit
	 * @returns {PromiseLike<object[]>} - resolves to signed and committed events
	 */
	*commit(events) {
		if (!Array.isArray(events)) throw new TypeError('events argument must be an Array');

		debug(`validating ${eventsToString(events)}...`);
		yield events.map(this.validator);

		debug(`committing ${eventsToString(events)}...`);
		yield this.storage.commitEvents(events);

		if (this.publishAsync) {
			debug(`publishing ${eventsToString(events)} asynchronously...`);
			setImmediate(() => Promise.all(events.map(event => this.bus.publish(event))).then(result => {
				info(`${eventsToString(events)} published`);
			}, err => {
				info(`${eventsToString(events)} publishing failed: ${err.stack || err}`);
			}));
		}
		else {
			debug(`publishing ${eventsToString(events)} synchronously...`);
			try {
				yield events.map(event => this.bus.publish(event));
			}
			catch (err) {
				info(`${eventsToString(events)} publishing failed: ${err.stack || err}`);
				throw err;
			}
{'fraction_non_alphanumeric': 0.15464005458887753, 'fraction_numerical': 0.000341180484476288, 'mean_word_length': 4.642444658325313, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3259375', 'n_tokens_mistral': 3781, 'n_tokens_neox': 3470, 'n_words': 936}