investor-flow/app/server/src/db/client.ts

// Investor Flow — SQLite DB client (node:sqlite). DESIGN.md §1 schema seam.
//
// Runtime adaptation (reversible): design specified Bun + bun:sqlite. Implemented on
// Node 26 native TS (type-stripping) + node:sqlite (built-in) so the backend has ZERO
// native dependencies. Deep-module architecture (CacheRepository, etc.) is unchanged.
import { DatabaseSync } from 'node:sqlite';
import { readFileSync, mkdirSync } from 'node:fs';
import { dirname, join, resolve } from 'node:path';
import { fileURLToPath } from 'node:url';

const __dirname = dirname(fileURLToPath(import.meta.url));
const SCHEMA_PATH = join(__dirname, 'schema.sql');
const DEFAULT_DB_DIR = join(__dirname, '..', '..', 'data');
const DEFAULT_DB_PATH = join(DEFAULT_DB_DIR, 'investor-flow.db');

export interface DbOptions {
  /** SQLite file path. Defaults to $IFLOW_DB_PATH or <server>/data/investor-flow.db. Use ':memory:' for tests. */
  path?: string;
  readonly?: boolean;
}

/** Open a SQLite connection with foreign keys + WAL enabled. */
export function createDb(opts: DbOptions = {}): DatabaseSync {
  const path = opts.path ?? process.env.IFLOW_DB_PATH ?? DEFAULT_DB_PATH;
  if (!opts.readonly && path !== ':memory:') {
    mkdirSync(dirname(path), { recursive: true });
  }
  const database = new DatabaseSync(path, {
    readOnly: opts.readonly ?? false,
    enableForeignKeyConstraints: true,
  });
  // WAL is persistent on the file; setting per-connection is harmless and ensures it.
  try {
    database.exec('PRAGMA journal_mode = WAL;');
  } catch {
    /* readOnly or disallowed — ignore */
  }
  // Concurrent queue drain + request handlers write often; wait instead of failing immediately.
  try {
    database.exec('PRAGMA busy_timeout = 5000;');
  } catch {
    /* ignore */
  }
  return database;
}

/** Apply the full schema (CREATE TABLE IF NOT EXISTS). Idempotent. */
export function initSchema(database: DatabaseSync): void {
  const sql = readFileSync(SCHEMA_PATH, 'utf8');
  database.exec(sql);
}

/** Idempotent migrations for existing databases (new columns, tables, dedupes). */
function runMigrations(db: DatabaseSync): void {
  // 1. New columns (ignore "already exists").
  const migrations: string[] = [
    `ALTER TABLE adapter_queue ADD COLUMN error TEXT`,
    `ALTER TABLE adapter_queue ADD COLUMN scheduled_for TEXT`,
    `ALTER TABLE insider_transactions ADD COLUMN accession TEXT`,
    `ALTER TABLE x_credentials ADD COLUMN fred_api_key_enc TEXT`,
    `ALTER TABLE users ADD COLUMN status TEXT NOT NULL DEFAULT 'active'`,
    // Onboarding / risk posture fields added after early installs (CREATE IF NOT EXISTS does not alter).
    `ALTER TABLE users ADD COLUMN drawdown_tolerance REAL`,
    `ALTER TABLE users ADD COLUMN backup_codes_hashed TEXT`,
    `ALTER TABLE users ADD COLUMN is_admin INTEGER NOT NULL DEFAULT 0`,
    `ALTER TABLE rotation_state ADD COLUMN signal_since TEXT`,
    `CREATE TABLE IF NOT EXISTS rotation_state (
      id         TEXT PRIMARY KEY DEFAULT 'singleton',
      signal     TEXT NOT NULL DEFAULT 'none',
      spread     REAL,
      inflow_avg REAL,
      outflow_avg REAL,
      sub_group  TEXT,
      checked_at TEXT,
      signal_since TEXT
    )`,
    `CREATE TABLE IF NOT EXISTS rotation_rank_snapshots (
      as_of_date   TEXT NOT NULL,
      symbol       TEXT NOT NULL,
      name         TEXT,
      grp          TEXT,
      rank_1m      INTEGER,
      rs_1m        REAL,
      rank_1w      INTEGER,
      rs_1w        REAL,
      leadership   TEXT,
      early_watch  INTEGER NOT NULL DEFAULT 0,
      strength     TEXT,
      PRIMARY KEY (as_of_date, symbol)
    )`,
    `CREATE INDEX IF NOT EXISTS idx_rotation_rank_date ON rotation_rank_snapshots(as_of_date DESC)`,
    // Option legs book (MVP risk contribution; M17 sleeve later).
    `CREATE TABLE IF NOT EXISTS portfolio_option_legs (
      id            TEXT PRIMARY KEY,
      owner_id      TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
      underlying    TEXT NOT NULL,
      right         TEXT NOT NULL,
      side          TEXT NOT NULL,
      strike        REAL NOT NULL,
      expiry        TEXT NOT NULL,
      contracts     REAL NOT NULL,
      premium       REAL NOT NULL,
      multiplier    INTEGER NOT NULL DEFAULT 100,
      role          TEXT NOT NULL,
      status        TEXT NOT NULL DEFAULT 'open',
      acquired_at   TEXT NOT NULL,
      note          TEXT
    )`,
    `CREATE INDEX IF NOT EXISTS idx_option_legs_owner_status
      ON portfolio_option_legs(owner_id, status)`,
    // 13F put/call indicator for options positions.
    `ALTER TABLE institution_filings ADD COLUMN put_call TEXT`,
  ];
  for (const sql of migrations) {
    try { db.exec(sql); } catch { /* column already exists */ }
  }

  // 2. De-dupe legacy institution_filings rows (keep the oldest rowid per group).
  //    Uses rowid (SQLite built-in) rather than tuple NOT IN which can hit limits
  //    with large duplicate sets. This must succeed before creating the unique index.
  try {
    db.exec(`
      DELETE FROM institution_filings
      WHERE rowid NOT IN (
        SELECT MIN(rowid)
        FROM institution_filings
        GROUP BY filer_cik, symbol, reported_quarter, form
      )
    `);
  } catch { /* empty / locked — safe to skip */ }

  // 3. Idempotency index for institution_filings (now safe after de-dupe).
  try {
    db.exec(`CREATE UNIQUE INDEX IF NOT EXISTS uq_inst_filings ON institution_filings(filer_cik, symbol, reported_quarter, form)`);
  } catch { /* ignore */ }

  // 4. Rename legacy 'Starter' watchlist to 'default'.
  try {
    db.exec(`UPDATE watchlists SET name='default' WHERE name='Starter'`);
  } catch { /* ignore */ }

  // 5. Data-quality lint table.
  try {
    db.exec(`
      CREATE TABLE IF NOT EXISTS data_quality (
        symbol TEXT NOT NULL,
        kind TEXT NOT NULL,
        last_checked_at TEXT,
        stored_count INTEGER,
        discovered_count INTEGER,
        missing_count INTEGER,
        stale INTEGER NOT NULL DEFAULT 0,
        status TEXT NOT NULL,
        detail TEXT,
        PRIMARY KEY (symbol, kind)
      )
    `);
  } catch { /* ignore */ }

  // 6. Analyst ratings / upgrades & downgrades (yahoo-finance2).
  try { db.exec(`ALTER TABLE analyst_ratings ADD COLUMN target_from REAL`); } catch { /* ignore */ }
  try { db.exec(`ALTER TABLE analyst_ratings ADD COLUMN target_to REAL`); } catch { /* ignore */ }
  try {
    db.exec(`
      CREATE TABLE IF NOT EXISTS analyst_ratings (
        symbol      TEXT NOT NULL,
        firm        TEXT NOT NULL,
        action      TEXT,
        grade_from  TEXT,
        grade_to    TEXT,
        target_from REAL,
        target_to   REAL,
        rating_date TEXT NOT NULL,
        fetched_at  TEXT NOT NULL,
        PRIMARY KEY (symbol, firm, rating_date)
      )
    `);
  } catch { /* ignore */ }
  try {
    db.exec(`
      CREATE TABLE IF NOT EXISTS analyst_consensus (
        symbol      TEXT PRIMARY KEY,
        strong_buy  INTEGER NOT NULL DEFAULT 0,
        buy         INTEGER NOT NULL DEFAULT 0,
        hold        INTEGER NOT NULL DEFAULT 0,
        sell        INTEGER NOT NULL DEFAULT 0,
        strong_sell INTEGER NOT NULL DEFAULT 0,
        fetched_at  TEXT NOT NULL
      )
      `);
    } catch { /* ignore */ }

  // 7. Alert subscriptions: add watchlist_id + enabled columns (existing DBs).
  try { db.exec(`ALTER TABLE alerts ADD COLUMN watchlist_id TEXT`); } catch { /* ignore */ }
  try { db.exec(`ALTER TABLE alerts ADD COLUMN enabled INTEGER NOT NULL DEFAULT 1`); } catch { /* ignore */ }

  // 8. Alert comparison state table.
  try {
    db.exec(`
      CREATE TABLE IF NOT EXISTS alert_comparison_state (
        symbol      TEXT NOT NULL,
        alert_type  TEXT NOT NULL,
        state       TEXT NOT NULL DEFAULT '{}',
        updated_at  TEXT NOT NULL,
        PRIMARY KEY (symbol, alert_type)
      )
      `);
    } catch { /* ignore */ }

  // 9. SMTP config table for email alerts.
  try {
    db.exec(`
      CREATE TABLE IF NOT EXISTS smtp_config (
        id          TEXT PRIMARY KEY DEFAULT 'singleton',
        host        TEXT NOT NULL DEFAULT 'smtp.mail.me.com',
        port        INTEGER NOT NULL DEFAULT 587,
        secure      INTEGER NOT NULL DEFAULT 0,
        user        TEXT,
        pass_enc    TEXT,
        from_name   TEXT NOT NULL DEFAULT 'Investor Flow',
        from_email  TEXT NOT NULL DEFAULT '',
        enabled     INTEGER NOT NULL DEFAULT 0,
        updated_at  TEXT NOT NULL
      )
    `);
  } catch { /* ignore */ }
}

let _db: DatabaseSync | null = null;
/** Process-wide singleton connection (auto-creates + inits schema on first call). */
export function db(): DatabaseSync {
  if (!_db) {
    _db = createDb();
    initSchema(_db);
    runMigrations(_db);
    // Ensure the anonymous user exists for unauthenticated watchlist/portfolio usage.
    try {
      _db.prepare("INSERT OR IGNORE INTO users (id, email, pw_hash, created_at) VALUES (?, ?, ?, ?)")
        .run('anonymous', 'anonymous@investor-flow.local', '', new Date().toISOString());
    } catch { /* table may not exist yet or already exists */ }
  }
  return _db;
}

/** CLI entry: `node src/db/client.ts` initializes the on-disk database. */
const isMain = process.argv[1] ? resolve(process.argv[1]) === fileURLToPath(import.meta.url) : false;
if (isMain) {
  const d = createDb();
  initSchema(d);
  const path = process.env.IFLOW_DB_PATH ?? DEFAULT_DB_PATH;
  const rows = d.prepare("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name").all() as Array<{ name: string }>;
  console.log(`[investor-flow] schema initialized at ${path} (${rows.length} tables)`);
  d.close();
}