import * as parser from 'cron-parser' ; import * as dot from 'dot-object' ; import { promise as sleep } from 'es6-sleep' ; import * as moment from 'moment' ; import { Collection } from 'mongodb' ; /**
 * Configuration object interface.
 */ export interface MongoCronCfg { collection : Collection | ( ( ) => Collection ) ; condition ? : any ; nextDelay ? : number ; reprocessDelay ? : number ; idleDelay ? : number ; lockDuration ? : number ; sleepUntilFieldPath ? : string ; intervalFieldPath ? : string ; repeatUntilFieldPath ? : string ; autoRemoveFieldPath ? : string ; onDocument ? ( doc ) : ( any | Promise < any > ) ; onStart ? ( doc ) : ( any | Promise < any > ) ; onStop ? ( ) : ( any | Promise < any > ) ; onIdle ? ( ) : ( any | Promise < any > ) ; onError ? ( err ) : ( any | Promise < any > ) ; } /**
 * Main class for converting a collection into cron.
 */ export class MongoCron { protected running = false ; protected processing = false ; protected idle = false ; protected readonly config : MongoCronCfg ; param config public constructor ( config ) { this . config = { onDocument : ( doc ) => doc , onError : console . error , nextDelay : 0 , reprocessDelay : 0 , idleDelay : 0 , lockDuration : 600000 , sleepUntilFieldPath : 'sleepUntil' , intervalFieldPath : 'interval' , repeatUntilFieldPath : 'repeatUntil' , autoRemoveFieldPath : 'autoRemove' , ... config , } ; } /**
     * Returns the collection instance (the collection can be provided in
     * the config as an instance or a function).
     */ protected getCollection ( ) { return typeof this . config . collection === 'function' ? this . config . collection ( ) : this . config . collection ; } /**
     * Tells if the process is started.
     */ public isRunning ( ) { return this . running ; } /**
     * Tells if a document is processing.
     */ public isProcessing ( ) { return this . processing ; } /**
     * Tells if the process is idle.
     */ public isIdle ( ) { return this . idle ; } /**
     * Starts the heartbit.
     */ public async start ( ) { if ( ! this . running ) { this . running = true ; if ( this . config . onStart ) { await this . config . onStart . call ( this , this ) ; } process . nextTick ( this . tick . bind ( this ) ) ; } } /**
     * Stops the heartbit.
     */ public async stop ( ) { this . running = false ; if ( this . processing ) { await sleep ( 300 ) ; return process . nextTick ( this . stop . bind ( this ) ) ; } if ( this . config . onStop ) { await this . config . onStop . call ( this , this ) ; } } /**
     * Private method which runs the heartbit tick.
     */ protected async tick ( ) { if ( ! this . running ) { return ; } await sleep ( this . config . nextDelay ) ; if ( ! this . running ) { return ; } this . processing = true ; try { const doc = await this . lockNext ( ) ; if ( ! doc ) { this . processing = false ; if ( ! this . idle ) { this . idle = true ; if ( this . config . onIdle ) { await this . config . onIdle . call ( this , this ) ; } } await sleep ( this . config . idleDelay ) ; } else { this . idle = false ; if ( this . config . onDocument ) { await this . config . onDocument . call ( this , doc , this ) ; } await this . reschedule ( doc ) ; this . processing = false ; } } catch ( err ) { await this . config . onError . call ( this , err , this ) ; } process . nextTick ( ( ) => this . tick ( ) ) ; } /**
     * Locks the next job document for processing and returns it.
     */ protected async lockNext ( ) { const sleepUntil = moment ( ) . add ( this . config . lockDuration , 'milliseconds' ) . toDate ( ) ; const currentDate = moment ( ) . toDate ( ) ; const res = await this . getCollection ( ) . findOneAndUpdate ( { $and : [ { [ this . config . sleepUntilFieldPath ] : { $exists : true , $ne : null } } , { [ this . config . sleepUntilFieldPath ] : { $not : { $gt : currentDate } } } , this . config . condition , ] . filter ( ( c ) => ! ! c ) , } , { $set : { [ this . config . sleepUntilFieldPath ] : sleepUntil } , } , { returnOriginal : true , } ) ; return res . value ; } param doc protected getNextStart ( doc ) { if ( ! dot . pick ( this . config . intervalFieldPath , doc ) ) { return null ; } const available = moment ( dot . pick ( this . config . sleepUntilFieldPath , doc ) ) ; const future = moment ( available ) . add ( this . config . reprocessDelay , 'milliseconds' ) ; try { const interval = parser . parseExpression ( dot . pick ( this . config . intervalFieldPath , doc ) , { currentDate : future . toDate ( ) , endDate : dot . pick ( this . config . repeatUntilFieldPath , doc ) , } ) ; const next = interval . next ( ) . toDate ( ) ; const now = moment ( ) . toDate ( ) ; return next < now ? now : next ; } catch ( err ) { return null ; } } param doc public async reschedule ( doc ) { const nextStart = this . getNextStart ( doc ) ; const _id = doc . _id ; if ( ! nextStart && dot . pick ( this . config . autoRemoveFieldPath , doc ) ) { await this . getCollection ( ) . deleteOne ( { _id } ) ; } else if ( ! nextStart ) { await this . getCollection ( ) . updateOne ( { _id } , { $set : { [ this . config . sleepUntilFieldPath ] : null } , } ) ; } else { await this . getCollection ( ) . updateOne ( { _id } , { $set : { [ this . config . sleepUntilFieldPath ] : nextStart } , } ) ; } } }