import { JobStore , RateLimitState , RateLimitStrategy , RetryStrategy , Step , StepPrioritizer , StepResult } from "./api" ; import Logger from "bunyan" ; /**
 * Information about the next action that should be taken after a step has been processed.
 */ export interface NextAction < JOB_ID > { /**
     * Identifier for the job.
     */ jobId : JOB_ID ; /**
     * If this is true, the job isn't finished, yet. Go ahead and schedule the next step of the job.
     * If this is false, the job is finished.
     */ scheduleNextStep : boolean ; /**
     * If this is set, the next step should be delayed.
     */ delay ? : { seconds : number ; reason : "retry" | "rate-limit" ; } ; } export class Looper < JOB_ID , JOB_STATE > { private readonly prioritizer : StepPrioritizer < JOB_ID , JOB_STATE > ; private readonly jobStore : JobStore < JOB_ID , JOB_STATE > ; private readonly rateLimitStrategy : RateLimitStrategy ; private readonly retryStrategy : RetryStrategy ; private readonly logger : Logger ; constructor ( prioritizer , jobStore , rateLimitStrategy , retryStrategy , logger ) { this . prioritizer = prioritizer ; this . jobStore = jobStore ; this . rateLimitStrategy = rateLimitStrategy ; this . retryStrategy = retryStrategy ; this . logger = logger ; } /**
     * Processes the next step of the given job. This method should be called again and again for the same job until
     * the return NextAction.scheduleNextStep is false. If NextAction.delayInSeconds is set, don't call the method
     * for the same job again before this number of seconds has passed to honor rate limits.
     */ public processStep ( step ) { const currentRateLimitState = this . jobStore . getRateLimitState ( step . jobId ) ; if ( this . rateLimitStrategy . getDelayInSeconds ( currentRateLimitState ) > 0 ) { this . logger . info ( "RATE LIMIT HIT!" ) ; return this . continueJobWithRateLimitDelay ( step , currentRateLimitState ) ; } const jobState = this . jobStore . getJobState ( step . jobId ) ; const stepProcessor = this . prioritizer . getStepProcessor ( step , jobState , currentRateLimitState ) ; if ( ! stepProcessor ) { return this . stopJob ( step ) ; } const stepResult = stepProcessor . process ( jobState , currentRateLimitState ) ; this . updateRateLimitState ( stepResult , step ) ; this . jobStore . setJobState ( step . jobId , stepResult . jobState ) ; if ( stepResult . success ) { return this . continueJobWithRateLimitDelay ( step , stepResult . rateLimit ) ; } if ( ! stepResult . error || stepResult . error ?. isRetryable ) { return this . retryStep ( step , stepResult , jobState ) ; } else if ( stepResult . error . isFatal ) { this . logger . error ( `Stopping job ${ step . jobId } due to error: ${ stepResult . error ?. message }` ) ; return this . stopJob ( step ) ; } else if ( ! stepResult . error . isRetryable ) { this . logger . error ( `Skipping step ${ JSON . stringify ( step ) } due to error: ${ stepResult . error ?. message }` ) ; this . skipStep ( step , jobState , stepResult ) ; } throw new Error ( `Job ${ step . jobId } ran into an invalid error state: ${ JSON . stringify ( stepResult . error ) }` ) ; } private skipStep ( step , jobState , stepResult ) { const updatedJobState = this . prioritizer . skip ( step , jobState , stepResult . rateLimit ) ; this . jobStore . setJobState ( step . jobId , updatedJobState ) ; return this . continueJobWithRateLimitDelay ( step , stepResult . rateLimit ) ; } /**
     * If the RetryStrategy determines the step should be retried, increment the failed attempts counter and return an
     * action to continue the job.
     *
     * If the retry limit has been reached, skip the step.
     */ private retryStep ( step , stepResult , jobState ) { const failedAttempts = this . jobStore . getFailedAttemptsCount ( step . jobId ) + 1 ; const retry = this . retryStrategy . getRetry ( failedAttempts ) ; if ( retry . shouldRetry ) { this . logger . warn ( `Retrying step ${ JSON . stringify ( step ) } due to error: ${ stepResult . error ?. message }` ) ; this . jobStore . setFailedAttemptsCount ( step . jobId , failedAttempts ) ; return this . continueJobWithRetryDelay ( step , retry . retryAfterSeconds ) ; } else { this . logger . warn ( `Not retrying step ${ JSON . stringify ( step ) } after ${ failedAttempts } failed attempts. Error was: ${ stepResult . error ?. message }` ) ; this . jobStore . setFailedAttemptsCount ( step . jobId , 0 ) ; return this . skipStep ( step , jobState , stepResult ) ; } return this . continueJobWithRateLimitDelay ( step , stepResult . rateLimit ) ; } private continueJobWithRetryDelay ( step , delayInSeconds ? ) { return { jobId : step . jobId , scheduleNextStep : true , delay : ! delayInSeconds ? undefined : { seconds : delayInSeconds , reason : "retry" } } ; } private continueJobWithRateLimitDelay ( step , rateLimitState ? ) { const delayInSeconds = this . rateLimitStrategy . getDelayInSeconds ( rateLimitState ) ; return { jobId : step . jobId , scheduleNextStep : true , delay : ! rateLimitState || delayInSeconds == 0 ? undefined : { seconds : delayInSeconds , reason : "rate-limit" } } ; } private stopJob ( step ) { return { jobId : step . jobId , scheduleNextStep : false , } ; } private updateRateLimitState ( stepResult , step ) { if ( stepResult . rateLimit ) { this . jobStore . updateRateLimitState ( step . jobId , stepResult . rateLimit ) ; } } }