moleculerjs/moleculer

/*
 * moleculer
 * Copyright (c) 2020 MoleculerJS (https://github.com/moleculerjs/moleculer)
 * MIT Licensed
 */

"use strict";

const _ = require("lodash");
const { ServiceSchemaError, MoleculerError } = require("./errors");
const { isObject, isFunction, flatten, functionArguments, deprecate, uniq } = require("./utils");

/**
 * Wrap a handler Function to an object with a `handler` property.
 *
 * @param {Function|Object} o
 * @returns {Object}
 */
function wrapToHandler(o) {
    return isFunction(o) ? { handler: o } : o;
}

/**
 * Wrap any value to an array.
 * @param {any} o
 * @returns {Array}
 */
function wrapToArray(o) {
    return Array.isArray(o) ? o : [o];
}

function isNewSignature(args) {
    return args.length > 0 && ["ctx", "context"].indexOf(args[0].toLowerCase()) !== -1;
}

/**
 * Service class
 *
 * @class Service
 */
class Service {
    /**
     * Creates an instance of Service by schema.
     *
     * @param {ServiceBroker}     broker    broker of service
     * @param {Object}             schema    schema of service
     * @param {any=}             schemaMods    Modified schema
     *
     * @memberof Service
     */
    constructor(broker, schema, schemaMods) {
        if (!isObject(broker)) throw new ServiceSchemaError("Must set a ServiceBroker instance!");

        this.broker = broker;

        if (broker) this.Promise = broker.Promise;
        if (schemaMods) {
            deprecate(
                "schemaMods",
                "Using 'schemaMods' parameter in 'broker.createService' is deprecated. Use 'mixins' instead."
            );
            schema = this.mergeSchemas(schema, schemaMods);
        }
        if (schema) this.parseServiceSchema(schema);
    }

    /**
     * Parse Service schema & register as local service
     *
     * @param {Object} schema of Service
     */
    parseServiceSchema(schema) {
        if (!isObject(schema))
            throw new ServiceSchemaError(
                "The service schema can't be null. Maybe is it not a service schema?"
            );

        this.originalSchema = _.cloneDeep(schema);

        if (schema.mixins) {
            schema = this.applyMixins(schema);
        }

        if (isFunction(schema.merged)) {
            schema.merged.call(this, schema);
        } else if (Array.isArray(schema.merged)) {
            schema.merged.forEach(fn => fn.call(this, schema));
        }

        this.broker.callMiddlewareHookSync("serviceCreating", [this, schema]);

        if (!schema.name) {
            /* eslint-disable-next-line no-console */
            console.error(
                "Service name can't be empty! Maybe it is not a valid Service schema. Maybe is it not a service schema?",
                { schema }
            );
            throw new ServiceSchemaError(
                "Service name can't be empty! Maybe it is not a valid Service schema. Maybe is it not a service schema?",
                { schema }
            );
        }

        this.name = schema.name;
        this.version = schema.version;
        this.settings = schema.settings || {};
        this.metadata = schema.metadata || {};
        this.schema = schema;

        this.fullName = Service.getVersionedFullName(
            this.name,
            this.settings.$noVersionPrefix !== true ? this.version : undefined
        );

        this.logger = this.broker.getLogger(this.fullName, {
            svc: this.name,
            ver: this.version
        });

        this.actions = {}; // external access to actions
        this.events = {}; // external access to event handlers.

        // Service item for Registry
        const serviceSpecification = {
            name: this.name,
            version: this.version,
            fullName: this.fullName,
            settings: this._getPublicSettings(this.settings),
            metadata: this.metadata,
            actions: {},
            events: {}
        };

        // Register methods
        if (isObject(schema.methods)) {
            _.forIn(schema.methods, (method, name) => {
                /* istanbul ignore next */
                if (
                    [
                        "name",
                        "version",
                        "settings",
                        "metadata",
                        "dependencies",
                        "schema",
                        "broker",
                        "actions",
                        "logger",
                        "created",
                        "started",
                        "stopped",
                        "_start",
                        "_stop",
                        "_init",
                        "applyMixins"
                    ].indexOf(name) !== -1 ||
                    name.startsWith("mergeSchema")
                ) {
                    throw new ServiceSchemaError(
                        `Invalid method name '${name}' in '${this.name}' service!`
                    );
                }

                this._createMethod(method, name);
            });
        }

        // Register actions
        if (isObject(schema.actions)) {
            _.forIn(schema.actions, (action, name) => {
                if (action === false) return;

                let innerAction = this._createAction(action, name);

                serviceSpecification.actions[innerAction.name] = innerAction;

                const wrappedHandler = this.broker.middlewares.wrapHandler(
                    "localAction",
                    innerAction.handler,
                    innerAction
                );

                // Expose to be callable as `this.actions.find({ ...params })`
                const ep = this.broker.registry.createPrivateActionEndpoint(innerAction);
                this.actions[name] = (params, opts) => {
                    let ctx;
                    if (opts && opts.ctx) {
                        // Reused context (in case of retry)
                        ctx = opts.ctx;
                    } else {
                        ctx = this.broker.ContextFactory.create(
                            this.broker,
                            ep,
                            params,
                            opts || {}
                        );
                    }
                    return wrappedHandler(ctx);
                };
            });
        }

        // Event subscriptions
        if (isObject(schema.events)) {
            _.forIn(schema.events, (event, name) => {
                const innerEvent = this._createEvent(event, name);
                serviceSpecification.events[innerEvent.name] = innerEvent;

                // Expose to be callable as `this.events[''](params, opts);
                this.events[innerEvent.name] = (params, opts) => {
                    let ctx;
                    if (opts && opts.ctx) {
                        // Reused context (in case of retry)
                        ctx = opts.ctx;
                    } else {
                        const ep = {
                            id: this.broker.nodeID,
                            event: innerEvent
                        };
                        ctx = this.broker.ContextFactory.create(
                            this.broker,
                            ep,
                            params,
                            opts || {}
                        );
                    }
                    ctx.eventName = name;
                    ctx.eventType = "emit";
                    ctx.eventGroups = [innerEvent.group || this.name];

                    return innerEvent.handler(ctx);
                };
            });
        }

        this._serviceSpecification = serviceSpecification;

        // Initialize
        this._init();
    }

    /**
     * Return a service settings without protected properties.
     *
     * @param {Object?} settings
     */
    _getPublicSettings(settings) {
        if (settings && Array.isArray(settings.$secureSettings)) {
            return _.omit(settings, [].concat(settings.$secureSettings, ["$secureSettings"]));
        }

        return settings;
    }

    /**
     * Initialize service. It called `created` handler in schema
     *
     * @private
     * @memberof Service
     */
    _init() {
        this.logger.debug(`Service '${this.fullName}' is creating...`);
        if (isFunction(this.schema.created)) {
            this.schema.created.call(this);
        } else if (Array.isArray(this.schema.created)) {
            this.schema.created.forEach(fn => fn.call(this));
        }

        this.broker.addLocalService(this);

        this.broker.callMiddlewareHookSync("serviceCreated", [this]);

        this.logger.debug(`Service '${this.fullName}' created.`);
    }

    /**
     * Start service
     *
     * @returns {Promise}
     * @private
     * @memberof Service
     */
    _start() {
        this.logger.debug(`Service '${this.fullName}' is starting...`);
        return this.Promise.resolve()
            .then(() => {
                return this.broker.callMiddlewareHook("serviceStarting", [this]);
            })
            .then(() => {
                // Wait for dependent services
                if (this.schema.dependencies)
                    return this.waitForServices(
                        this.schema.dependencies,
                        this.settings.$dependencyTimeout || this.broker.options.dependencyTimeout,
                        this.settings.$dependencyInterval || this.broker.options.dependencyInterval
                    );
            })
            .then(() => {
                if (isFunction(this.schema.started))
                    return this.Promise.method(this.schema.started).call(this);

                if (Array.isArray(this.schema.started)) {
                    return this.schema.started
                        .map(fn => this.Promise.method(fn.bind(this)))
                        .reduce((p, fn) => p.then(() => fn()), this.Promise.resolve());
                }
            })
            .then(() => {
                // Register service
                return this.broker.registerLocalService(this._serviceSpecification);
            })
            .then(() => {
                return this.broker.callMiddlewareHook("serviceStarted", [this]);
            })
            .then(() => this.logger.info(`Service '${this.fullName}' started.`));
    }

    /**
     * Stop service
     *
     * @returns {Promise}
     * @private
     * @memberof Service
     */
    _stop() {
        this.logger.debug(`Service '${this.fullName}' is stopping...`);
        return this.Promise.resolve()
            .then(() => {
                return this.broker.callMiddlewareHook("serviceStopping", [this], { reverse: true });
            })
            .then(() => {
                if (isFunction(this.schema.stopped))
                    return this.Promise.method(this.schema.stopped).call(this);

                if (Array.isArray(this.schema.stopped)) {
                    const arr = Array.from(this.schema.stopped).reverse();
                    return arr
                        .map(fn => this.Promise.method(fn.bind(this)))
                        .reduce((p, fn) => p.then(() => fn()), this.Promise.resolve());
                }

                return this.Promise.resolve();
            })
            .then(() => {
                return this.broker.callMiddlewareHook("serviceStopped", [this], { reverse: true });
            })
            .then(() => this.logger.info(`Service '${this.fullName}' stopped.`));
    }

    /**
     * Create an external action handler for broker (internal command!)
     *
     * @param {Object|Function} actionDef
     * @param {String} name
     * @returns {Object}
     *
     * @private
     * @memberof Service
     */
    _createAction(actionDef, name) {
        let action;
        if (isFunction(actionDef)) {
            // Wrap to an object
            action = {
                handler: actionDef
            };
        } else if (isObject(actionDef)) {
            action = _.cloneDeep(actionDef);
        } else {
            throw new ServiceSchemaError(
                `Invalid action definition in '${name}' action in '${this.fullName}' service!`
            );
        }

        let handler = action.handler;
        if (!isFunction(handler)) {
            throw new ServiceSchemaError(
                `Missing action handler on '${name}' action in '${this.fullName}' service!`
            );
        }

        action.rawName = action.name || name;
        if (this.settings.$noServiceNamePrefix !== true)
            action.name = this.fullName + "." + action.rawName;
        else action.name = action.rawName;

        if (action.cache === undefined && this.settings.$cache !== undefined) {
            action.cache = this.settings.$cache;
        }

        action.service = this;
        action.handler = this.Promise.method(handler.bind(this));

        return action;
    }

    /**
     * Create an internal service method.
     *
     * @param {Object|Function} methodDef
     * @param {String} name
     * @returns {Object}
     */
    _createMethod(methodDef, name) {
        let method;
        if (isFunction(methodDef)) {
            // Wrap to an object
            method = {
                handler: methodDef
            };
        } else if (isObject(methodDef)) {
            method = methodDef;
        } else {
            throw new ServiceSchemaError(
                `Invalid method definition in '${name}' method in '${this.fullName}' service!`
            );
        }

        if (!isFunction(method.handler)) {
            throw new ServiceSchemaError(
                `Missing method handler on '${name}' method in '${this.fullName}' service!`
            );
        }

        method.name = name;
        method.service = this;
        method.handler = method.handler.bind(this);

        this[name] = this.broker.middlewares.wrapHandler("localMethod", method.handler, method);

        return method;
    }

    /**
     * Create an event subscription for broker
     *
     * @param {Object|Function} eventDef
     * @param {String} name
     * @returns {Object}
     *
     * @private
     * @memberof Service
     */
    _createEvent(eventDef, name) {
        let event;
        if (isFunction(eventDef) || Array.isArray(eventDef)) {
            event = {
                handler: eventDef
            };
        } else if (isObject(eventDef)) {
            event = _.cloneDeep(eventDef);
        } else {
            throw new ServiceSchemaError(
                `Invalid event definition in '${name}' event in '${this.fullName}' service!`
            );
        }

        if (!isFunction(event.handler) && !Array.isArray(event.handler)) {
            throw new ServiceSchemaError(
                `Missing event handler on '${name}' event in '${this.fullName}' service!`
            );
        }

        // Detect new or legacy parameter list of event handler
        // Legacy: handler(payload, sender, eventName)
        // New: handler(ctx)
        let handler;
        if (isFunction(event.handler)) {
            const args = functionArguments(event.handler);
            handler = this.Promise.method(event.handler);
            handler.__newSignature = event.context === true || isNewSignature(args);
        } else if (Array.isArray(event.handler)) {
            handler = event.handler.map(h => {
                const args = functionArguments(h);
                h = this.Promise.method(h);
                h.__newSignature = event.context === true || isNewSignature(args);
                return h;
            });
        }

        if (!event.name) event.name = name;

        event.service = this;
        const self = this;
        if (isFunction(handler)) {
            // Call single handler
            event.handler = function (ctx) {
                return handler.apply(
                    self,
                    handler.__newSignature ? [ctx] : [ctx.params, ctx.nodeID, ctx.eventName, ctx]
                );
            };
        } else if (Array.isArray(handler)) {
            // Call multiple handler
            event.handler = function (ctx) {
                return self.Promise.all(
                    handler.map(fn =>
                        fn.apply(
                            self,
                            fn.__newSignature ? [ctx] : [ctx.params, ctx.nodeID, ctx.eventName, ctx]
                        )
                    )
                );
            };
        }

        return event;
    }

    /**
     * Call a local event handler. Useful for unit tests.
     *
     * @param {String} eventName
     * @param {any?} params
     * @param {Object?} opts
     */
    emitLocalEventHandler(eventName, params, opts) {
        if (!this.events[eventName])
            return Promise.reject(
                new MoleculerError(
                    `No '${eventName}' registered local event handler`,
                    500,
                    "NOT_FOUND_EVENT",
                    { eventName }
                )
            );

        return this.events[eventName](params, opts);
    }

    /**
     * Getter of current Context.
     * @returns {Context?}
     *
     * @memberof Service
     *
    get currentContext() {
        return this.broker.getCurrentContext();
    }*/

    /**
     * Setter of current Context
     *
     * @memberof Service
     *
    set currentContext(ctx) {
        this.broker.setCurrentContext(ctx);
    }*/

    /**
     * Wait for other services
     *
     * @param {String|Array<String>} serviceNames
     * @param {Number} timeout Timeout in milliseconds
     * @param {Number} interval Check interval in milliseconds
     * @returns {Promise}
     * @memberof Service
     */
    waitForServices(serviceNames, timeout, interval) {
        return this.broker.waitForServices(serviceNames, timeout, interval, this.logger);
    }

    /**
     * Apply `mixins` list in schema. Merge the schema with mixins schemas. Returns with the mixed schema
     *
     * @param {Schema} schema
     * @returns {Schema}
     *
     * @memberof Service
     */
    applyMixins(schema) {
        if (schema.mixins) {
            const mixins = Array.isArray(schema.mixins) ? schema.mixins : [schema.mixins];
            if (mixins.length > 0) {
                const mixedSchema = Array.from(mixins)
                    .reverse()
                    .reduce((s, mixin) => {
                        if (mixin.mixins) mixin = this.applyMixins(mixin);

                        return s ? this.mergeSchemas(s, mixin) : mixin;
                    }, null);

                return this.mergeSchemas(mixedSchema, schema);
            }
        }

        /* istanbul ignore next */
        return schema;
    }

    /**
     * Merge two Service schema
     *
     * @param {Object} mixinSchema        Mixin schema
     * @param {Object} svcSchema         Service schema
     * @returns {Object} Mixed schema
     *
     * @memberof Service
     */
    mergeSchemas(mixinSchema, svcSchema) {
        const res = _.cloneDeep(mixinSchema);
        if (!svcSchema) return res;
        const mods = _.cloneDeep(svcSchema);
        if (!mixinSchema) return mods;

        Object.keys(mods).forEach(key => {
            if ((key === "name" || key === "version") && mods[key] !== undefined) {
                // Simple overwrite
                res[key] = mods[key];
            } else if (key === "settings") {
                // Merge with defaultsDeep
                res[key] = this.mergeSchemaSettings(mods[key], res[key]);
            } else if (key === "metadata") {
                // Merge with defaultsDeep
                res[key] = this.mergeSchemaMetadata(mods[key], res[key]);
            } else if (key === "hooks") {
                // Merge & concat
                res[key] = this.mergeSchemaHooks(mods[key], res[key] || {});
            } else if (key === "actions") {
                // Merge with defaultsDeep
                res[key] = this.mergeSchemaActions(mods[key], res[key] || {});
            } else if (key === "methods") {
                // Overwrite
                res[key] = this.mergeSchemaMethods(mods[key], res[key]);
            } else if (key === "events") {
                // Merge & concat by groups
                res[key] = this.mergeSchemaEvents(mods[key], res[key] || {});
            } else if (["merged", "created", "started", "stopped"].indexOf(key) !== -1) {
                // Concat lifecycle event handlers
                res[key] = this.mergeSchemaLifecycleHandlers(mods[key], res[key]);
            } else if (key === "mixins") {
                // Concat mixins
                res[key] = this.mergeSchemaUniqArray(mods[key], res[key]);
            } else if (key === "dependencies") {
                // Concat mixins
                res[key] = this.mergeSchemaUniqArray(mods[key], res[key]);
            } else {
                const customFnName = "mergeSchema" + key.replace(/./, key[0].toUpperCase()); // capitalize first letter
                // TODO: add middleware hook
                if (isFunction(this[customFnName])) {
                    res[key] = this[customFnName](mods[key], res[key]);
                } else {
                    res[key] = this.mergeSchemaUnknown(mods[key], res[key]);
                }
            }
        });

        return res;
    }

    /**
     * Merge `settings` property in schema
     *
     * @param {Object} src Source schema property
     * @param {Object} target Target schema property
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaSettings(src, target) {
        if ((target && target.$secureSettings) || (src && src.$secureSettings)) {
            const srcSS = src && src.$secureSettings ? src.$secureSettings : [];
            const targetSS = target && target.$secureSettings ? target.$secureSettings : [];
            if (!target) target = {};

            target.$secureSettings = uniq([...srcSS, ...targetSS]);
        }

        return _.defaultsDeep(src, target);
    }

    /**
     * Merge `metadata` property in schema
     *
     * @param {Object} src Source schema property
     * @param {Object} target Target schema property
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaMetadata(src, target) {
        return _.defaultsDeep(src, target);
    }

    /**
     * Merge `mixins` property in schema
     *
     * @param {Object} src Source schema property
     * @param {Object} target Target schema property
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaUniqArray(src, target) {
        return _.uniqWith(_.compact(flatten([src, target])), _.isEqual);
    }

    /**
     * Merge `dependencies` property in schema
     *
     * @param {Object} src Source schema property
     * @param {Object} target Target schema property
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaDependencies(src, target) {
        return this.mergeSchemaUniqArray(src, target);
    }

    /**
     * Merge `hooks` property in schema
     *
     * @param {Object} src Source schema property
     * @param {Object} target Target schema property
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaHooks(src, target) {
        Object.keys(src).forEach(k => {
            if (target[k] == null) target[k] = {};

            Object.keys(src[k]).forEach(k2 => {
                const modHook = wrapToArray(src[k][k2]);
                const resHook = wrapToArray(target[k][k2]);

                target[k][k2] = _.compact(
                    flatten(k === "before" ? [resHook, modHook] : [modHook, resHook])
                );
            });
        });

        return target;
    }

    /**
     * Merge `actions` property in schema
     *
     * @param {Object} src Source schema property (real schema)
     * @param {Object} target Target schema property (mixin schema)
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaActions(src, target) {
        Object.keys(src).forEach(k => {
            if (src[k] === false && target[k]) {
                delete target[k];
                return;
            }

            const srcAction = wrapToHandler(src[k]);
            const targetAction = wrapToHandler(target[k]);

            if (srcAction && srcAction.hooks && targetAction && targetAction.hooks) {
                Object.keys(srcAction.hooks).forEach(k => {
                    const modHook = wrapToArray(srcAction.hooks[k]);
                    const resHook = wrapToArray(targetAction.hooks[k]);

                    srcAction.hooks[k] = _.compact(
                        flatten(k === "before" ? [resHook, modHook] : [modHook, resHook])
                    );
                });
            }

            target[k] = _.defaultsDeep(srcAction, targetAction);
        });

        return target;
    }

    /**
     * Merge `methods` property in schema
     *
     * @param {Object} src Source schema property
     * @param {Object} target Target schema property
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaMethods(src, target) {
        return Object.assign(target || {}, src || {});
    }

    /**
     * Merge `events` property in schema
     *
     * @param {Object} src Source schema property
     * @param {Object} target Target schema property
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaEvents(src, target) {
        Object.keys(src).forEach(k => {
            const modEvent = wrapToHandler(src[k]);
            const resEvent = wrapToHandler(target[k]);

            let handler = _.compact(
                flatten([resEvent ? resEvent.handler : null, modEvent ? modEvent.handler : null])
            );
            if (handler.length === 1) handler = handler[0];

            target[k] = _.defaultsDeep(modEvent, resEvent);
            target[k].handler = handler;
        });

        return target;
    }

    /**
     * Merge `started`, `stopped`, `created` event handler properties in schema
     *
     * @param {Object} src Source schema property
     * @param {Object} target Target schema property
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaLifecycleHandlers(src, target) {
        return _.compact(flatten([target, src]));
    }

    /**
     * Merge unknown properties in schema
     *
     * @param {Object} src Source schema property
     * @param {Object} target Target schema property
     *
     * @returns {Object} Merged schema
     */
    mergeSchemaUnknown(src, target) {
        if (src !== undefined) return src;

        return target;
    }

    /**
     * Return a versioned full service name.
     * @param {String} name
     * @param {String|Number?} version
     */
    static getVersionedFullName(name, version) {
        if (version != null)
            return (typeof version == "number" ? "v" + version : version) + "." + name;

        return name;
    }
}

module.exports = Service;