1import * as _ from 'lodash-es'
2import * as _dbActions from './dbActionsF.mts'
3import * as _connLessDbActions from './connLessDbActionsF.mts'
4import * as _kyselyDbActions from './kyselyDbActionsF.mts'
5
6import { dbConnWrapNoMap } from '../../serverF/dbF/dbConnWrapF.mts'
7import { k8sPortFwdCtx, needsProxyQs } from '../../serverF/kubePortFwdF/k8sPortFwdCtxF.mts'
8import { wrapProxyMinikNodePorts } from '../../serverF/kubePortFwdF/proxyNodePortsAI.mts'
9import { portForwardMgr } from '../../serverF/kubePortFwdF/portForwardMgrAI.mts'
10import { getQsFromSecSu } from './getQsFromSecSuF.mts'
11import { genContext } from '../../serverF/genContextF.mts'
12import { tsSec } from '../../serverF/secretsF/tsSecF.mts'
13import type { K8sProxyItem } from '../../serverF/kubePortFwdF/k8sPortFwdCtxF.mts'
14import { isStaleDbErr } from '../../serverF/dbF/isStaleDbErrAI.mts'
15import { recoverStaleDbConns } from '../../serverF/dbF/staleConnRecoverAI.mts'
16
17_.each(_connLessDbActions, (fnc: any) => fnc.noWrapDbConn = true)
18_.each(_kyselyDbActions, (fnc: any) => fnc.noWrapDbConn = true)
19
20const {regenkyselyToFile: _skip, ...baseDbActions} = {..._connLessDbActions, ..._kyselyDbActions, ..._dbActions}
21
22export const extraDbActionsCtx = genContext()
23
24export const getDbActions = () => {
25  const extraDbActions = extraDbActionsCtx.getStore() || {}
26  return {...baseDbActions, ...extraDbActions}
27}
28
29export const dbActions = baseDbActions
30
31export const runDbAction = async ({action, qsName}: {action: string, qsName: string}) => {
32  const allDbActions = getDbActions() as Record<string, any>
33  const dbActionFnc = allDbActions[action]
34  if (!dbActionFnc) return false
35
36  // su deprec since we can't run a single migration with superuser (nec for create extension)
37  // and tables will have wrong owner
38
39  const k8sProxyA = [await getQsFromSecSu({qsName: tsSec(qsName)})].filter(needsProxyQs) as K8sProxyItem[]
40  k8sPortFwdCtx.enterWith({...k8sPortFwdCtx.getStore(), k8sProxyA})
41  try {
42    const maxStaleRetries = 2
43    for (let attempt = 0; ; attempt++) {
44      try {
45        await wrapProxyMinikNodePorts(async () => {
46          if (dbActionFnc.noWrapDbConn) {
47            await dbActionFnc({qsName})
48          } else {
49            const connInfo = await getQsFromSecSu({qsName: tsSec(qsName)})
50            await dbConnWrapNoMap({connInfo, reuseConn: false, pgAdType: 'Knex'}, async (db) => {
51              await dbActionFnc({db, qsName})
52            })
53          }
54        })
55        break
56      } catch (err) {
57        if (!isStaleDbErr(err) || attempt >= maxStaleRetries) throw err
58        await recoverStaleDbConns()
59      }
60    }
61  } finally {
62    if (k8sProxyA.length) portForwardMgr.teardown(k8sProxyA.map(it => it.svcName))
63  }
64
65  return true
66}