1import { ptDiskDir } from "../../serverF/ptDiskF/ptDiskF.mts"
2import fs from 'fs'
3import { sleep } from "../../sharedF/sleepF.mts"
4import * as _ from 'lodash-es'
5import { fs1Promises } from "../../serverF/libFs/fsPromisesF.mts"
6import { json1ParseCatch } from '../../sharedF/jsonF.mts'
7import { secToHumanAbs } from '../../sharedF/langF/dur2HumanF.mts'
8import { debug1Sleep } from '../../serverF/sleepF/debugSleepF.mts'
9import { parseCli, cliHelpText } from '../../serverF/cliArgAI.mts'
10import { chalkGreen, chalkGray, chalkCyan, chalkRed, chalkBold, chalkDim } from '../../serverF/libChalkF.mts'
11
12type ZhongJob = { name: string; runFnc: (lastData: unknown) => Promise<unknown>; intervalSec?: number; runInPoll?: boolean }
13
14export const zhongLongRun = async ({jobsA, pollIntervalSec=10}: {jobsA: ZhongJob[], pollIntervalSec?: number}) => {
15  while (true) {
16    const ranJobNames = await zhongRunOnce({jobsA, withinLongRun: true})
17    jobsA = _.reject(jobsA, ({runInPoll}) => runInPoll === false)
18    if (ranJobNames.length > 0) {
19      await debug1Sleep(pollIntervalSec * 1000, {ranJobNames})
20    } else {
21      await sleep(pollIntervalSec * 1000)
22    }
23  }
24}
25
26export const zhongInfo = ({jobsA}: {jobsA: (ZhongJob | undefined)[]}) => {
27  const zhongRunsDir = ptDiskDir + '/zhongruns/'
28  for (const jobH of _.compact(jobsA)) {
29    const {name, intervalSec} = jobH
30    const lastRunFile = zhongRunsDir + name
31    let lastRunStr = 'never'
32    let wouldRun = true
33    if (fs.existsSync(lastRunFile)) {
34      const stats = fs.statSync(lastRunFile)
35      const elapsedSec = (Date.now() - stats.mtimeMs) / 1000
36      lastRunStr = secToHumanAbs(elapsedSec) + ' ago'
37      if (intervalSec) {
38        wouldRun = elapsedSec >= intervalSec
39      }
40    }
41    const intervalStr = intervalSec ? secToHumanAbs(intervalSec) : 'on-demand'
42    const coloredInterval = wouldRun ? chalkGreen(intervalStr) : chalkGray(intervalStr)
43    console.log(name.padEnd(40), lastRunStr.padEnd(14), coloredInterval)
44  }
45}
46
47export const eptZhong = async ({jobsA}: {jobsA: (ZhongJob | undefined)[]}) => {
48  const cliSchema = { ignoreRecent: { flag: true, desc: 'run job regardless of interval throttling' }, retrySkipped: { flag: true, desc: 'retry skip-listed import sessions' } } as const
49  const cliActions = {
50    info: { desc: 'list jobs and last-run times without running any' },
51    help: { desc: 'show this help' },
52  }
53  const { ignoreRecent, retrySkipped, _action } = parseCli(cliSchema, cliActions)
54  if (_action === 'help') {
55    console.log(cliHelpText('', cliSchema, cliActions))
56    return
57  }
58  if (_action === 'info') {
59    zhongInfo({jobsA})
60    return
61  }
62  let jobs = _.compact(jobsA)
63  if (_action) {
64    jobs = _.filter(jobs, ({name}) => name == _action)
65  }
66  await zhongRunOnce({jobsA: jobs, ignoreRecent})
67}
68
69export const zhongRunOnce = async ({jobsA, withinLongRun=false, ignoreRecent=false}: {jobsA: ZhongJob[], withinLongRun?: boolean, ignoreRecent?: boolean}) => {
70  jobsA = _.compact(jobsA) // in case someone wants to comment out jobs w/ "false &&" at higher level
71  const ranJobNames = []
72  let throttledAny = false
73  for (const ii in jobsA) {
74    const jobH = jobsA[ii]
75    const {name, runFnc, intervalSec} = jobH
76  
77    const zhongRunsDir = ptDiskDir + '/zhongruns/'
78    const lastRunFile = zhongRunsDir + name
79  
80    let lastRunData = null
81    if (fs.existsSync(lastRunFile)) {
82      lastRunData = json1ParseCatch(fs.readFileSync(lastRunFile, 'utf8'))
83      if (intervalSec) {
84        const stats = fs.statSync(lastRunFile)
85        const elapsedSec = (Date.now() - stats.mtimeMs) / 1000
86        const isRecent = elapsedSec < intervalSec
87        if (!withinLongRun) {
88          console.log(chalkDim(`${name} { isRecent: ${isRecent} } ${secToHumanAbs(elapsedSec)} / ${secToHumanAbs(intervalSec)}`))
89        }
90        if (isRecent && !ignoreRecent) {
91          throttledAny = true
92          continue
93        }  
94      }
95    }
96  
97    console.log(chalkBold(chalkCyan('▸')), name)
98    const startedAt = _.now()
99    let err
100    try {
101      lastRunData = await runFnc(lastRunData)
102    } catch (_err) {
103      err = _err
104      // TODO retry when: Error: ROLLBACK - Client has encountered a connection error and is not queryable
105      console.log(chalkRed(chalkBold('✗')), name, _err)
106    }
107    if (!err) {
108      await fs1Promises.mkdir(zhongRunsDir, {recursive: true})
109      await fs1Promises.writeFile(lastRunFile, _.isUndefined(lastRunData) ? '' : JSON.stringify(lastRunData))  
110    }
111
112    ranJobNames.push(name)
113    const elapsedSec = (_.now() - startedAt) / 1000
114    console.log(chalkBold(chalkGreen('✓')), `${elapsedSec}s`, name)
115  }
116  if (throttledAny && !withinLongRun) {
117    console.log(chalkDim('use --ignoreRecent to force-run skipped jobs'))
118  }
119  return ranJobNames
120}