1import { dbCtx, loadersCtx } from './dbCtxF.mts'
2import { wrapProxyMinikNodePorts } from '../kubePortFwdF/proxyNodePortsAI.mts'
3import { k8sPortFwdCtx, needsProxyQs } from '../kubePortFwdF/k8sPortFwdCtxF.mts'
4import { secMapDbConnWrap } from './dbConnWrapF.mts'
5import { clearAllCachedConnections } from './dbConnCacheF.mts'
6import { getEnvConf } from '../envConfF.mts'
7import { decomposeQsFromSec } from '../secretsF/decomposeQsFromSecF.mts'
8import type { secretNameType } from '../secretsF/getPlainSecAI.mts'
9import { disabledSecret } from '../../sharedF/disabledSecretF.mts'
10import type { DbHandles } from './dbTypesF.mts'
11
12import { pickBy, mapKeys, uniq, compact, unionBy } from 'lodash-es'
13import { isStaleDbErr } from './isStaleDbErrAI.mts'
14import { recoverStaleDbConns } from './staleConnRecoverAI.mts'
15
16let cleanupWired = false
17const wireProcessCleanup = () => {
18  if (cleanupWired) return
19  cleanupWired = true
20  const cleanup = () => { clearAllCachedConnections() }
21  process.on('beforeExit', cleanup)
22  process.on('SIGTERM', () => { cleanup(); process.exit(143) })
23  process.on('SIGINT', () => { cleanup(); process.exit(130) })
24}
25
26export const withDb = async <T,>(wrappedFnc: (dbH: DbHandles) => Promise<T>, opts: {reuseConn?: boolean}): Promise<T> => {
27  const ctx = dbCtx.getStore()
28  if (ctx) {
29    try {
30      return await wrappedFnc(ctx)
31    } catch (err) {
32      if (!isStaleDbErr(err)) throw err
33      // Cached conn died (e.g., port-forward died after laptop sleep). Evict stale
34      // pools, bounded-wait for the watchdog to revive the tunnel, then fall through
35      // to the full rebuild below which retries wrappedFnc with fresh connections.
36      await recoverStaleDbConns()
37    }
38  }
39
40  wireProcessCleanup()
41
42  const {reuseConn} = opts
43
44  // dbSecrets: { db: 'games1db_qs', authDb: 'authprod_qs' }
45  const { dbSecrets, secretsMapping } = getEnvConf()
46  
47  // Build { dbKey: secretName } from dbSecrets or fall back to secretsMapping
48  const dbKeyToSecretH: Record<string, string> = dbSecrets
49    ? pickBy(dbSecrets, secName => secName !== disabledSecret)
50    // fallback for entrypoints not yet migrated to dbSecrets
51    : mapKeys(
52        pickBy(secretsMapping, (secName, qsName) => qsName.endsWith('db_qs') && secName !== disabledSecret),
53        (secName, qsName) => {
54          // defaultdb_qs → db, authdb_qs → authDb
55          if (qsName === 'defaultdb_qs') return 'db'
56          return qsName.replace(/db_qs$/, '') + 'Db'
57        }
58      )
59
60  // k8sPortFwdCtx is a process-global accumulator (genGlobal2Ctx); a skip-guard would
61  // pin the first job's svc list and starve later jobs needing different dbs — always union.
62  const existingProxyA = k8sPortFwdCtx.getStore()?.k8sProxyA || []
63  const decomposed = await Promise.all(
64    uniq(Object.values(dbKeyToSecretH)).map(async secName => {
65      try { return await decomposeQsFromSec(secName as secretNameType) }
66      catch { return null }
67    })
68  )
69  const k8sProxyA = unionBy([...compact(decomposed).filter(needsProxyQs), ...existingProxyA], 'svcName')
70  k8sPortFwdCtx.enterWith({...k8sPortFwdCtx.getStore(), k8sProxyA})
71
72  const wrappedFnc2 = async (): Promise<T> => {
73    return await (async function chain(idx = 0, dbH: DbHandles = {}): Promise<T> {
74    
75      const entries = Object.entries(dbKeyToSecretH)
76      if (idx >= entries.length) {
77        return dbCtx.run(dbH, async () => {
78          return await loadersCtx.run({}, async () => {
79            return await wrappedFnc(dbH)
80          })
81        })
82      }
83    
84      const [dbKey, secretName] = entries[idx]
85      return await secMapDbConnWrap({qsName: secretName, reuseConn, pgAdType: 'Knex'}, async (knexConn) => {
86        if (knexConn) dbH[dbKey] = knexConn
87        const pgAdType = 'Kys'
88        return await secMapDbConnWrap({qsName: secretName, reuseConn, pgAdType}, async (kysConn) => {
89          if (kysConn) dbH[`${dbKey}Kys`] = kysConn
90          return await chain(idx + 1, dbH)
91        })
92      }) as T
93    }())
94  }
95
96  const runWithPortFwd = () => wrapProxyMinikNodePorts(wrappedFnc2)
97
98  const maxStaleRetries = 2
99  for (let attempt = 0; ; attempt++) {
100    try {
101      return await runWithPortFwd()
102    } catch (err) {
103      if (!isStaleDbErr(err) || attempt >= maxStaleRetries) throw err
104      await recoverStaleDbConns()
105    }
106  }
107}
108