Ghost/core/server/adapters/scheduling/SchedulingDefault.js

const util = require('util');
const moment = require('moment');
const debug = require('ghost-ignition').debug('scheduling-default');
const SchedulingBase = require('./SchedulingBase');
const common = require('../../lib/common');
const request = require('../../lib/request');

/**
 * allJobs is a sorted list by time attribute
 */
function SchedulingDefault(options) {
    SchedulingBase.call(this, options);

    this.runTimeoutInMs = 1000 * 60 * 5;
    this.offsetInMinutes = 10;
    this.beforePingInMs = -50;
    this.retryTimeoutInMs = 1000 * 5;

    this.rescheduleOnBoot = true;
    this.allJobs = {};
    this.deletedJobs = {};
    this.isRunning = false;
}

util.inherits(SchedulingDefault, SchedulingBase);

/**
 * add to list
 */
SchedulingDefault.prototype.schedule = function (object) {
    this._addJob(object);
};

/**
 * remove from list
 * add to list
 */
SchedulingDefault.prototype.reschedule = function (object) {
    this._deleteJob({time: object.extra.oldTime, url: object.url});
    this._addJob(object);
};

/**
 * remove from list
 * deletion happens right before execution
 */
SchedulingDefault.prototype.unschedule = function (object) {
    this._deleteJob(object);
};

/**
 * check if there are new jobs which needs to be published in the next x minutes
 * because allJobs is a sorted list, we don't have to iterate over all jobs, just until the offset is too big
 */
SchedulingDefault.prototype.run = function () {
    const self = this;
    let timeout = null,
        recursiveRun;

    if (this.isRunning) {
        return;
    }

    this.isRunning = true;

    recursiveRun = function recursiveRun() {
        timeout = setTimeout(function () {
            const times = Object.keys(self.allJobs),
                nextJobs = {};

            times.every(function (time) {
                if (moment(Number(time)).diff(moment(), 'minutes') <= self.offsetInMinutes) {
                    nextJobs[time] = self.allJobs[time];
                    delete self.allJobs[time];
                    return true;
                }

                // break!
                return false;
            });

            clearTimeout(timeout);
            self._execute(nextJobs);

            recursiveRun();
        }, self.runTimeoutInMs);
    };

    recursiveRun();
};

/**
 * each timestamp key entry can have multiple jobs
 */
SchedulingDefault.prototype._addJob = function (object) {
    let timestamp = moment(object.time).valueOf(),
        keys = [],
        sortedJobs = {},
        instantJob = {},
        i = 0;

    // CASE: should have been already pinged or should be pinged soon
    if (moment(timestamp).diff(moment(), 'minutes') < this.offsetInMinutes) {
        debug('Imergency job', object.url, moment(object.time).format('YYYY-MM-DD HH:mm:ss'));

        instantJob[timestamp] = [object];
        this._execute(instantJob);
        return;
    }

    // CASE: are there jobs already scheduled for the same time?
    if (!this.allJobs[timestamp]) {
        this.allJobs[timestamp] = [];
    }

    debug('Added job', object.url, moment(object.time).format('YYYY-MM-DD HH:mm:ss'));
    this.allJobs[timestamp].push(object);

    keys = Object.keys(this.allJobs);
    keys.sort();

    for (i = 0; i < keys.length; i = i + 1) {
        sortedJobs[keys[i]] = this.allJobs[keys[i]];
    }

    this.allJobs = sortedJobs;
};

SchedulingDefault.prototype._deleteJob = function (object) {
    const {url, time} = object;

    if (!time) {
        return;
    }

    const deleteKey = `${url}_${moment(time).valueOf()}`;

    if (!this.deletedJobs[deleteKey]) {
        this.deletedJobs[deleteKey] = [];
    }

    debug('Deleted job', url, moment(time).format('YYYY-MM-DD HH:mm:ss'));
    this.deletedJobs[deleteKey].push(object);
};

/**
 * ping jobs
 * setTimeout is not accurate, but we can live with that fact and use setImmediate feature to qualify
 * we don't want to use process.nextTick, this would block any I/O operation
 */
SchedulingDefault.prototype._execute = function (jobs) {
    const keys = Object.keys(jobs),
        self = this;

    keys.forEach(function (timestamp) {
        let timeout = null,
            diff = moment(Number(timestamp)).diff(moment());

        // awake a little before
        timeout = setTimeout(function () {
            clearTimeout(timeout);

            (function retry() {
                let immediate = setImmediate(function () {
                    clearImmediate(immediate);

                    if (moment().diff(moment(Number(timestamp))) <= self.beforePingInMs) {
                        return retry();
                    }

                    const toExecute = jobs[timestamp];
                    delete jobs[timestamp];

                    toExecute.forEach(function (job) {
                        const {url, time} = job;
                        const deleteKey = `${url}_${moment(time).valueOf()}`;

                        if (self.deletedJobs[deleteKey]) {
                            if (self.deletedJobs[deleteKey].length === 1) {
                                delete self.deletedJobs[deleteKey];
                            } else {
                                self.deletedJobs[deleteKey].pop();
                            }

                            return;
                        }

                        self._pingUrl(job);
                    });
                });
            })();
        }, diff - 70);
    });
};

/**
 * - if we detect to publish a post in the past (case blog is down), we add a force flag
 */
SchedulingDefault.prototype._pingUrl = function (object) {
    const {url, time} = object;

    debug('Ping url', url, moment().format('YYYY-MM-DD HH:mm:ss'), moment(time).format('YYYY-MM-DD HH:mm:ss'));

    const httpMethod = object.extra ? object.extra.httpMethod : 'PUT';
    const tries = object.tries || 0;
    const requestTimeout = object.extra ? object.extra.timeoutInMS : 1000 * 5;
    const maxTries = 30;

    const options = {
        timeout: requestTimeout,
        method: httpMethod.toLowerCase(),
        headers: {
            'Content-Type': 'application/json'
        }
    };

    if (moment(time).isBefore(moment())) {
        if (httpMethod === 'GET') {
            // @todo: rename to searchParams when updating to Got v10
            options.query = 'force=true';
        } else {
            options.body = JSON.stringify({force: true});
        }
    }

    return request(url, options).catch((err) => {
        const {statusCode} = err;

        // CASE: post/page was deleted already
        if (statusCode === 404) {
            return;
        }

        // CASE: blog is in maintenance mode, retry
        if (statusCode === 503 && tries < maxTries) {
            setTimeout(() => {
                object.tries = tries + 1;
                this._pingUrl(object);
            }, this.retryTimeoutInMs);

            common.logging.error(new common.errors.GhostError({
                err,
                context: 'Retrying...',
                level: 'normal'
            }));

            return;
        }

        common.logging.error(new common.errors.GhostError({
            err,
            level: 'critical'
        }));
    });
};

module.exports = SchedulingDefault;