kestrelos/server/utils/db.js

import { join, dirname } from 'node:path'
import { mkdirSync, existsSync } from 'node:fs'
import { createRequire } from 'node:module'
import { promisify } from 'node:util'
import { randomBytes } from 'node:crypto'
import { hashPassword } from './password.js'
import { registerCleanup } from './shutdown.js'

// Resolve from project root so bundled server (e.g. .output) finds node_modules/sqlite3
const requireFromRoot = createRequire(join(process.cwd(), 'package.json'))
const sqlite3 = requireFromRoot('sqlite3')

const SCHEMA_VERSION = 4
const DB_BUSY_TIMEOUT_MS = 5000

let dbInstance = null
let testPath = null

const SCHEMA = {
  schema_version: 'CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)',
  users: `CREATE TABLE IF NOT EXISTS users (
    id TEXT PRIMARY KEY,
    identifier TEXT UNIQUE NOT NULL,
    password_hash TEXT NOT NULL,
    role TEXT NOT NULL DEFAULT 'member',
    created_at TEXT NOT NULL
  )`,
  users_v2: `CREATE TABLE users_new (
    id TEXT PRIMARY KEY,
    identifier TEXT UNIQUE NOT NULL,
    password_hash TEXT,
    role TEXT NOT NULL DEFAULT 'member',
    created_at TEXT NOT NULL,
    auth_provider TEXT NOT NULL DEFAULT 'local',
    oidc_issuer TEXT,
    oidc_sub TEXT
  )`,
  users_oidc_index: `CREATE UNIQUE INDEX IF NOT EXISTS users_oidc_unique ON users(oidc_issuer, oidc_sub) WHERE oidc_issuer IS NOT NULL AND oidc_sub IS NOT NULL`,
  sessions: `CREATE TABLE IF NOT EXISTS sessions (
    id TEXT PRIMARY KEY,
    user_id TEXT NOT NULL,
    created_at TEXT NOT NULL,
    expires_at TEXT NOT NULL,
    FOREIGN KEY(user_id) REFERENCES users(id)
  )`,
  pois: `CREATE TABLE IF NOT EXISTS pois (
    id TEXT PRIMARY KEY,
    lat REAL NOT NULL,
    lng REAL NOT NULL,
    label TEXT NOT NULL DEFAULT '',
    icon_type TEXT NOT NULL DEFAULT 'pin'
  )`,
  devices: `CREATE TABLE IF NOT EXISTS devices (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL DEFAULT '',
    device_type TEXT NOT NULL,
    vendor TEXT,
    lat REAL NOT NULL,
    lng REAL NOT NULL,
    stream_url TEXT NOT NULL DEFAULT '',
    source_type TEXT NOT NULL DEFAULT 'mjpeg',
    config TEXT
  )`,
}

const getDbPath = () => {
  if (testPath) return testPath
  if (process.env.DB_PATH) return process.env.DB_PATH
  const dir = join(process.cwd(), 'data')
  if (!existsSync(dir)) mkdirSync(dir, { recursive: true })
  return join(dir, 'kestrelos.db')
}

export const getAvatarsDir = () => {
  const dir = join(dirname(getDbPath()), 'avatars')
  if (!existsSync(dir)) mkdirSync(dir, { recursive: true })
  return dir
}

const getSchemaVersion = async (get) => {
  try {
    const row = await get('SELECT version FROM schema_version ORDER BY version DESC LIMIT 1')
    return row?.version || 0
  }
  catch {
    return 0
  }
}

const setSchemaVersion = (run, version) => run('INSERT OR REPLACE INTO schema_version (version) VALUES (?)', [version])

const migrateToV2 = async (run, all) => {
  const info = await all('PRAGMA table_info(users)')
  if (info.some(c => c.name === 'auth_provider')) return

  await run('BEGIN TRANSACTION')
  try {
    await run(SCHEMA.users_v2)
    await run('INSERT INTO users_new (id, identifier, password_hash, role, created_at, auth_provider, oidc_issuer, oidc_sub) SELECT id, identifier, password_hash, role, created_at, ?, ?, ? FROM users', ['local', null, null])
    await run('DROP TABLE users')
    await run('ALTER TABLE users_new RENAME TO users')
    await run(SCHEMA.users_oidc_index)
    await run('COMMIT')
  }
  catch (error) {
    await run('ROLLBACK').catch(() => {})
    throw error
  }
}

const migrateToV3 = async (run, all) => {
  const info = await all('PRAGMA table_info(users)')
  if (info.some(c => c.name === 'avatar_path')) return
  await run('ALTER TABLE users ADD COLUMN avatar_path TEXT')
}

const migrateToV4 = async (run, all) => {
  const info = await all('PRAGMA table_info(users)')
  if (info.some(c => c.name === 'cot_password_hash')) return
  await run('ALTER TABLE users ADD COLUMN cot_password_hash TEXT')
}

const runMigrations = async (run, all, get) => {
  const version = await getSchemaVersion(get)
  if (version >= SCHEMA_VERSION) return
  if (version < 2) {
    await migrateToV2(run, all)
    await setSchemaVersion(run, 2)
  }
  if (version < 3) {
    await migrateToV3(run, all)
    await setSchemaVersion(run, 3)
  }
  if (version < 4) {
    await migrateToV4(run, all)
    await setSchemaVersion(run, 4)
  }
}

const initDb = async (db, run, all, get) => {
  try {
    await run('PRAGMA journal_mode = WAL')
  }
  catch {
    // WAL not supported (e.g., network filesystem)
  }
  db.configure('busyTimeout', DB_BUSY_TIMEOUT_MS)

  await run(SCHEMA.schema_version)
  await run(SCHEMA.users)
  await runMigrations(run, all, get)
  await run(SCHEMA.sessions)
  await run(SCHEMA.pois)
  await run(SCHEMA.devices)

  if (!testPath) {
    // Bootstrap admin user on first run
    const PASSWORD_CHARS = Object.freeze('abcdefghjkmnopqrstuvwxyzABCDEFGHJKMNPQRSTUVWXYZ23456789')
    const generateRandomPassword = () =>
      Array.from(randomBytes(14), b => PASSWORD_CHARS[b % PASSWORD_CHARS.length]).join('')

    const row = await get('SELECT COUNT(*) as n FROM users')
    if (row?.n === 0) {
      const email = process.env.BOOTSTRAP_EMAIL?.trim()
      const password = process.env.BOOTSTRAP_PASSWORD
      const identifier = (email && password) ? email : 'admin'
      const plainPassword = (email && password) ? password : generateRandomPassword()

      await run(
        'INSERT INTO users (id, identifier, password_hash, role, created_at, auth_provider, oidc_issuer, oidc_sub) VALUES (?, ?, ?, ?, ?, ?, ?, ?)',
        [crypto.randomUUID(), identifier, hashPassword(plainPassword), 'admin', new Date().toISOString(), 'local', null, null],
      )

      if (!email || !password) {
        console.log(`\n[KestrelOS] No bootstrap admin configured. Default admin created. Sign in at /login with:\n\n  Identifier: ${identifier}\n  Password:   ${plainPassword}\n\n  Set BOOTSTRAP_EMAIL and BOOTSTRAP_PASSWORD to use your own credentials on first run.\n`)
      }
    }
  }
}

export async function getDb() {
  if (dbInstance) return dbInstance

  const db = new sqlite3.Database(getDbPath(), (err) => {
    if (err) {
      console.error('[db] Failed to open database:', err.message)
      throw err
    }
  })

  const run = promisify(db.run.bind(db))
  const all = promisify(db.all.bind(db))
  const get = promisify(db.get.bind(db))

  try {
    await initDb(db, run, all, get)
  }
  catch (error) {
    db.close()
    console.error('[db] Database initialization failed:', error.message)
    throw error
  }

  dbInstance = { db, run, all, get }

  registerCleanup(async () => {
    if (dbInstance) {
      try {
        await new Promise((resolve, reject) => {
          dbInstance.db.close((err) => {
            if (err) reject(err)
            else resolve()
          })
        })
      }
      catch (error) {
        console.error('[db] Error closing database during shutdown:', error?.message)
      }
      dbInstance = null
    }
  })

  return dbInstance
}

/**
 * Health check for database connection.
 * @returns {Promise<{ healthy: boolean, error?: string }>} Health status
 */
export async function healthCheck() {
  try {
    const db = await getDb()
    await db.get('SELECT 1')
    return { healthy: true }
  }
  catch (error) {
    return {
      healthy: false,
      error: error?.message || String(error),
    }
  }
}

/**
 * Database connection model documentation:
 *
 * KestrelOS uses SQLite with WAL (Write-Ahead Logging) mode for concurrent access.
 * - Single connection instance shared across all requests (singleton pattern)
 * - WAL mode allows multiple readers and one writer concurrently
 * - Connection is initialized on first getDb() call and reused thereafter
 * - Busy timeout is set to 5000ms to handle concurrent access gracefully
 * - Transactions are supported via withTransaction() helper
 *
 * Concurrency considerations:
 * - SQLite with WAL handles concurrent reads efficiently
 * - Writes are serialized (one at a time)
 * - For high write loads, consider migrating to PostgreSQL
 * - Current model is suitable for moderate traffic (< 100 req/sec)
 *
 * Connection lifecycle:
 * - Created on first getDb() call
 * - Persists for application lifetime
 * - Closed during graceful shutdown
 * - Test path can be set via setDbPathForTest() for testing
 */

/**
 * Execute a callback within a database transaction.
 * Automatically commits on success or rolls back on error.
 * @param {object} db - Database instance from getDb()
 * @param {Function} callback - Async function receiving { run, all, get }
 * @returns {Promise<any>} Result of callback
 */
export async function withTransaction(db, callback) {
  const { run } = db
  await run('BEGIN TRANSACTION')
  try {
    const result = await callback(db)
    await run('COMMIT')
    return result
  }
  catch (error) {
    await run('ROLLBACK').catch(() => {
      // Ignore rollback errors
    })
    throw error
  }
}

export function closeDb() {
  if (!dbInstance) return
  try {
    dbInstance.db.close((err) => {
      if (err) console.error('[db] Error closing database:', err.message)
    })
  }
  catch (error) {
    console.error('[db] Error closing database:', error.message)
  }
  dbInstance = null
}

export function setDbPathForTest(path) {
  testPath = path || null
  closeDb()
}