watcher.ts

import fs from 'node:fs';
import path from 'node:path';
import { closeDb, getNodeId as getNodeIdQuery, initSchema, openDb } from '../../db/index.js';
import { debug, info, warn } from '../../infrastructure/logger.js';
import { isSupportedFile, normalizePath, shouldIgnore } from '../../shared/constants.js';
import { DbError } from '../../shared/errors.js';
import { createParseTreeCache, getActiveEngine } from '../parser.js';
import { type IncrementalStmts, rebuildFile } from './builder/incremental.js';
import { appendChangeEvents, buildChangeEvent, diffSymbols } from './change-journal.js';
import { appendJournalEntriesAndStampHeader } from './journal.js';

function shouldIgnorePath(filePath: string): boolean {
  const parts = filePath.split(path.sep);
  return parts.some((p) => shouldIgnore(p));
}

/** Prepare all SQL statements needed by the watcher's incremental rebuild. */
function prepareWatcherStatements(db: ReturnType<typeof openDb>): IncrementalStmts {
  return {
    insertNode: db.prepare(
      'INSERT OR IGNORE INTO nodes (name, kind, file, line, end_line) VALUES (?, ?, ?, ?, ?)',
    ),
    getNodeId: {
      get: (...params: unknown[]) => {
        const [name, kind, file, line] = params as [string, string, string, number];
        const id = getNodeIdQuery(db, name, kind, file, line);
        return id != null ? { id } : undefined;
      },
    },
    insertEdge: db.prepare(
      'INSERT INTO edges (source_id, target_id, kind, confidence, dynamic) VALUES (?, ?, ?, ?, ?)',
    ),
    countNodes: db.prepare('SELECT COUNT(*) as c FROM nodes WHERE file = ?'),
    findNodeInFile: db.prepare(
      "SELECT id, file FROM nodes WHERE name = ? AND kind IN ('function', 'method', 'class', 'interface', 'type', 'struct', 'enum', 'trait', 'record', 'module', 'constant') AND file = ?",
    ),
    findNodeByName: db.prepare(
      "SELECT id, file FROM nodes WHERE name = ? AND kind IN ('function', 'method', 'class', 'interface', 'type', 'struct', 'enum', 'trait', 'record', 'module', 'constant')",
    ),
    listSymbols: db.prepare("SELECT name, kind, line FROM nodes WHERE file = ? AND kind != 'file'"),
  };
}

/** Rebuild result shape from rebuildFile. */
interface RebuildResult {
  file: string;
  deleted?: boolean;
  event: string;
  symbolDiff: unknown;
  nodesBefore: number;
  nodesAfter: number;
  nodesAdded: number;
  nodesRemoved: number;
  edgesAdded: number;
}

/** Process a batch of pending file changes: rebuild, journal, and log. */
async function processPendingFiles(
  files: string[],
  db: ReturnType<typeof openDb>,
  rootDir: string,
  stmts: IncrementalStmts,
  engineOpts: import('../../types.js').EngineOpts,
  cache: ReturnType<typeof createParseTreeCache>,
): Promise<void> {
  const results: RebuildResult[] = [];
  for (const filePath of files) {
    // Per-file try/catch so one bad rebuild doesn't crash the watcher loop.
    // The watcher is a long-running session — any SQLite error, parse failure,
    // or filesystem race must be reported and skipped, not propagated. Issue #1176.
    try {
      const result = (await rebuildFile(db, rootDir, filePath, stmts, engineOpts, cache, {
        diffSymbols: diffSymbols as (old: unknown[], new_: unknown[]) => unknown,
      })) as RebuildResult | null;
      if (result) results.push(result);
    } catch (err: unknown) {
      const relPath = normalizePath(path.relative(rootDir, filePath));
      // Narrow with `instanceof` instead of casting: a non-Error throw (a plain
      // string, `null`, or any value a third-party dependency throws) would log
      // `(err as Error).message` as `undefined`. See Greptile review on #1182.
      const message = err instanceof Error ? err.message : String(err);
      warn(`Failed to rebuild ${relPath}: ${message} — skipping`);
      debug(err instanceof Error ? (err.stack ?? message) : String(err));
    }
  }

  if (results.length > 0) {
    writeJournalAndChangeEvents(rootDir, results);
  }

  logRebuildResults(results);
}

/** Write journal entries and change events for processed files. */
function writeJournalAndChangeEvents(rootDir: string, updates: RebuildResult[]): void {
  const entries = updates.map((r) => ({
    file: r.file,
    deleted: r.deleted || false,
  }));
  try {
    appendJournalEntriesAndStampHeader(rootDir, entries, Date.now());
  } catch (e: unknown) {
    debug(`Journal write failed (non-fatal): ${(e as Error).message}`);
  }

  const changeEvents = updates.map((r) =>
    buildChangeEvent(r.file, r.event, r.symbolDiff, {
      nodesBefore: r.nodesBefore,
      nodesAfter: r.nodesAfter,
      edgesAdded: r.edgesAdded,
    }),
  );
  try {
    appendChangeEvents(rootDir, changeEvents);
  } catch (e: unknown) {
    debug(`Change event write failed (non-fatal): ${(e as Error).message}`);
  }
}

/** Log rebuild results to the user. */
function logRebuildResults(updates: RebuildResult[]): void {
  for (const r of updates) {
    const nodeDelta = r.nodesAdded - r.nodesRemoved;
    const nodeStr = nodeDelta >= 0 ? `+${nodeDelta}` : `${nodeDelta}`;
    if (r.deleted) {
      info(`Removed: ${r.file} (-${r.nodesRemoved} nodes)`);
    } else {
      info(`Updated: ${r.file} (${nodeStr} nodes, +${r.edgesAdded} edges)`);
    }
  }
}

/** Recursively collect tracked source files for stat-based polling. */
function collectTrackedFiles(dir: string, result: string[]): void {
  let entries: fs.Dirent[];
  try {
    entries = fs.readdirSync(dir, { withFileTypes: true });
  } catch (e: unknown) {
    debug(`collectTrackedFiles: cannot read ${dir}: ${(e as Error).message}`);
    return;
  }
  for (const entry of entries) {
    if (shouldIgnore(entry.name)) continue;
    const full = path.join(dir, entry.name);
    if (entry.isDirectory()) {
      collectTrackedFiles(full, result);
    } else if (isSupportedFile(entry.name)) {
      result.push(full);
    }
  }
}

/** Shared watcher state passed between setup and watcher sub-functions. */
interface WatcherContext {
  rootDir: string;
  db: ReturnType<typeof openDb>;
  stmts: IncrementalStmts;
  engineOpts: import('../../types.js').EngineOpts;
  cache: ReturnType<typeof createParseTreeCache>;
  pending: Set<string>;
  timer: ReturnType<typeof setTimeout> | null;
  debounceMs: number;
}

/** Initialize DB, engine, cache, and statements for watch mode. */
function setupWatcher(rootDir: string, opts: { engine?: string; dbPath?: string }): WatcherContext {
  const dbPath = opts.dbPath ?? path.join(rootDir, '.codegraph', 'graph.db');
  if (!fs.existsSync(dbPath)) {
    throw new DbError('No graph.db found. Run `codegraph build` first.', { file: dbPath });
  }

  const db = openDb(dbPath);
  initSchema(db);
  const engineOpts: import('../../types.js').EngineOpts = {
    engine: (opts.engine || 'auto') as import('../../types.js').EngineMode,
    dataflow: false,
    ast: false,
  };
  const { name: engineName, version: engineVersion } = getActiveEngine(engineOpts);
  info(`Watch mode using ${engineName} engine${engineVersion ? ` (v${engineVersion})` : ''}`);

  const cache = createParseTreeCache();
  info(
    cache
      ? 'Incremental parsing enabled (native tree cache)'
      : 'Incremental parsing unavailable (full re-parse)',
  );

  const stmts = prepareWatcherStatements(db);

  return {
    rootDir,
    db,
    stmts,
    engineOpts,
    cache,
    pending: new Set<string>(),
    timer: null,
    debounceMs: 300,
  };
}

/** Schedule debounced processing of pending files. */
function scheduleDebouncedProcess(ctx: WatcherContext): void {
  if (ctx.timer) clearTimeout(ctx.timer);
  ctx.timer = setTimeout(async () => {
    const files = [...ctx.pending];
    ctx.pending.clear();
    await processPendingFiles(files, ctx.db, ctx.rootDir, ctx.stmts, ctx.engineOpts, ctx.cache);
  }, ctx.debounceMs);
}

/** Start polling-based file watcher. Returns cleanup function. */
function startPollingWatcher(ctx: WatcherContext, pollIntervalMs: number): () => void {
  const mtimeMap = new Map<string, number>();

  const initial: string[] = [];
  collectTrackedFiles(ctx.rootDir, initial);
  for (const f of initial) {
    try {
      mtimeMap.set(f, fs.statSync(f).mtimeMs);
    } catch {
      /* deleted between collect and stat */
    }
  }
  info(`Polling ${initial.length} tracked files every ${pollIntervalMs}ms`);

  const pollTimer = setInterval(() => {
    const current: string[] = [];
    collectTrackedFiles(ctx.rootDir, current);
    const currentSet = new Set(current);

    for (const f of current) {
      try {
        const mtime = fs.statSync(f).mtimeMs;
        const prev = mtimeMap.get(f);
        if (prev === undefined || mtime !== prev) {
          mtimeMap.set(f, mtime);
          ctx.pending.add(f);
        }
      } catch {
        /* deleted between collect and stat */
      }
    }

    for (const f of mtimeMap.keys()) {
      if (!currentSet.has(f)) {
        mtimeMap.delete(f);
        ctx.pending.add(f);
      }
    }

    if (ctx.pending.size > 0) {
      scheduleDebouncedProcess(ctx);
    }
  }, pollIntervalMs);

  return () => clearInterval(pollTimer);
}

/** Start native OS file watcher. Returns cleanup function. */
function startNativeWatcher(ctx: WatcherContext): () => void {
  const watcher = fs.watch(ctx.rootDir, { recursive: true }, (_eventType, filename) => {
    if (!filename) return;
    if (shouldIgnorePath(filename)) return;
    if (!isSupportedFile(filename)) return;

    ctx.pending.add(path.join(ctx.rootDir, filename));
    scheduleDebouncedProcess(ctx);
  });

  return () => watcher.close();
}

/**
 * Build journal entries for a pending-path set, detecting deletions by
 * existence check.
 *
 * `ctx.pending` is an untyped `Set<string>` — it carries no event-type
 * metadata. Without this check, a file deleted during the watch session
 * would be journaled as "changed", causing the next incremental build to
 * try to re-parse a non-existent file instead of removing it from the graph.
 * Mirrors the deletion detection in `rebuildFile` (see builder/incremental.ts).
 *
 * Exported for unit-testing; prefer `setupShutdownHandler` in production paths.
 */
export function buildFlushEntriesFromPending(
  rootDir: string,
  pending: Iterable<string>,
): Array<{ file: string; deleted: boolean }> {
  return [...pending].map((filePath) => ({
    file: normalizePath(path.relative(rootDir, filePath)),
    deleted: !fs.existsSync(filePath),
  }));
}

/** Register SIGINT handler to flush journal and clean up. */
function setupShutdownHandler(ctx: WatcherContext, cleanup: () => void): void {
  process.once('SIGINT', () => {
    info('Stopping watcher...');
    cleanup();
    if (ctx.pending.size > 0) {
      const entries = buildFlushEntriesFromPending(ctx.rootDir, ctx.pending);
      try {
        appendJournalEntriesAndStampHeader(ctx.rootDir, entries, Date.now());
      } catch (e: unknown) {
        debug(`Journal flush on exit failed (non-fatal): ${(e as Error).message}`);
      }
    }
    if (ctx.cache) ctx.cache.clear();
    closeDb(ctx.db);
    process.exit(0);
  });
}

export async function watchProject(
  rootDir: string,
  opts: { engine?: string; poll?: boolean; pollInterval?: number; dbPath?: string } = {},
): Promise<void> {
  const ctx = setupWatcher(rootDir, opts);

  const usePoll = opts.poll ?? process.platform === 'win32';
  const pollIntervalMs = opts.pollInterval ?? 2000;

  info(`Watching ${rootDir} for changes${usePoll ? ' (polling mode)' : ''}...`);
  info('Press Ctrl+C to stop.');

  const cleanup = usePoll ? startPollingWatcher(ctx, pollIntervalMs) : startNativeWatcher(ctx);

  setupShutdownHandler(ctx, cleanup);
}