1import * as _ from 'lodash-es'
2import * as net from 'net'
3import * as fs from 'fs'
4import { PortForward } from '@kubernetes/client-node'
5import { get1ClusterPods } from '../k8sF/getClusterPodsF.mts'
6import { getKubeApis } from '../k8sF/getApisF.mts'
7import { ret1PodLogs } from '../k8sF/logsF/retPodLogsF.mts'
8import { getOfflineReason } from '../k8sF/ctxF/klusterReachabilityAI.mts'
9import { luxNow } from '../../sharedF/timeF/luxon0NowF.mts'
10import { DateTime } from 'luxon'
11import { chalkGreen, chalkRed } from '../../serverF/libChalkF.mts'
12import { jsObjStringify } from '../../sharedF/moDashF/jsObjStringifyF.mts'
13import { ensurePtTmpDir, ptTmpDir } from '../../serverF/libFs/ensurePtTmpDirF.mts'
14
15const historyFilePath = `${ptTmpDir}/ethstatus-history.json`
16const staleImageDays = 30
17
18const parseHexBlock = (b: any) => typeof b === 'string' ? parseInt(b, 16) || 0 : (b || 0)
19const parseStageMap = (stages: any[]) => Object.fromEntries(stages.map((s: any) => [s.name, parseHexBlock(s.block)]))
20
21import { ethImgToClientType as imgToClientType } from './ethClientRegistryAI.mts'
22
23export const parseImageDatesFromFile = (epFilePath: string | null) => {
24  if (!epFilePath || !fs.existsSync(epFilePath)) return {}
25  const content = fs.readFileSync(epFilePath, 'utf8')
26  const result: Record<string, string | null> = {exec: null, cons: null}
27  for (const line of content.split('\n')) {
28    const dateMatch = line.match(/\/\/ (\d{6})$/)
29    if (!dateMatch) continue
30    const varMatch = line.match(/(\w+(?:V|Version)):\s*'/)
31    if (varMatch && imgToClientType[varMatch[1]]) {
32      const clientType = imgToClientType[varMatch[1]]
33      if (clientType !== 'skip') result[clientType] = dateMatch[1]
34      continue
35    }
36    const imgMatch = line.match(/image:\s*'([^':]+)/)
37    if (imgMatch && imgToClientType[imgMatch[1]]) {
38      const clientType = imgToClientType[imgMatch[1]]
39      if (clientType !== 'skip') result[clientType] = dateMatch[1]
40    }
41  }
42  return result
43}
44
45const formatImageAge = (dateStr: string | null) => {
46  if (!dateStr) return null
47  const imgDt = DateTime.fromFormat(dateStr, 'yyLLdd')
48  if (!imgDt.isValid) return null
49  const ageDur = luxNow().diff(imgDt)
50  const days = Math.floor(ageDur.as('days'))
51  const ageStr = `img ${formatPodAge(ageDur)}`
52  return days >= staleImageDays ? chalkRed(ageStr) : chalkGreen(ageStr)
53}
54
55// Extract reth indexing progress from pod logs (format: "Collecting indices progress=XX.XXXX%")
56const getRethIndexingProgress = async ({cluster_name, podName, containerName}: {cluster_name: string, podName: string, containerName: string}) => {
57  try {
58    let logs = await ret1PodLogs({cluster_name, podName, containerName, tailLines: 100})
59    logs = logs.replace(/\x1B\[[0-9;]*m/g, '') // strip ANSI color codes
60    const matches = logs.match(/Collecting indices.*progress=(\d+\.?\d*)%/g)
61    if (!matches?.length) return null
62    const lastMatch = matches[matches.length - 1]
63    const pct = parseFloat(lastMatch.match(/progress=(\d+\.?\d*)%/)?.[1] || '0')
64    return pct > 0 ? pct : null
65  } catch {
66    return null
67  }
68}
69
70const loadHistory = () => {
71  try {
72    if (fs.existsSync(historyFilePath)) {
73      return JSON.parse(fs.readFileSync(historyFilePath, 'utf8'))
74    }
75  } catch {}
76  return {}
77}
78
79const saveHistory = (history: Record<string, any>) => {
80  ensurePtTmpDir()
81  fs.writeFileSync(historyFilePath, JSON.stringify(history, null, 2))
82}
83
84// Detect if reth is in indexing state with no checkpoint (needs log-based progress)
85const isRethIndexingNoCheckpoint = (syncStatus: any) => {
86  const stages = syncStatus?.stages || []
87  if (stages.length === 0) return false
88  const stageMap = parseStageMap(stages)
89  const headers = stageMap.Headers || 0, execution = stageMap.Execution || 0
90  const indexAccount = stageMap.IndexAccountHistory || 0, indexStorage = stageMap.IndexStorageHistory || 0
91  const execCaughtUp = execution > 0 && execution === headers
92  return execCaughtUp && Math.min(indexAccount, indexStorage) === 0
93}
94
95// Extract a single trackable metric from sync status for comparison
96const extractMetric = ({syncStatus, isConsensus, logIndexingPct}: {syncStatus: any, isConsensus: boolean, logIndexingPct?: number | null}) => {
97  if (!syncStatus) return null
98  if (isConsensus) {
99    return {type: 'slot', value: syncStatus.headSlot || 0}
100  }
101  // For exec clients, track indexing progress if still indexing, otherwise block
102  const stages = syncStatus.stages || []
103  if (stages.length > 0) {
104    const stageMap = parseStageMap(stages)
105    const headers = stageMap.Headers || 0
106    const execution = stageMap.Execution || 0
107    const finish = stageMap.Finish || 0
108    const indexAccount = stageMap.IndexAccountHistory || 0
109    const indexStorage = stageMap.IndexStorageHistory || 0
110    const minIndex = Math.min(indexAccount, indexStorage)
111    const execCaughtUp = finish > 0 || (execution > 0 && execution === headers)
112    // If indexing is behind execution, track index progress
113    if (execCaughtUp && execution > 0 && minIndex < execution && minIndex > 0) {
114      return {type: 'indexPct', value: (minIndex / execution) * 100, minIndex, execution}
115    }
116    // During indexing with no checkpoint, use log-based progress if available
117    if (execCaughtUp && execution > 0 && minIndex === 0) {
118      if (logIndexingPct != null) return {type: 'indexPct', value: logIndexingPct}
119      return null
120    }
121    if (execution > 0) return {type: 'block', value: execution}
122  }
123  if (syncStatus.isSynced) return {type: 'block', value: syncStatus.currentBlock}
124  return {type: 'block', value: syncStatus.currentBlock || syncStatus.highestBlock || 0}
125}
126
127const formatDelta = (prev: any, curr: any, prevTs: number) => {
128  if (!prev || !curr || prev.type !== curr.type) return null
129  const delta = curr.value - prev.value
130  if (delta === 0) {
131    const stuckMin = Math.round((Date.now() - prevTs) / 60000)
132    return {delta: 0, formatted: chalkRed(`stuck ${stuckMin}m`), isStuck: true}
133  }
134  const sign = delta > 0 ? '+' : ''
135  const minElapsed = Math.max(1, (Date.now() - prevTs) / 60000)
136  const rate = delta / minElapsed
137  if (curr.type === 'slot') {
138    const rateStr = rate >= 1 ? `${rate.toFixed(1)}/m` : ''
139    return {delta, formatted: chalkGreen(`${sign}${delta} slots${rateStr ? ' ' + rateStr : ''}`), isStuck: false}
140  }
141  if (curr.type === 'indexPct') {
142    if (delta < 0) return null // indexing reset/restart - not comparable
143    const rateStr = rate >= 0.01 ? `${rate.toFixed(2)}%/m` : ''
144    return {delta, formatted: chalkGreen(`+${delta.toFixed(2)}%${rateStr ? ' ' + rateStr : ''}`), isStuck: false}
145  }
146
147  // block
148  const rateStr = rate >= 1 ? `${rate.toFixed(0)}/m` : ''
149  return {delta, formatted: chalkGreen(`${sign}${delta.toLocaleString()} blocks${rateStr ? ' ' + rateStr : ''}`), isStuck: false}
150}
151
152let nextLocalPort = 19000
153
154const getNextLocalPort = () => {
155  return nextLocalPort++
156}
157
158export const withPortForward = async <T,>({cluster_name, podName, portNo, checkerFn}: {cluster_name: string, podName: string, portNo: number, checkerFn: (p: {httpHostPort: string}) => Promise<T>}): Promise<T> => {
159  const {kubeConfig} = await getKubeApis({cluster_name})
160  const forward = new PortForward(kubeConfig)
161
162  const localPort = getNextLocalPort()
163  const sockets = new Set<net.Socket>()
164  const wsPromises: Promise<unknown>[] = []
165  const server = net.createServer((socket) => {
166    sockets.add(socket)
167    socket.on('error', (err: any) => {
168      if (err.code === 'EPIPE' || err.code === 'ECONNRESET') return
169      console.error('port-forward socket error:', err.message)
170    })
171    socket.on('close', () => { sockets.delete(socket) })
172    wsPromises.push(forward.portForward('default', podName, [portNo], socket, null, socket).catch(() => null))
173  })
174
175  server.listen(localPort, '127.0.0.1')
176  const httpHostPort = `http://127.0.0.1:${localPort}`
177
178  try {
179    return await checkerFn({httpHostPort})
180  } finally {
181    server.close()
182    for (const socket of sockets) socket.destroy()
183    const wss = await Promise.all(wsPromises)
184    for (const ws of wss) {
185      try { (typeof ws === 'function' ? (ws as () => any)() : ws)?.close() } catch {} // catch:userapproved
186    }
187  }
188}
189
190const statusFetchTimeout = 20000
191
192const jsonRpcCall = async ({httpHostPort, method, params = []}: {httpHostPort: string, method: string, params?: any[]}) => {
193  const resp = await fetch(httpHostPort, {
194    method: 'POST',
195    headers: {'Content-Type': 'application/json'},
196    body: JSON.stringify({jsonrpc: '2.0', method, params, id: 1}),
197    signal: AbortSignal.timeout(statusFetchTimeout),
198  })
199  const json = await resp.json()
200  return json.result
201}
202
203const getExecSyncStatus = async ({httpHostPort}: {httpHostPort: string}) => {
204  const [syncingRes, blockNumRes, peerCountRes] = await Promise.all([
205    jsonRpcCall({httpHostPort, method: 'eth_syncing'}),
206    jsonRpcCall({httpHostPort, method: 'eth_blockNumber'}),
207    jsonRpcCall({httpHostPort, method: 'net_peerCount'}),
208  ])
209
210  let versionRes = null
211  try {
212    versionRes = await jsonRpcCall({httpHostPort, method: 'web3_clientVersion'})
213  } catch {} // catch:userapproved
214
215  const peerCount = parseInt(peerCountRes, 16) || 0
216  const currentBlockNum = parseInt(blockNumRes, 16) || 0
217  const isSynced = syncingRes === false
218
219  let syncStatus
220  if (isSynced) {
221    syncStatus = {isSynced: true, currentBlock: currentBlockNum}
222  } else {
223    const currentBlock = parseInt(syncingRes.currentBlock, 16) || 0
224    const highestBlock = parseInt(syncingRes.highestBlock, 16) || 0
225    const pct = highestBlock > 0 ? ((currentBlock / highestBlock) * 100).toFixed(2) : 0
226    // Extract reth stages for debugging
227    const stages = syncingRes.stages || []
228    syncStatus = {isSynced: false, currentBlock, highestBlock, pct, stages}
229  }
230
231  return {syncStatus, version: versionRes, peerCount}
232}
233
234export const getConsensusSyncStatus = async ({httpHostPort}: {httpHostPort: string}) => {
235  const [syncingResp, versionResp, peerCountResp] = await Promise.all([
236    fetch(`${httpHostPort}/eth/v1/node/syncing`, {signal: AbortSignal.timeout(statusFetchTimeout)}),
237    fetch(`${httpHostPort}/eth/v1/node/version`, {signal: AbortSignal.timeout(statusFetchTimeout)}),
238    fetch(`${httpHostPort}/eth/v1/node/peer_count`, {signal: AbortSignal.timeout(statusFetchTimeout)}),
239  ])
240
241  const syncingJson = await syncingResp.json()
242  const versionJson = await versionResp.json()
243  const peerCountJson = await peerCountResp.json()
244
245  const {head_slot, sync_distance, is_syncing} = syncingJson.data
246  const headSlot = _.toInteger(head_slot)
247  const syncDistance = _.toInteger(sync_distance)
248  const isSynced = !is_syncing && syncDistance === 0
249
250  let syncStatus
251  if (isSynced) {
252    syncStatus = {isSynced: true, headSlot}
253  } else {
254    const pct = headSlot > 0 ? ((headSlot / (headSlot + syncDistance)) * 100).toFixed(2) : 0
255    syncStatus = {isSynced: false, headSlot, syncDistance, pct}
256  }
257
258  const version = versionJson.data?.version || 'unknown'
259  const peerCount = _.toInteger(peerCountJson.data?.connected || 0)
260
261  return {syncStatus, version, peerCount, headSlot}
262}
263
264const getPodAgeDur = (creationTimestamp: any) => {
265  if (!creationTimestamp) return null
266  const createdDt = DateTime.fromJSDate(new Date(creationTimestamp))
267  if (!createdDt.isValid) return null
268  return luxNow().diff(createdDt)
269}
270
271const formatPodAge = (diffDur: any) => {
272  if (!diffDur) return '?'
273  const days = Math.floor(diffDur.as('days'))
274  const hours = Math.floor(diffDur.as('hours') % 24)
275  if (days > 0) return `${days}d${hours}h`
276  const mins = Math.floor(diffDur.as('minutes') % 60)
277  if (hours > 0) return `${hours}h${mins}m`
278  return `${mins}m`
279}
280
281export const getPodHealthInfo = async ({cluster_name, fuzzyPodName}: {cluster_name: string, fuzzyPodName: string}) => {
282  const pods = await get1ClusterPods({cluster_name, labels: {name: fuzzyPodName}})
283  const activePods = _.filter(pods, (pod: any) => !_.includes(['Evicted', 'Failed'], pod.status?.phase) && pod.status?.reason !== 'Evicted')
284  const pod = _.first(activePods) as any
285  if (!pod) return {status: 'NotFound', age: '-', restarts: 0, podName: null}
286
287  const {metadata, status} = pod
288  const ageDur = getPodAgeDur(metadata?.creationTimestamp)
289  const age = formatPodAge(ageDur)
290  const containerStatus: any = _.first(status?.containerStatuses) || {}
291  const restarts = containerStatus.restartCount || 0
292
293  const lastTerm = containerStatus.lastState?.terminated
294  const lastExitReason = lastTerm?.reason
295  const lastExitAge = lastTerm?.finishedAt ? formatPodAge(getPodAgeDur(lastTerm.finishedAt)) : null
296
297  let displayStatus = status?.phase
298  const waitingReason = containerStatus.state?.waiting?.reason
299  if (waitingReason) {
300    displayStatus = waitingReason
301  }
302
303  const isInitializing = ageDur && ageDur.as('minutes') < 10
304  return {status: displayStatus, age, restarts, podName: metadata?.name, isRunning: status?.phase === 'Running' && containerStatus.ready, isInitializing, lastExitReason, lastExitAge}
305}
306
307export const fmtPodDownErr = ({podHealth, errMsg}: {podHealth: any, errMsg: string}) => {
308  const {lastExitReason, lastExitAge, restarts} = podHealth || {}
309  if (!lastExitReason) return errMsg
310  const restartPart = restarts > 0 ? `, ${restarts} restarts` : ''
311  return `down (last exit ${lastExitReason} ${lastExitAge} ago${restartPart})`
312}
313
314const formatSyncStatus = (syncStatus: any, isConsensus: boolean) => {
315  if (syncStatus.isSynced) {
316    if (isConsensus) return `synced (slot ${syncStatus.headSlot})`
317    return `synced (block ${syncStatus.currentBlock.toLocaleString()})`
318  }
319  if (isConsensus) {
320    // Treat syncDistance <= 0 as synced (handles -1, 0 cases)
321    if (syncStatus.syncDistance <= 0) {
322      return `synced (slot ${syncStatus.headSlot})`
323    }
324    return `syncing ${syncStatus.pct}% (slot ${syncStatus.headSlot}, -${syncStatus.syncDistance})`
325  }
326  // Show reth stages if available (reth uses staged sync)
327  const stages = syncStatus.stages || []
328  if (stages.length > 0) {
329    const stageMap = parseStageMap(stages)
330    const headers = stageMap.Headers || 0
331    const execution = stageMap.Execution || 0
332    const finish = stageMap.Finish || 0
333    const indexAccount = stageMap.IndexAccountHistory || 0
334    const indexStorage = stageMap.IndexStorageHistory || 0
335    const minIndex = Math.min(indexAccount, indexStorage)
336    // Execution caught up to headers?
337    const execCaughtUp = finish > 0 || (execution > 0 && execution === headers)
338    if (execCaughtUp) {
339      // Check if indexing is still in progress (reth checkpoints lag behind actual progress)
340      if (execution > 0 && minIndex < execution) {
341        if (minIndex === 0) return `indexing (block ${execution.toLocaleString()})` // checkpoint not yet written
342        const idxPct = ((minIndex / execution) * 100).toFixed(1)
343        return `indexing ${idxPct}% (idx ${minIndex.toLocaleString()}/${execution.toLocaleString()})`
344      }
345      return `synced (block ${execution.toLocaleString()})`
346    }
347    // Otherwise show execution progress against headers
348    const pct = headers > 0 ? ((execution / headers) * 100).toFixed(1) : 0
349    return `syncing ${pct}% (exec ${execution.toLocaleString()}/${headers.toLocaleString()})`
350  }
351  // Nethermind snap sync shows currentBlock=0 while downloading state
352  if (syncStatus.currentBlock === 0 && syncStatus.highestBlock > 0) {
353    return `snap syncing (target block ${syncStatus.highestBlock.toLocaleString()})`
354  }
355  return `syncing ${syncStatus.pct}% (block ${syncStatus.currentBlock.toLocaleString()}/${syncStatus.highestBlock.toLocaleString()})`
356}
357
358const getFullClientStatus = async ({cluster_name, fuzzyPodName, portNo, isConsensus}: {cluster_name: string, fuzzyPodName: string, portNo: number, isConsensus: boolean}) => {
359  const podHealth = await getPodHealthInfo({cluster_name, fuzzyPodName})
360
361  if (!podHealth.isRunning || !podHealth.podName) {
362    return {podHealth, clientStatus: null}
363  }
364
365  const checkerFn: (p: {httpHostPort: string}) => Promise<any> = isConsensus ? getConsensusSyncStatus : getExecSyncStatus
366
367  let clientStatus
368  try {
369    clientStatus = await withPortForward({
370      cluster_name, podName: podHealth.podName, portNo, checkerFn
371    })
372  } catch (err) {
373    clientStatus = {error: (err as Error).message}
374  }
375
376  return {podHealth, clientStatus}
377}
378
379const formatClientLine = ({fuzzyPodName, podHealth, clientStatus, isConsensus, isActiveValidator, historyDelta, imageAge}: {
380  fuzzyPodName: string, podHealth: any, clientStatus: any, isConsensus: boolean,
381  isActiveValidator: boolean, historyDelta: any, imageAge: string | null,
382}) => {
383  const name = fuzzyPodName.padEnd(16)
384  const restartsPart = podHealth.restarts > 0 ? ` [${podHealth.restarts} restarts]` : ''
385
386  if (!podHealth.isRunning) {
387    return `${name} ${podHealth.status} (${podHealth.age})${restartsPart}`
388  }
389
390  const podPart = `Running (${podHealth.age})`
391
392  if (clientStatus?.error) {
393    if (podHealth.isInitializing) {
394      return `${name} ${podPart} | initializing...`
395    }
396    return `${name} ${podPart} | ERROR: ${clientStatus.error}`
397  }
398
399  if (!clientStatus) {
400    return `${name} ${podPart} | status unavailable`
401  }
402
403  const syncPart = formatSyncStatus(clientStatus.syncStatus, isConsensus)
404  const peerPart = `${clientStatus.peerCount} peers`
405  const versionPart = clientStatus.version?.split('/').slice(0, 2).join('/') || '-'
406  const deltaPart = historyDelta ? ` [${historyDelta.formatted}]` : ''
407  const imgPart = imageAge ? ` | ${imageAge}` : ''
408
409  const validatorSuffix = isActiveValidator ? ' ' + chalkGreen('[ACTIVE VALIDATOR]') : ''
410  return `${name} ${podPart} | ${syncPart}${deltaPart} | ${peerPart} | ${versionPart}${imgPart}${validatorSuffix}`
411}
412
413export const runEthStatus = async ({ethStatusCfgs, safeVal = null}: {ethStatusCfgs: any[], safeVal?: any}) => {
414  const history = loadHistory()
415  const now = Date.now()
416  const newHistory: Record<string, any> = {}
417
418  // Stream results per cluster - print header then status immediately
419  for (const [idx, clusterCfg] of ethStatusCfgs.entries()) {
420    const {cluster_name, epPath, epFilePath, execClient, consClient} = clusterCfg
421    const epPathPart = epPath ? ` ${epPath}` : ''
422    console.log(`${idx > 0 ? '\n' : ''}${jsObjStringify({cluster_name})}${epPathPart}`)
423    const imageDates = parseImageDatesFromFile(epFilePath)
424    const execImageAge = formatImageAge(imageDates.exec)
425    const consImageAge = formatImageAge(imageDates.cons)
426
427    const offlineReason = getOfflineReason(cluster_name)
428    if (offlineReason) {
429      console.log(`(offline: ${offlineReason})`)
430      continue
431    }
432
433    try {
434      const [execStatus, consStatus] = await Promise.all([
435        getFullClientStatus({cluster_name, ...execClient, isConsensus: false}),
436        getFullClientStatus({cluster_name, ...consClient, isConsensus: true}),
437      ])
438
439      // Extract metrics and compare to history
440      const execKey = `${cluster_name}:${execClient.fuzzyPodName}`
441      const consKey = `${cluster_name}:${consClient.fuzzyPodName}`
442
443      // For reth indexing with no checkpoint, fetch progress from logs
444      let logIndexingPct = null
445      if (isRethIndexingNoCheckpoint(execStatus.clientStatus?.syncStatus) && execStatus.podHealth?.podName) {
446        logIndexingPct = await getRethIndexingProgress({
447          cluster_name, podName: execStatus.podHealth.podName, containerName: execClient.fuzzyPodName,
448        })
449      }
450
451      const execMetric = extractMetric({syncStatus: execStatus.clientStatus?.syncStatus, isConsensus: false, logIndexingPct})
452      const consMetric = extractMetric({syncStatus: consStatus.clientStatus?.syncStatus, isConsensus: true})
453
454      const execDelta = formatDelta(history[execKey]?.metric, execMetric, history[execKey]?.ts)
455      const consDelta = formatDelta(history[consKey]?.metric, consMetric, history[consKey]?.ts)
456
457      // Only update history if metric changed (to preserve "stuck since" timestamp)
458      if (execMetric) {
459        const changed = !history[execKey] || history[execKey].metric?.value !== execMetric.value
460        newHistory[execKey] = {metric: execMetric, ts: changed ? now : history[execKey]?.ts || now}
461      }
462      if (consMetric) {
463        const changed = !history[consKey] || history[consKey].metric?.value !== consMetric.value
464        newHistory[consKey] = {metric: consMetric, ts: changed ? now : history[consKey]?.ts || now}
465      }
466
467      const isActiveValidator = safeVal?.cluster_name === cluster_name && safeVal?.name === consClient.fuzzyPodName
468      console.log(formatClientLine({fuzzyPodName: execClient.fuzzyPodName, ...execStatus, isConsensus: false, isActiveValidator: false, historyDelta: execDelta, imageAge: execImageAge}))
469      console.log(formatClientLine({fuzzyPodName: consClient.fuzzyPodName, ...consStatus, isConsensus: true, isActiveValidator, historyDelta: consDelta, imageAge: consImageAge}))
470    } catch (err) {
471      console.log(`(skipped: ${(err as Error).message})`)
472    }
473  }
474
475  saveHistory(newHistory)
476}
477