1import net from 'net'
2import { liveSpawn } from '../libFs/live1SpawnF.mts'
3import { getDockName } from '../dockNameF.mts'
4import { chalkRed } from '../libChalkF.mts'
5import { getKubeConfigPath } from './kubeConfigPathAI.mts'
6import { isDevPc, isVeryVerbose } from '../../sharedF/isDevPcF.mts'
7import { sleep, pollUntil } from '../../sharedF/sleepF.mts'
8import { throPtErr } from '../../sharedF/throwErrorF/sharedErrF.mts'
9import type { K8sProxyItem } from './k8sPortFwdCtxF.mts'
10import * as _ from 'lodash-es'
11
12const canProxyNodePorts = !!isDevPc
13
14export const isPortOpen = (host: string, port: number, timeoutMs = 1000): Promise<boolean> =>
15  new Promise((resolve) => {
16    const socket = new net.Socket()
17    socket.setTimeout(timeoutMs)
18    socket.on('connect', () => { socket.destroy(); resolve(true) })
19    socket.on('timeout', () => { socket.destroy(); resolve(false) })
20    socket.on('error', () => { socket.destroy(); resolve(false) })
21    socket.connect(port, host)
22  })
23
24// Single owner of kubectl port-forward tunnel lifecycle. Tunnels are established
25// lazily via ensureForwards, monitored by per-svc watchdogs that self-heal on the
26// same port. Dev server keeps them for process-lifetime; CLI one-shot callers
27// (guardPendingMigrations) call teardown() to release the subprocess handles so the
28// process can exit. Replaces the genGlobal2Ctx k8sLocalFwdSvcPortsH slot + the
29// callback-coupled logic formerly in proxyNodePortsAI.
30class PortForwardMgr {
31  private endpoints = new Map<string, {host: string, port: number}>()
32  private watchdogs = new Map<string, Promise<void>>()
33  private aborts = new Map<string, AbortController>()
34  private cleanupWired = false
35
36  getEndpointPort(svcName: string): number | undefined {
37    return this.endpoints.get(svcName)?.port
38  }
39
40  getEndpointMap(): Record<string, number> {
41    const h: Record<string, number> = {}
42    for (const [svc, {port}] of this.endpoints) h[svc] = port
43    return h
44  }
45
46  async ensureForwards(items: K8sProxyItem[]): Promise<void> {
47    if (!canProxyNodePorts || !items.length) return
48    this.wireCleanup()
49    await Promise.all(items.map(it => this.ensureForward(it)))
50  }
51
52  // Bounded-wait for all known forwards to be (re)connectable. Used by the stale-conn
53  // retry path after a tunnel dies (e.g. laptop sleep) — the watchdog is already
54  // restarting on the same port, so this just waits for it to come back.
55  async awaitRecovery(timeoutMs = 30000): Promise<void> {
56    const ports = [...this.endpoints.values()].map(e => e.port)
57    if (!ports.length) return
58    await pollUntil(
59      async () => (await Promise.all(ports.map(p => isPortOpen('127.0.0.1', p)))).every(Boolean),
60      {timeoutMs, intervalMs: 500},
61    )
62  }
63
64  // Abort watchdogs and clear state for specified svcs (or all). Aborted watchdogs
65  // exit on their own — the abort kills the kubectl subprocess via liveSpawn's signal
66  // handler, then runWatchdog sees stderr==='killed' and returns.
67  teardown(svcNames?: string[]): void {
68    const names = svcNames ?? [...this.aborts.keys()]
69    for (const name of names) {
70      this.aborts.get(name)?.abort()
71      this.aborts.delete(name)
72      this.endpoints.delete(name)
73      this.watchdogs.delete(name)
74    }
75  }
76
77  private ensureForward = async (item: K8sProxyItem): Promise<void> => {
78    if (this.endpoints.has(item.svcName)) return
79    if (!this.watchdogs.has(item.svcName)) {
80      this.watchdogs.set(item.svcName, this.runWatchdog(item).catch(err => {
81        console.error(`port-forward ${item.svcName} watchdog failed:`, err)
82      }))
83    }
84    await pollUntil(() => this.endpoints.has(item.svcName), {timeoutMs: 60_000, intervalMs: 200})
85  }
86
87  private runWatchdog = async ({svcName, cluster_name}: K8sProxyItem): Promise<void> => {
88    const abort = new AbortController()
89    this.aborts.set(svcName, abort)
90    const {signal} = abort
91
92    const assignPort = (port: number) => {
93      if (this.endpoints.has(svcName)) return
94      this.endpoints.set(svcName, {host: '127.0.0.1', port})
95    }
96
97    if (cluster_name !== 'minik') {
98      const inclAddrInUse = (str: string) => _.includes(str, 'address already in use')
99      let portNo = _.random(5000, 9000)
100      let initRetries = 0, restartCount = 0
101      const maxInitRetries = 5
102      while (true) {
103        const ret = await liveSpawn({
104          signal, isQuiet: true, noOutCmd: true,
105          cmd: `KUBECONFIG=${getKubeConfigPath(cluster_name)} kubectl port-forward service/${svcName}-rw ${portNo}:5432`,
106          onDataFnc: (data) => {
107            if (!(data.match('Forwarding from') || inclAddrInUse(data))) return
108            assignPort(portNo)
109            return undefined
110          },
111        })
112        const {isSuccess, stdout, stderr, exitCode} = ret
113        if (stderr === 'killed') return
114        if (exitCode == 123) return // to ctrl-d out of dbconsole without throw
115        if (signal.aborted) return
116        if (this.endpoints.has(svcName)) {
117          restartCount++
118          console.log(`port-forward ${svcName} died, restarting (${restartCount})...`)
119          await sleep(Math.min(1000 * restartCount, 30_000))
120          continue
121        }
122        const portInUse = inclAddrInUse(stdout) || inclAddrInUse(stderr)
123        if (portInUse && initRetries < maxInitRetries - 1) {
124          initRetries++
125          portNo = _.random(5000, 9000)
126          continue
127        }
128        if (!isSuccess && !portInUse) {
129          if (initRetries < maxInitRetries - 1) {
130            initRetries++
131            await sleep(Math.min(1000 * initRetries, 10_000))
132            continue
133          }
134          throPtErr('portFwdFailed', ret)
135        }
136        if (portInUse) throPtErr('portFwdAllRetriesFailed', {attempts: maxInitRetries, svcName})
137        return
138      }
139    }
140    // minikube path
141    const cmd = `minikube service ${svcName} --url`
142    const restH = await liveSpawn({ // dont change to liveSpawnThrow or it won't be able to catch below
143      signal, isQuiet: true, noOutCmd: true,
144      cmd,
145      onDataFnc: (data) => {
146        const match = data.match(/http:\/\/127\.0\.0\.1:(\d+)/)
147        if (!match) return
148        assignPort(_.parseInt(match[1]))
149        return undefined
150      },
151    })
152    const {isSuccess, stderr, stdout} = restH
153    if (isSuccess || stderr === 'killed') return
154
155    if (isVeryVerbose) console.log('k8sProxyFail', {...restH, cmd})
156
157    if (_.includes(stdout, 'Exiting due to SVC_NOT_FOUND')) {
158      console.log(chalkRed(`\npls make sure ${svcName} is deployed`))
159    }
160    if (_.includes(stdout, 'The control plane node must be running for this command') || _.includes(stdout, 'Exiting due to GUEST_STATUS: Unable to get machine status')) {
161      console.log(chalkRed(`\npls make sure minikube is running:`))
162      console.log(`minikube start --network=socket_vmnet --driver=${getDockName()}\n`)
163    }
164    process.exit(1)
165  }
166
167  private wireCleanup() {
168    if (this.cleanupWired) return
169    this.cleanupWired = true
170    const cleanup = () => { for (const a of this.aborts.values()) a.abort() }
171    process.on('beforeExit', cleanup)
172    process.on('SIGTERM', () => { cleanup(); process.exit(143) })
173    process.on('SIGINT', () => { cleanup(); process.exit(130) })
174  }
175}
176
177// Hoist to globalThis so all webpack module instances (Next.js dev recompiles
178// per-route + HMR) share one mgr — otherwise endpoints/watchdogs reset on
179// re-evaluation and tunnels re-establish needlessly. Mirrors dbConnCacheF.
180const g = globalThis as unknown as { __portForwardMgr?: PortForwardMgr }
181export const portForwardMgr: PortForwardMgr = g.__portForwardMgr ||= new PortForwardMgr()
182