Moleculer DB Adapter - Service

db-adapter.js

"use strict";

const _ = require("lodash");
const Sequelize = require("sequelize");
const SequelizeDbAdapter = require("moleculer-db-adapter-sequelize");

/**
 * DbAdapter
 */
class DbAdapter extends SequelizeDbAdapter {
    /**
     * Constructor
     *
     * @param {any} opts Properties
     */
    constructor(opts) {
        super(opts);
        this.setOptions(opts);
    }

    /**
     * Re-Initialize Database
     */
    setOptions(opts = {}, envPrefix = null) {
        envPrefix = envPrefix ? envPrefix : _.get(process.env, "CONFIG_PREFIX", "");

        this.opts = _.defaultsDeep(opts || {}, this.opts, {
            dialect: "mysql",
            host: process.env[`${envPrefix}DB_HOST`] || "localhost",
            port: process.env[`${envPrefix}DB_PORT`] || 3306,
            username: process.env[`${envPrefix}DB_USERNAME`] || "root",
            password: process.env[`${envPrefix}DB_PASSWORD`],
            database: process.env[`${envPrefix}DB_DATABASE`],
            sync: process.env[`${envPrefix}DB_SYNC`] || true,
            logging: process.env[`${envPrefix}DB_LOGGING`] || console.log,
        });

        return this.opts;
    }

    /**
     * Connect to database
     *
     * @returns {Promise}
     *
     * @memberof SequelizeDbConnector
     */
    connect() {
        this.db = new Sequelize(this.opts);

        return Promise.resolve(this.db.authenticate()).then(() => {
            let m = this.service.schema.model;

            // Check if its an instance of a model
            if (m instanceof Sequelize.Model) {
                this.model = m;
            } else if (m instanceof Promise) {
                this.model = Promise.resolve(m.bind(this));
            } else {
                this.model = this.db.define(m.name, m.define, m.options);
            }

            this.service.model = this.model;

            // Sync
            let sync = _.get(m.options, "sync", _.get(this.opts, "sync", true));
            return sync ? this.model.sync() : Promise.resolve(this.model);
        });
    }

    /**
     * Insert Many Or Update
     * use $where and $exclude
     *
     * @param {Object} entities Attribute Options
     *
     * @returns {Promise}
     */
    async insertManyOrUpdate(entities) {
        const p = entities.map(e => this.insertOrUpdate(e, e["$where"] || null, e["$exclude"] || []));
        return Promise.all(p);
    }

    /**
     * Insert Or Update
     *
     * @param {Object} entity Attributes
     * @param {Object} where Query
     * @param {any} exclude Do not update fields
     *
     * @returns {Object<Sequelize.Model>}
     */
    async insertOrUpdate(entity, where, exclude = null) {
        // Check for where
        where = where ? where : _.get(entity, "$where", null);
        if (where === null) throw new Error("Invalid `where` parameter.");
        delete entity["$where"];

        // Check for exclude
        exclude = exclude && Array.isArray(exclude) ? exclude : _.get(entity, "$exclude", null);
        if (!Array.isArray(exclude)) exclude = [];
        delete entity["$exclude"];

        const found = await this.findOne({ where });
        const timestamps = _.has(this.service.schema.model.options, "timestamps") === true;

        if (found) {
            if (timestamps && this.service.schema.model.options.updatedAt) {
                entity[this.service.schema.model.options.updatedAt] = Date.now();
            }

            if (!_.isEmpty(exclude) && _.isString(exclude)) {
                exclude = [exclude];
            }

            return await this.updateById(found.id, {
                $set: _.omit(entity, exclude || [])
            });
        } else {
            if (timestamps && this.service.schema.model.options.createdAt) {
                entity[this.service.schema.model.options.createdAt] = Date.now();
            }

            return await this.insert(entity);
        }
    }

    /**
     * Raw queries for Sequilize
     * @param {String} query
     * @param {Object} object with options for raw query
     * @param {String} sequilize query type name
     *
     * @returns {Promise}
     * @memberof SequelizeDbConnector
     */
    raw(query, options, type) {
        const uppercased = (type) ? type.toUpperCase() : "SELECT";

        return this.db.query(query, options, Sequelize.QueryType[uppercased]);
    }

    /**
     * Get Sequilize object with connection
     *
     * @return {Sequelize}
     */
    getSequelize() {
        return this.db;
    }
}

module.exports = DbAdapter;

db-service.js

"use strict";

const _ = require("lodash");
const { wait } = require("./helpers");
let MoleculerDB = require("moleculer-db");

// DB Service/Adapter Action Names
const _dbActionNames = [
    "find", "count", "list", "create", "insert", "get", "update", "remove"
];

/**
 * Moleculer DB Service Mixin
 */
let DbService = module.exports = _.defaultsDeep({
    /**
     * Custom service settings
     */
    settings: {
        idField: "id",
        connectOnStart: true,
        connectMaxFail: 5,
        connectDelay: 1000,
        actionNames: _dbActionNames
    },

    /**
     * Mixins
     */
    mixins: [],

    /**
     * Hooks
     */
    hooks: {
        before: {
            "*": "_validateAndConnect"
        }
    },

    /**
     * Events
     */
    events: {
        "db.config.changed"(payload) {
            this.adapter.setOptions(payload);
        }
    },

    /**
     * Methods
     */
    methods: {
        /**
         * Connecting
         *
         * @returns {Promise<boolean>}
         * @private
         */
        async _connecting() {
            if (this.settings.connectOnStart === true) {
                for (let tries = this.settings.connectMaxFail; tries > -1; tries--) {
                    try {
                        await this.connect();
                        return true;
                    } catch (err) {
                        if (tries <= 0) {
                            this.logger.error(`'${this.name}' connect failed for ${this.settings.connectMaxFail} tries.`, err);
                            break;
                        } else {
                            this.adapter.setOptions();
                            this.logger.warn(`'${this.name}' connect failed. ${tries} Retrying...`);
                        }
                    }
                    await wait(this.settings.connectDelay);
                }
            }
        },

        /**
         * Validate And Connect
         * -- Checks adapter model before executing database functions
         *
         * @param ctx
         * @returns {Promise<void>}
         * @private
         */
        async _validateAndConnect(ctx) {
            const action = (ctx ? _.indexOf(this.settings.actionNames, ctx.endpoint.action.rawName) !== -1 : true) && ! this.adapter.model;
            if (action) {
                try {
                    await this._connecting();
                } catch (e) {
                    throw e;
                }
            }
        }
    },

    /**
     * Started handler
     */
    async started() {
        if (!this.adapter) {
            return Promise.reject(new Error("Please set the store adapter in schema!"));
        }
        await this._connecting();
    }

}, MoleculerDB);

/**
 * ProtectActions Mixins
 * https://moleculer.services/docs/0.13/actions.html#Action-visibility
 *
 * @param {String} visibility Values: protected, private, public. Defaults: null / published
 *
 * @returns {Object}
 */
function protectActions(visibility = null) {
    if (visibility === null || (_.isBoolean(visibility) && visibility === true)) {
        visibility = "protected";
    }
    if (_.indexOf(["protected", "private", "public", "published"], visibility) === -1) {
        visibility = "protected";
    }

    let actions = {};
    _.each(_dbActionNames, (value, key) => {
        actions[_dbActionNames[key]] = { visibility };
    });

    return { actions };
}

/**
 * Configure DB Service
 *
 * mixins: [ DbService.configure({ ... }) ]
 *
 * @param {Object} opts Attributes
 */
module.exports.configure = function(opts = {}) {
    // Properties
    _.each({

        "protectActions": (ctx) => ctx.mixins.push(protectActions())

    }, (fn, prop) => {
        delete opts[prop];
        fn(DbService);
    });

    return _.defaultsDeep(opts, DbService);
};