'use strict';

const config = require('./config/config');

const _ = require('lodash');
const path = require('path');
const cluster = require('cluster');
const logger = require('log4js').getLogger('app');
const models = require('./models');

// Таймер для проверки количества воркеров и остановки мастера
let shutdownInterval = null;

if (cluster.isMaster) {
    // Запускаем воркеры
    _.each(config.workers, (conf, name) => {
        if (conf.enabled) {
            startWorker(name);
        }
    });
} else {
    // Инициализируем воркер
    let name = process.env.WORKER_NAME;
    let WorkerClass = require(path.join(__dirname, 'workers', name + '.js'));
    let worker = null;
    if (WorkerClass) {
        worker = new WorkerClass(name, config.workers[name]);
        // Запускаем воркер
        worker.start();
        // Подписываемся на событие, когда воркер остановлен
        worker.on('stop', () => {
            process.exit();
        });
    }
    // Подписываемся на события от мастера
    process.on('message', message => {
        if ('shutdown' === message) {
            if (worker) {
                worker.stop();
            } else {
                process.exit();
            }
        }
    });
}

// Shutdown
process.on('SIGTERM', shutdownCluster);
process.on('SIGINT', shutdownCluster);

// Метод запуска воркера
function startWorker(name) {
    let worker = cluster.fork({WORKER_NAME: name}).on('online', () => {
        logger.info('Start %s worker #%d.', name, worker.id);
    }).on('exit', status => {
        // Когда воркер был остановлен
        if ((worker.exitedAfterDisconnect || worker.suicide) === true || status === 0) {
            // Если воркер был остановлен контролируемо, то ни чего не делаем
            logger.info('Worker %s #%d was killed.', name, worker.id);
        } else {
            // Если воркер был остановлен неконтролируемо, то запускаем его еще раз
            logger.warn('Worker %s #%d was died. Replace it with a new one.', name, worker.id);
            startWorker(name);
        }
    });
}

// Метод остановки кластера
function shutdownCluster() {
    if (cluster.isMaster) {
        clearInterval(shutdownInterval);
        if (_.size(cluster.workers) > 0) {
            // Посылаем сигнал останова каждому воркеру
            logger.info('Shutdown workers:', _.size(cluster.workers));
            _.each(cluster.workers, worker => {
                try {
                    worker.send('shutdown');
                } catch (err) {
                    logger.warn('Cannot send shutdown message to worker:', err);
                }
            });
            // Ожидаем останов всех воркеров
            shutdownInterval = setInterval(() => {
                if (_.size(cluster.workers) === 0) {
                    process.exit();
                }
            }, config.shutdownInterval);
        } else {
            process.exit();
        }
    }
}

'use strict';

const _ = require('lodash');
const Promise = require('bluebird');
const log4js = require('log4js');
const EventEmitter = require('events');
const WorkerStates = require('./worker_states');

class Worker extends EventEmitter {

    constructor(name, conf) {
        super();
        this.name = name;
        // Значения настроек по умолчанию
        this.conf = _.defaults({}, conf, {
            sleep: 1000 // Задержка между запусками
        });
        this.logger = log4js.getLogger('worker-' + name);
        // Флаг останова воркера
        this.stopped = true;
        // Таймер для задержки между запусками loop мотода
        this.timer = null;
        // Состояние воркера
        this.state = null;
    }

    // Метод запуска воркера
    start() {
        this.logger.info('Start');
        this.stopped = false;
        this.state = WorkerStates.STATE_IDLE;
        return this._startLoop();
    }

    // Метод останова воркера
    stop() {
        this.logger.info('Stop');
        this.stopped = true;
        if (this.state === WorkerStates.STATE_IDLE) {
            // Останавливаем таймер задержки
            if (this.timer) {
                clearTimeout(this.timer);
                this.timer = null;
            }
            this.state = WorkerStates.STATE_STOP;
            // Вызываем событие останова воркера
            this.emit('stop');
        }
    }

    // Метод для выполнения бизнес задач
    loop() {
        return Promise.resolve();
    }

    // Метод запуска и обработки loop метода
    _startLoop() {
        this.state = WorkerStates.STATE_WORK;
        return this.loop().catch(err => {
            this.logger.warn('Loop error:', err);
        }).finally(() => {
            this.state = WorkerStates.STATE_IDLE;
            if (!this.stopped) {
                // Делаем повторный запуск по окончании работы loop метода
                this.timer = setTimeout(() => {
                    this._startLoop();
                }, this.conf.sleep);
            } else {
                this.state = WorkerStates.STATE_STOP;
                // Вызываем событие останова воркера
                this.emit('stop');
            }
        });
    }

}

module.exports = Worker;

'use strict';

const config = require('../config/config');

const _ = require('lodash');
const Promise = require('bluebird');
const Worker = require('./worker');
const WorkerStates = require('./worker_states');
const models = require('../models');

class TaskWorker extends Worker {

    constructor(name, conf) {
        super(name, conf);
        // Значения настроек по умолчанию
        this.conf = _.defaults({}, this.conf, {
            maxAttempts: 3, // Максимальное количество попыток запуска задачи
            delayRatio: 300000, // Базовое значение отсрочки запуска задачи
            count: 1, // Максимальное количество одновременно обрабатываемых задач
            queue: '', // Название очереди для получения задач
            update: 3000 // Интервал обновления статуса задачи
        });
        // Счетчик одновременно обрабатываемых задач
        this.count = 0;
    }

    loop() {
        if (this.count < this.conf.count && !this.stopped) {
            // Получение очередной задачи
            return this._getTask().then(task => {
                if (task) {
                    // Увеличение счетчика одновременно обрабатываемых задач
                    this.count++;
                    // Запуск периодического обновления статуса задачи
                    let interval = setInterval(() => {
                        return models.sequelize.transaction(t => {
                            return task.touch({transaction: t});
                        });
                    }, this.conf.update);
                    // Запуск метода обработки задачи
                    this.handleTask(task.get({plain: true})).then(() => {
                        // Завершение задачи
                        return models.sequelize.transaction(t => {
                            return task.complete({transaction: t}).then(() => {
                                this.logger.info('Task completed:', task.id);
                            });
                        });
                    }).catch(err => {
                        // Задача не была выполнена - перезапуск задачи
                        this.logger.warn('Handle error:', err);
                        return this.delay(task).then(delay => {
                            return models.sequelize.transaction(t => {
                                return task.fail(delay, {transaction: t}).then(() => {
                                    this.logger.warn('Task failed:', task.id);
                                });
                            });
                        });
                    }).finally(() => {
                        clearInterval(interval);
                        this.count--;
                    }).done();
                    return null;
                }
            });
        } else {
            return Promise.resolve();
        }
    }

    // Метод обработки задачи
    handleTask() {
        return Promise.resolve();
    }

    // Метод вычисления отсрочки задачи после неуспешного запуска
    delay(task) {
        return Promise.resolve().then(() => {
            return task.attempts * this.conf.delayRatio;
        });
    }

    // Метод получения задачи для обработки
    _getTask() {
        return models.sequelize.transaction({autocommit: false}, t => {
            return models.Task.scope({
                method: ['forWork', this.conf.queue, config.node_id]
            }).find({transaction: t, lock: t.LOCK.UPDATE}).then(task => {
                if (task) {
                    return task.work(config.node_id, {transaction: t});
                }
            });
        });
    }

    _startLoop() {
        this.state = WorkerStates.STATE_WORK;
        return this.loop().catch(err => {
            this.logger.warn('Loop error:', err);
        }).finally(() => {
            if (this.count === 0) {
                this.state = WorkerStates.STATE_IDLE;
            }
            if (this.stopped && this.count === 0) {
                this.state = WorkerStates.STATE_STOP;
                this.emit('stop');
            } else {
                this.timer = setTimeout(() => {
                    this._startLoop();
                }, this.conf.sleep);
            }
        });
    }

}

module.exports = TaskWorker;

'use strict';

const moment = require('moment');

module.exports = function(sequelize, Sequelize) {

    return sequelize.define('Task', {
        id: {
            type: Sequelize.INTEGER,
            primaryKey: true,
            autoIncrement: true
        },
        node_id: {
            type: Sequelize.INTEGER
        },
        queue: {
            type: Sequelize.STRING,
            allowNull: false
        },
        status: {
            type: Sequelize.ENUM,
            values: ['pending', 'working', 'done', 'failure'],
            defaultValue: 'pending',
            allowNull: false
        },
        attempts: {
            type: Sequelize.INTEGER,
            defaultValue: 0,
            allowNull: false
        },
        priority: {
            type: Sequelize.INTEGER,
            defaultValue: 10,
            allowNull: false
        },
        body: {
            type: Sequelize.TEXT,
            set: function(body) {
                return this.setDataValue('body', JSON.stringify(body));
            },
            get: function() {
                try {
                    return JSON.parse(this.getDataValue('body'));
                } catch (e) {
                    return null;
                }
            }
        },
        start_at: {
            type: Sequelize.DATE
        },
        finish_at: {
            type: Sequelize.DATE
        },
        worker_node_id: {
            type: Sequelize.INTEGER
        },
        worker_started_at: {
            type: Sequelize.DATE
        },
        checked_at: {
            type: Sequelize.DATE
        }
    }, {
        tableName: 'tasks',
        freezeTableName: true,
        underscored: true,

        scopes: {
            forWork: function(queue, node_id) {
                return {
                    where: {
                        node_id: {
                            $or: [
                                null,
                                node_id
                            ]
                        },
                        queue: queue,
                        status: 'pending',
                        start_at: {
                            $or: [
                                null,
                                {
                                    $lt: moment().toDate()
                                }
                            ]
                        },
                        finish_at: {
                            $or: [
                                null,
                                {
                                    $gte: moment().toDate()
                                }
                            ]
                        }
                    },
                    order: [
                        ['priority', 'DESC'],
                        ['attempts', 'ASC'],
                        [sequelize.fn('IFNULL', sequelize.col('start_at'), sequelize.col('created_at')), 'ASC']
                    ]
                };
            }
        },

        instanceMethods: {
            fail: function(delay, options) {
                this.start_at = delay ? moment().add(delay, 'ms').toDate() : null;
                this.attempts = sequelize.literal('attempts + 1');
                this.status = 'failure';
                return this.save(options);
            },
            complete: function(options) {
                this.status = 'done';
                return this.save(options);
            },
            work: function(node_id, options) {
                this.status = 'working';
                this.worker_node_id = node_id;
                this.worker_started_at = moment().toDate();
                return this.save(options);
            },
            check: function(options) {
                this.checked_at = moment().toDate();
                return this.save(options);
            }
        }
    });

};

'use strict';

const _ = require('lodash');
const moment = require('moment');
const Promise = require('bluebird');
const Worker = require('../components/worker');
const models = require('../models');
const config = require('../config/config');

class Manager extends Worker {

    constructor(name, conf) {
        super(name, conf);
        this.conf = _.defaults({}, this.conf, {
            maxUpdate: 30000,       // 30 seconds
            maxCompleted: 3600000,  // 1 hour
            maxFailed: 259200000    // 3 days
        });
    }

    loop() {
        return models.sequelize.transaction(t => {
            return Promise.resolve()
                .then(() => {
                    return this._checkCurrentNode(t);
                })
                .then(() => {
                    return this._activateNodes(t);
                })
                .then(() => {
                    return this._pauseNodes(t);
                })
                .then(() => {
                    return this._restoreFrozenTasks(t);
                })
                .then(() => {
                    return this._restoreFailedTasks(t);
                })
                .then(() => {
                    return this._deleteDeadTasks(t);
                })
                .then(() => {
                    return this._deleteCompletedTasks(t);
                })
                .then(() => {
                    return this._deleteFailedTasks(t);
                });
        });
    }

    _checkCurrentNode(t) {
        return models.Node.findById(config.node_id, {transaction: t}).then(node => {
            if (node) {
                return node.check();
            }
        });
    }

    _activateNodes(t) {
        return models.Node.update({
            is_active: true
        }, {
            where: {
                is_active: false,
                checked_at: {
                    $gte: moment().subtract(2 * this.conf.sleep).toDate()
                }
            },
            transaction: t
        }).spread(count => {
            if (count > 0) {
                this.logger.info('Activate nodes:', count);
            }
        });
    }

    _pauseNodes(t) {
        return models.Node.update({
            is_active: false
        }, {
            where: {
                is_active: true,
                checked_at: {
                    $lt: moment().subtract(2 * this.conf.sleep).toDate()
                }
            },
            transaction: t
        }).spread(count => {
            if (count > 0) {
                this.logger.info('Pause nodes:', count);
            }
        });
    }

    _restoreFrozenTasks(t) {
        return models.Task.update({
            status: 'failure',
            attempts: models.sequelize.literal('attempts + 1')
        }, {
            where: {
                status: 'working',
                checked_at: {
                    $lt: moment().subtract(this.conf.maxUpdate).toDate()
                }
            },
            transaction: t
        }).spread(count => {
            if (count > 0) {
                this.logger.info('Restore frozen tasks:', count);
            }
        });
    }

    _restoreFailedTasks(t) {
        let where = [{status: 'failure'}];
        let conditions = this._failedTasksConditions();
        if (conditions.length) {
            where.push({$or: conditions});
        }

        return models.Task.update({
            status: 'pending',
            worker_node_id: null,
            worker_started_at: null
        }, {
            where: where,
            transaction: t
        }).spread(count => {
            if (count > 0) {
                this.logger.info('Restore failure tasks:', count);
            }
        });
    }

    _deleteDeadTasks(t) {
        return models.Task.destroy({
            where: {
                status: 'pending',
                finish_at: {
                    $lt: moment().toDate()
                }
            },
            transaction: t
        }).then(count => {
            if (count > 0) {
                this.logger.info('Delete dead tasks:', count);
            }
        });
    }

    _deleteCompletedTasks(t) {
        return models.Task.destroy({
            where: {
                status: 'done',
                checked_at: {
                    $lt: moment().subtract(this.conf.maxCompleted).toDate()
                }
            },
            transaction: t
        }).then(count => {
            if (count > 0) {
                this.logger.info('Delete completed tasks:', count);
            }
        });
    }

    _deleteFailedTasks(t) {
        let where = [
            {status: 'failure'},
            {checked_at: {
                $lt: moment().subtract(this.conf.maxFailed).toDate()
            }}
        ];
        let conditions = this._failedTasksConditions();
        if (conditions.length) {
            where.push({$or: conditions});
        }
        return models.Task.destroy({
            where: where,
            transaction: t
        }).then(count => {
            if (count > 0) {
                this.logger.info('Delete failed tasks:', count);
            }
        });
    }

    _failedTasksConditions() {
        let conditions = [];
        _.each(config.workers, (worker) => {
            if (worker.queue) {
                let item = {queue: worker.queue};
                if (worker.maxAttempts !== undefined) {
                    item.attempts = {
                        $lt: worker.maxAttempts
                    };
                }
                conditions.push(item);
            }
        });
        return conditions;
    }

}

module.exports = Manager;