<commit_before>const Promise = require('bluebird')
const EventEmitter = require('events')
const plans = require('./plans')
const cronParser = require('cron-parser')
const Attorney = require('./attorney')

const queues = {
  CRON: '__pgboss__cron',
  SEND_IT: '__pgboss__send-it'
}

const events = {
  error: 'error',
  schedule: 'schedule'
}

class Timekeeper extends EventEmitter {
  constructor (db, config) {
    super()

    this.db = db
    this.config = config
    this.manager = config.manager
    this.monitorIntervalMs = config.clockMonitorIntervalSeconds * 1000
    this.clockSkew = 0

    this.events = events

    this.getTimeCommand = plans.getTime(config.schema)
    this.getSchedulesCommand = plans.getSchedules(config.schema)
    this.scheduleCommand = plans.schedule(config.schema)
    this.unscheduleCommand = plans.unschedule(config.schema)

    this.functions = [
      this.schedule,
      this.unschedule,
      this.getSchedules
    ]
  }

  async start () {
    await this.cacheClockSkew()
    await this.watch()

    this.monitorInterval = setInterval(() => this.cacheClockSkew(), this.monitorIntervalMs)
    this.stopped = false
  }

  async stop () {
    if (!this.stopped) {
      this.stopped = true

      if (this.watching) {
        await this.unwatch()
      }

      if (this.monitorInterval) {
        clearInterval(this.monitorInterval)
        this.monitorInterval = null
      }
    }
  }

  async cacheClockSkew () {
    const start = Date.now()

    const { rows } = await this.db.executeSql(this.getTimeCommand)

    const end = Date.now()

    const latency = end - start

    const dbTime = Math.round(parseFloat(rows[0].time) - (latency / 2))

    const skew = dbTime - start

    const skewSeconds = Math.abs(skew) / 1000

    if (skewSeconds >= 60 || this.config.__test__force_clock_skew_warning) {
      Attorney.warnClockSkew(`Instance clock is ${skewSeconds}s ${skew > 0 ? 'slower' : 'faster'} than database.`)
    }

    this.clockSkew = skew
  }

  async watch () {
    await this.manager.subscribe(queues.CRON, (job) => this.onCron(job))
    await this.manager.subscribe(queues.SEND_IT, { teamSize: 50, teamConcurrency: 5 }, (job) => this.onSendIt(job))

    await this.cronMonitorAsync()

    this.watching = true
  }

  async unwatch () {
    await this.manager.unsubscribe(queues.CRON)
    await this.manager.unsubscribe(queues.SEND_IT)

    this.watching = false
  }

  async cronMonitorAsync (options = {}) {
    const { startAfter } = options

    const opts = {
      singletonKey: queues.CRON,
      singletonSeconds: 60,
      startAfter,
      retryLimit: 2
    }

    await this.manager.publish(queues.CRON, null, opts)
  }

  async onCron () {
    if (this.stopped || !this.watching) {
      return
    }

    try {
      if (this.config.__test__throw_clock_monitoring) {
        throw new Error('clock monitoring error')
      }

      const items = await this.getSchedules()

      const sending = items.filter(i => this.shouldSendIt(i.cron, i.timezone))

      if (sending.length) {
        await Promise.map(sending, it => this.send(it), { concurrency: 5 })
      }
    } catch (err) {
      this.emit(this.events.error, err)
    }

    await this.cronMonitorAsync({ startAfter: 30 })
  }

  shouldSendIt (cron, tz) {
    const interval = cronParser.parseExpression(cron, { tz })

    const prevTime = interval.prev()

    const databaseTime = Date.now() + this.clockSkew

    const prevDiff = (databaseTime - prevTime.getTime()) / 1000

    return prevDiff < 60
  }

  async send (job) {
    const options = {
      singletonKey: job.name,
      singletonSeconds: 60
    }
<commit_msg>fix throttline in cron handler<commit_after>const Promise = require('bluebird')
const EventEmitter = require('events')
const plans = require('./plans')
const cronParser = require('cron-parser')
const Attorney = require('./attorney')

const queues = {
  CRON: '__pgboss__cron',
  SEND_IT: '__pgboss__send-it'
}

const events = {
  error: 'error',
  schedule: 'schedule'
}

class Timekeeper extends EventEmitter {
  constructor (db, config) {
    super()

    this.db = db
    this.config = config
    this.manager = config.manager
    this.monitorIntervalMs = config.clockMonitorIntervalSeconds * 1000
    this.clockSkew = 0

    this.events = events

    this.getTimeCommand = plans.getTime(config.schema)
    this.getSchedulesCommand = plans.getSchedules(config.schema)
    this.scheduleCommand = plans.schedule(config.schema)
    this.unscheduleCommand = plans.unschedule(config.schema)

    this.functions = [
      this.schedule,
      this.unschedule,
      this.getSchedules
    ]
  }

  async start () {
    await this.cacheClockSkew()
    await this.watch()

    this.monitorInterval = setInterval(() => this.cacheClockSkew(), this.monitorIntervalMs)
    this.stopped = false
  }

  async stop () {
    if (!this.stopped) {
      this.stopped = true

      if (this.monitorInterval) {
        clearInterval(this.monitorInterval)
        this.monitorInterval = null
      }
    }
  }

  async cacheClockSkew () {
    const start = Date.now()

    const { rows } = await this.db.executeSql(this.getTimeCommand)

    const end = Date.now()

    const latency = end - start

    const dbTime = Math.round(parseFloat(rows[0].time) - (latency / 2))

    const skew = dbTime - start

    const skewSeconds = Math.abs(skew) / 1000

    if (skewSeconds >= 60 || this.config.__test__force_clock_skew_warning) {
      Attorney.warnClockSkew(`Instance clock is ${skewSeconds}s ${skew > 0 ? 'slower' : 'faster'} than database.`)
    }

    this.clockSkew = skew
  }

  async watch () {
    await this.manager.subscribe(queues.CRON, (job) => this.onCron(job))
    await this.manager.subscribe(queues.SEND_IT, { teamSize: 50, teamConcurrency: 5 }, (job) => this.onSendIt(job))

    await this.cronMonitorAsync()
  }

  async cronMonitorAsync (options = {}) {
    const { startAfter } = options

    const opts = {
      singletonKey: queues.CRON,
      singletonSeconds: 60,
      retentionSeconds: 60,
      startAfter,
      retryLimit: 2
    }

    await this.manager.publish(queues.CRON, null, opts)
  }

  async onCron (job) {
    if (this.stopped) {
      return
    }

    try {
      if (this.config.__test__throw_clock_monitoring) {
        throw new Error('clock monitoring error')
      }

      const items = await this.getSchedules()

      const sending = items.filter(i => this.shouldSendIt(i.cron, i.timezone))

      if (sending.length) {
        await Promise.map(sending, it => this.send(it), { concurrency: 5 })
      }
    } catch (err) {
      this.emit(this.events.error, err)
    }

    if (!this.stopped) {
      await job.done() // pre-complete to bypass throttling
      await this.cronMonitorAsync({ startAfter: 30 })
    }
  }

  shouldSendIt (cron, tz) {
    const interval = cronParser.parseExpression(cron, { tz })

    const prevTime = interval.prev()

    const databaseTime = Date.now() + this.clockSkew

    const prevDiff = (databaseTime - prevTime.getTime()) / 1000

    return prevDiff < 60
  }

  async send (job) {
    const options = {
      singletonKey: job.name,
      singletonSeconds: 60
    }

{'fraction_non_alphanumeric': 0.1196438508625487, 'fraction_numerical': 0.008903728436282694, 'mean_word_length': 2.8137931034482757, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3171659', 'n_tokens_mistral': 2435, 'n_tokens_neox': 2298, 'n_words': 494}