1import { deployDbCtx } from "../ctxF/deployDbCtxF.mts";
2import type { Knex } from 'knex'
3import { getQsFromSecSu } from './getQsFromSecSuF.mts'
4import { tsSec } from '../../serverF/secretsF/tsSecF.mts'
5import { cliFlag } from '../../serverF/cliArgAI.mts'
6import { hasUserConfirmedGit } from '../../sharedF/gitDestructivePatternsF.mts'
7import { getProcArgv } from '../../serverF/isDirectlyRunF.mts'
8import { dbConnWrapNoMap } from '../../serverF/dbF/dbConnWrapF.mts'
9import type { DbConnInfo } from '../../serverF/dbF/dbTypesF.mts'
10import { getUncommittedChanges, parseUncommittedPaths, guard1CleanCommit } from '../gitF/guardCleanCommitF.mts'
11import { ptDir } from '../../serverF/ptDirF.mts'
12import { throPtErr } from '../../sharedF/throwErrorF/sharedErrF.mts'
13import { regenkyselyToFile } from './kyselyDbActionsF.mts'
14import * as _ from 'lodash-es'
15
16type DbCtx = {db: Knex}
17
18const staleStates = ['idle', 'idle in transaction', 'idle in transaction (aborted)', 'disabled']
19
20const getSuConnInfo = async (qsName: string) => {
21  return await getQsFromSecSu({qsName: tsSec(qsName), needSu: true}) as DbConnInfo
22}
23
24export const killstaleconns = async ({qsName}: {qsName: string}) => {
25  const connInfo = await getSuConnInfo(qsName)
26  const force = cliFlag('--force')
27
28  await dbConnWrapNoMap({connInfo, reuseConn: false, pgAdType: 'Knex'}, async (_db) => {
29    const db = _db as Knex
30    const {rows: conns} = await db.raw(`
31      SELECT pid, state, application_name, left(query, 80) as query
32      FROM pg_stat_activity
33      WHERE usename = current_user AND pid <> pg_backend_pid()
34      ORDER BY query_start desc
35    `)
36
37    console.log(`Found ${conns.length} connections from current user`)
38    const staleConns = conns.filter((c: {state: string}) => staleStates.includes(c.state))
39    const activeConns = conns.filter((c: {state: string}) => !staleStates.includes(c.state))
40
41    if (force) {
42      if (conns.length === 0) { console.log('No connections to kill'); return }
43      console.log(`FORCE: terminating ALL ${conns.length} connections (stale + active)`)
44      for (const c of conns) { await db.raw(`SELECT pg_terminate_backend(${c.pid})`) }
45      console.log(`Terminated ${conns.length} connections`)
46      return
47    }
48
49    if (activeConns.length > 0) {
50      console.log(`Skipping ${activeConns.length} active connections:`, activeConns.map((c: {pid: number, state: string, query: string}) => ({pid: c.pid, state: c.state, query: c.query})))
51    }
52
53    if (staleConns.length === 0) { console.log('No stale connections to kill'); return }
54
55    console.log(`Terminating ${staleConns.length} stale connections:`, staleConns.map((c: {pid: number, state: string}) => ({pid: c.pid, state: c.state})))
56    for (const c of staleConns) { await db.raw(`SELECT pg_terminate_backend(${c.pid})`) }
57    console.log(`Terminated ${staleConns.length} stale connections`)
58  })
59}
60killstaleconns.cliDescript = 'terminate idle/stuck connections from this user (--force: kill all including active; connects as superuser)'
61killstaleconns.cliSchema = { force: { flag: true, desc: 'also terminate active connections' } }
62killstaleconns.noWrapDbConn = true
63
64export const connsall = async ({qsName}: {qsName: string}) => {
65  const connInfo = await getSuConnInfo(qsName)
66  await dbConnWrapNoMap({connInfo, reuseConn: false, pgAdType: 'Knex'}, async (_db) => {
67    const db = _db as Knex
68    const {rows} = await db.raw(`
69      SELECT pid, state, usename, application_name, left(query, 60) as query
70      FROM pg_stat_activity
71      WHERE pid <> pg_backend_pid()
72      ORDER BY query_start desc
73    `)
74    console.table(rows)
75  })
76}
77connsall.cliDescript = 'show all connections (superuser)'
78connsall.noWrapDbConn = true
79
80export const migratestatus = async ({db}: DbCtx) => {
81  const {migrationOpts} = deployDbCtx.getStore() ?? {}
82  const [lastAppliedMigrations, unappliedMigrations] = await db.migrate.list(migrationOpts)
83  const last5AppliedMigrations = _.takeRight(lastAppliedMigrations, 5)
84  console.log({last5AppliedMigrations, totalApplied: lastAppliedMigrations.length, unappliedMigrations})
85}
86migratestatus.cliDescript = 'show pending/applied migrations'
87
88export const rollback = async ({db}: DbCtx) => {
89  const {migrationOpts} = deployDbCtx.getStore() ?? {}
90  const ret = await db.migrate.down(migrationOpts)
91  const [idk, migrationNames] = ret
92  console.log('rolled back', migrationNames)
93}
94
95export const migrate = async ({db}: DbCtx) => {
96  const {migrationOpts, qsName, kyselyOutFile} = deployDbCtx.getStore() ?? {}
97
98  const migDirs = _.castArray((migrationOpts as any).directory).map((d: string) => d.replace(ptDir + '/', ''))
99  const dirtyMigs = parseUncommittedPaths(getUncommittedChanges())
100    .filter(p => migDirs.some(d => p === d || p.startsWith(d + '/')))
101  if (dirtyMigs.length && !hasUserConfirmedGit(getProcArgv())) {
102    throPtErr(
103      `uncommitted migration files - commit before migrating:\n  ${dirtyMigs.join('\n  ')}\n(override: --confirmedByUser=<secret>)`,
104      {dirtyMigs}
105    )
106  }
107
108  const cfg = {
109    ...migrationOpts,
110    disableTransactions: false
111  }
112  const [batchNo, log] = await db.migrate.latest(cfg)
113  if (!log.length) {
114    console.info('Database is already up to date');
115  } else {
116    console.info('Ran migrations: ' + log.join(', '));
117  }
118
119  if (kyselyOutFile && qsName) {
120    try {
121      await guard1CleanCommit(
122        () => regenkyselyToFile({qsName, outFile: kyselyOutFile}),
123        {commitMsg: `pt kysely-regen: ${qsName}`, relevantFilesA: [kyselyOutFile], skipViolationCheck: true}
124      )
125    } catch (err) {
126      console.warn(`kysely regen failed (migrations already applied): ${err}`)
127    }
128  }
129}
130migrate.cliDescript = 'run pending migrations'
131migrate.cliSchema = { confirmedByUser: { type: 'string', hidden: true, desc: 'override dirty-migration guard' } }
132