/**
 * Pipeline orchestrator — runs build stages sequentially through a shared PipelineContext.
 *
 * This is the heart of the builder refactor (ROADMAP 3.9): the monolithic buildGraph()
 * is decomposed into independently testable stages that communicate via PipelineContext.
 */
import fs from 'node:fs';
import path from 'node:path';
import { performance } from 'node:perf_hooks';
import {
  acquireAdvisoryLock,
  closeDb,
  closeDbPair,
  getBuildMeta,
  initSchema,
  MIGRATIONS,
  openDb,
  purgeFilesData,
  releaseAdvisoryLock,
  setBuildMeta,
} from '../../../db/index.js';
import { detectWorkspaces, loadConfig } from '../../../infrastructure/config.js';
import { debug, info, warn } from '../../../infrastructure/logger.js';
import { loadNative } from '../../../infrastructure/native.js';
import { semverCompare } from '../../../infrastructure/update-check.js';
import { normalizePath } from '../../../shared/constants.js';
import { toErrorMessage } from '../../../shared/errors.js';
import { CODEGRAPH_VERSION } from '../../../shared/version.js';
import type {
  BetterSqlite3Database,
  BuildGraphOpts,
  BuildResult,
  Definition,
  ExtractorOutput,
  SqliteStatement,
} from '../../../types.js';
import {
  classifyNativeDrops,
  formatDropExtensionSummary,
  getActiveEngine,
  getInstalledWasmExtensions,
  NATIVE_SUPPORTED_EXTENSIONS,
  parseFilesWasmForBackfill,
} from '../../parser.js';
import { writeJournalHeader } from '../journal.js';
import { setWorkspaces } from '../resolve.js';
import { PipelineContext } from './context.js';
import {
  batchInsertNodes,
  collectFiles as collectFilesUtil,
  fileHash,
  fileStat,
  loadPathAliases,
  readFileSafe,
} from './helpers.js';
import { NativeDbProxy } from './native-db-proxy.js';
import { buildEdges } from './stages/build-edges.js';
import { buildStructure } from './stages/build-structure.js';
// Pipeline stages
import { collectFiles } from './stages/collect-files.js';
import { detectChanges, detectNoChanges } from './stages/detect-changes.js';
import { finalize } from './stages/finalize.js';
import { insertNodes } from './stages/insert-nodes.js';
import { parseFiles } from './stages/parse-files.js';
import { resolveImports } from './stages/resolve-imports.js';
import { runAnalyses } from './stages/run-analyses.js';

// ── Setup helpers ───────────────────────────────────────────────────────

function initializeEngine(ctx: PipelineContext): void {
  ctx.engineOpts = {
    engine: ctx.opts.engine || 'auto',
    dataflow: ctx.opts.dataflow !== false,
    ast: ctx.opts.ast !== false,
    // nativeDb and WAL callbacks are set later when NativeDatabase is opened
    // (deferred to skip overhead on no-op rebuilds).
    nativeDb: undefined,
    suspendJsDb: undefined,
    resumeJsDb: undefined,
  };
  const {
    name: engineName,
    version: engineVersion,
    binaryVersion: nativeBinaryVersion,
  } = getActiveEngine(ctx.engineOpts);
  ctx.engineName = engineName as 'native' | 'wasm';
  ctx.engineVersion = engineVersion;
  ctx.nativeBinaryVersion = nativeBinaryVersion;
  info(`Using ${engineName} engine${engineVersion ? ` (v${engineVersion})` : ''}`);
}

function checkEngineSchemaMismatch(ctx: PipelineContext): void {
  const lastMigration = MIGRATIONS[MIGRATIONS.length - 1] as { version: number } | undefined;
  ctx.schemaVersion = lastMigration?.version ?? 0;
  ctx.forceFullRebuild = false;
  if (!ctx.incremental) return;

  // NativeDatabase is deferred until after change detection, so always use
  // better-sqlite3 for metadata reads here. Reads are safe — WAL conflicts
  // only arise from concurrent writes.
  const meta = (key: string): string | null => getBuildMeta(ctx.db, key);

  const prevEngine = meta('engine');
  if (prevEngine && prevEngine !== ctx.engineName) {
    info(`Engine changed (${prevEngine} → ${ctx.engineName}), promoting to full rebuild.`);
    ctx.forceFullRebuild = true;
  }
  const prevSchema = meta('schema_version');
  if (prevSchema && Number(prevSchema) !== ctx.schemaVersion) {
    info(
      `Schema version changed (${prevSchema} → ${ctx.schemaVersion}), promoting to full rebuild.`,
    );
    ctx.forceFullRebuild = true;
  }
  // When the native engine is active, the Rust orchestrator writes
  // build_meta.codegraph_version = CARGO_PKG_VERSION (the binary's own value).
  // Compare against the same value here so a CI hot-swap that leaves the
  // platform package.json behind doesn't trigger a perpetual full-rebuild
  // loop on every incremental (#1066).
  const effectiveVersion =
    ctx.engineName === 'native' && ctx.nativeBinaryVersion
      ? ctx.nativeBinaryVersion
      : CODEGRAPH_VERSION;
  const prevVersion = meta('codegraph_version');
  if (prevVersion && prevVersion !== effectiveVersion) {
    info(
      `Codegraph version changed (${prevVersion} → ${effectiveVersion}), promoting to full rebuild.`,
    );
    ctx.forceFullRebuild = true;
  }
}

function warnOnEmbeddingsWipe(ctx: PipelineContext): void {
  const willBeFullBuild = !ctx.incremental || ctx.forceFullRebuild;
  if (!willBeFullBuild) return;
  let count = 0;
  try {
    count = (ctx.db.prepare('SELECT COUNT(*) AS c FROM embeddings').get() as { c: number }).c;
  } catch {
    return; // embeddings table missing — nothing to warn about
  }
  if (count === 0) return;
  warn(
    `Full rebuild will discard ${count} embedding${count === 1 ? '' : 's'}; re-run \`codegraph embed\` after the build.`,
  );
}

function loadAliases(ctx: PipelineContext): void {
  ctx.aliases = loadPathAliases(ctx.rootDir);
  if (ctx.config.aliases) {
    for (const [key, value] of Object.entries(ctx.config.aliases)) {
      if (typeof value !== 'string') {
        warn(`Alias target for "${key}" must be a string, got ${typeof value}. Skipping.`);
        continue;
      }
      const pattern = key.endsWith('/') ? `${key}*` : key;
      const target = path.resolve(ctx.rootDir, value);
      ctx.aliases.paths[pattern] = [target.endsWith('/') ? `${target}*` : `${target}/*`];
    }
  }
  if (ctx.aliases.baseUrl || Object.keys(ctx.aliases.paths).length > 0) {
    info(
      `Loaded path aliases: baseUrl=${ctx.aliases.baseUrl || 'none'}, ${Object.keys(ctx.aliases.paths).length} path mappings`,
    );
  }
}

function setupPipeline(ctx: PipelineContext): void {
  ctx.rootDir = path.resolve(ctx.rootDir);
  ctx.dbPath = ctx.opts.dbPath
    ? path.resolve(ctx.opts.dbPath)
    : path.join(ctx.rootDir, '.codegraph', 'graph.db');

  // Detect whether native engine is available.
  const enginePref = ctx.opts.engine || 'auto';
  const native = enginePref !== 'wasm' ? loadNative() : null;
  ctx.nativeAvailable = !!native?.NativeDatabase;

  // Always use better-sqlite3 for setup — it's cheap (~4ms) and only needed
  // for metadata reads (schema mismatch check). NativeDatabase.openReadWrite
  // is deferred to tryNativeOrchestrator, saving ~60ms on incremental builds
  // where the Rust orchestrator handles the full pipeline, and avoiding the
  // cost entirely on no-op builds that exit before reaching the orchestrator.
  const dir = path.dirname(ctx.dbPath);
  if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
  ctx.db = openDb(ctx.dbPath);
  initSchema(ctx.db);

  ctx.config = loadConfig(ctx.rootDir);
  // Merge caller-supplied excludes on top of the file-config excludes so
  // programmatic callers (e.g. benchmark scripts) can extend exclusion
  // without mutating .codegraphrc.json. Native orchestrator picks this up
  // automatically — it reads exclude off the serialized ctx.config below.
  if (ctx.opts.exclude?.length) {
    ctx.config = {
      ...ctx.config,
      exclude: [...(ctx.config.exclude ?? []), ...ctx.opts.exclude],
    };
  }
  ctx.incremental =
    ctx.opts.incremental !== false && ctx.config.build && ctx.config.build.incremental !== false;

  initializeEngine(ctx);
  checkEngineSchemaMismatch(ctx);
  warnOnEmbeddingsWipe(ctx);
  loadAliases(ctx);

  // Workspace packages (monorepo)
  const workspaces = detectWorkspaces(ctx.rootDir);
  if (workspaces.size > 0) {
    setWorkspaces(ctx.rootDir, workspaces);
    info(`Detected ${workspaces.size} workspace packages`);
  }

  ctx.timing.setupMs = performance.now() - ctx.buildStart;
}

function formatTimingResult(ctx: PipelineContext): BuildResult {
  const t = ctx.timing;
  return {
    phases: {
      setupMs: +(t.setupMs ?? 0).toFixed(1),
      collectMs: +(t.collectMs ?? 0).toFixed(1),
      detectMs: +(t.detectMs ?? 0).toFixed(1),
      parseMs: +(t.parseMs ?? 0).toFixed(1),
      insertMs: +(t.insertMs ?? 0).toFixed(1),
      resolveMs: +(t.resolveMs ?? 0).toFixed(1),
      edgesMs: +(t.edgesMs ?? 0).toFixed(1),
      structureMs: +(t.structureMs ?? 0).toFixed(1),
      rolesMs: +(t.rolesMs ?? 0).toFixed(1),
      astMs: +(t.astMs ?? 0).toFixed(1),
      complexityMs: +(t.complexityMs ?? 0).toFixed(1),
      cfgMs: +(t.cfgMs ?? 0).toFixed(1),
      dataflowMs: +(t.dataflowMs ?? 0).toFixed(1),
      finalizeMs: +(t.finalizeMs ?? 0).toFixed(1),
    },
  };
}

// ── NativeDb lifecycle helpers ──────────────────────────────────────────

/** Checkpoint WAL through rusqlite and close the native connection. */
function closeNativeDb(ctx: PipelineContext, label: string): void {
  if (!ctx.nativeDb) return;
  try {
    ctx.nativeDb.exec('PRAGMA wal_checkpoint(TRUNCATE)');
  } catch (e) {
    debug(`${label} WAL checkpoint failed: ${toErrorMessage(e)}`);
  }
  try {
    ctx.nativeDb.close();
  } catch (e) {
    debug(`${label} nativeDb close failed: ${toErrorMessage(e)}`);
  }
  ctx.nativeDb = undefined;
}

/** Try to reopen the native connection for a given pipeline phase. */
function reopenNativeDb(ctx: PipelineContext, label: string): void {
  if ((ctx.opts.engine ?? 'auto') === 'wasm') return;
  const native = loadNative();
  if (!native?.NativeDatabase) return;
  try {
    ctx.nativeDb = native.NativeDatabase.openReadWrite(ctx.dbPath);
  } catch (e) {
    debug(`reopen nativeDb for ${label} failed: ${toErrorMessage(e)}`);
    ctx.nativeDb = undefined;
  }
}

/** Close nativeDb and clear stale references in engineOpts. */
function suspendNativeDb(ctx: PipelineContext, label: string): void {
  closeNativeDb(ctx, label);
  if (ctx.engineOpts?.nativeDb) {
    ctx.engineOpts.nativeDb = undefined;
  }
}

/**
 * After native writes, reopen the JS db connection to get a fresh page cache.
 * Rusqlite WAL truncation invalidates better-sqlite3's internal WAL index,
 * causing SQLITE_CORRUPT on the next read (#715, #736).
 */
function refreshJsDb(ctx: PipelineContext): void {
  try {
    ctx.db.close();
  } catch (e) {
    debug(`refreshJsDb close failed: ${toErrorMessage(e)}`);
  }
  ctx.db = openDb(ctx.dbPath);
}

// ── Native orchestrator types ──────────────────────────────────────────

interface NativeOrchestratorResult {
  phases: Record<string, number>;
  earlyExit?: boolean;
  nodeCount?: number;
  edgeCount?: number;
  fileCount?: number;
  changedFiles?: string[];
  changedCount?: number;
  removedCount?: number;
  isFullBuild?: boolean;
  /** Whether the Rust pipeline handled the structure phase (small-incremental fast path). */
  structureHandled?: boolean;
  /** Whether the Rust pipeline wrote AST/complexity/CFG/dataflow to DB. */
  analysisComplete?: boolean;
}

// ── Native orchestrator helpers ───────────────────────────────────────

/** Determine whether the native orchestrator should be skipped. Returns a reason string, or null if it should run. */
function shouldSkipNativeOrchestrator(ctx: PipelineContext): string | null {
  if (ctx.forceFullRebuild) return 'forceFullRebuild';
  // v3.9.0 addon had buggy incremental purge (wrong SQL on analysis tables,
  // scoped removal over-detection). Fixed in v3.9.1 by PR #865. Gate on
  // < 3.9.1 so v3.9.1+ uses the fast Rust orchestrator path.
  const orchestratorBuggy = !!ctx.engineVersion && semverCompare(ctx.engineVersion, '3.9.1') < 0;
  if (orchestratorBuggy) return `buggy addon ${ctx.engineVersion}`;
  if (ctx.engineName !== 'native') return `engine=${ctx.engineName}`;
  return null;
}

/** Checkpoint WAL through rusqlite, close nativeDb, and reopen better-sqlite3.
 *  Returns false if the DB reopen fails (caller should return partial result). */
function handoffWalAfterNativeBuild(ctx: PipelineContext): boolean {
  closeNativeDb(ctx, 'post-native-build');
  try {
    ctx.db.close();
  } catch (e) {
    debug(`handoffWal JS db close failed: ${toErrorMessage(e)}`);
  }
  try {
    ctx.db = openDb(ctx.dbPath);
    return true;
  } catch (reopenErr) {
    warn(`Failed to reopen DB after native build: ${(reopenErr as Error).message}`);
    return false;
  }
}

/**
 * Reconstruct fileSymbols from the DB after a native orchestrator build.
 * When `scopeFiles` is provided, only loads those files (for analysis-only).
 * When omitted, loads all files (needed for structure rebuilds).
 */
function reconstructFileSymbolsFromDb(
  ctx: PipelineContext,
  scopeFiles?: string[],
): Map<string, ExtractorOutput> {
  let query =
    'SELECT file, name, kind, line, end_line as endLine FROM nodes WHERE file IS NOT NULL';
  const params: string[] = [];
  if (scopeFiles && scopeFiles.length > 0) {
    const placeholders = scopeFiles.map(() => '?').join(',');
    query += ` AND file IN (${placeholders})`;
    params.push(...scopeFiles);
  }
  query += ' ORDER BY file, line';

  const rows = ctx.db.prepare(query).all(...params) as {
    file: string;
    name: string;
    kind: string;
    line: number;
    endLine: number | null;
  }[];

  const fileSymbols = new Map<string, ExtractorOutput>();
  for (const row of rows) {
    let entry = fileSymbols.get(row.file);
    if (!entry) {
      entry = {
        definitions: [],
        calls: [],
        imports: [],
        classes: [],
        exports: [],
        typeMap: new Map(),
      };
      fileSymbols.set(row.file, entry);
    }
    entry.definitions.push({
      name: row.name,
      kind: row.kind as Definition['kind'],
      line: row.line,
      endLine: row.endLine ?? undefined,
    });
  }

  // Populate import/export counts from DB edges so buildStructure
  // computes correct import_count/export_count in node_metrics.
  // The extractor arrays aren't persisted to the DB, so we derive
  // counts from edge data instead (#804).
  const importCountRows = ctx.db
    .prepare(
      `SELECT n.file, COUNT(*) AS cnt
       FROM edges e JOIN nodes n ON e.source_id = n.id
       WHERE e.kind IN ('imports', 'imports-type', 'dynamic-imports')
         AND n.file IS NOT NULL
       GROUP BY n.file`,
    )
    .all() as { file: string; cnt: number }[];
  for (const row of importCountRows) {
    const entry = fileSymbols.get(row.file);
    if (entry) entry.imports = new Array(row.cnt) as ExtractorOutput['imports'];
  }

  const exportCountRows = ctx.db
    .prepare(
      `SELECT n_tgt.file, COUNT(DISTINCT n_tgt.id) AS cnt
       FROM edges e
       JOIN nodes n_tgt ON e.target_id = n_tgt.id
       JOIN nodes n_src ON e.source_id = n_src.id
       WHERE e.kind IN ('imports', 'imports-type', 'reexports')
         AND n_tgt.file IS NOT NULL
         AND n_src.file != n_tgt.file
       GROUP BY n_tgt.file`,
    )
    .all() as { file: string; cnt: number }[];
  for (const row of exportCountRows) {
    const entry = fileSymbols.get(row.file);
    if (entry) entry.exports = new Array(row.cnt) as ExtractorOutput['exports'];
  }

  return fileSymbols;
}

/**
 * Run JS buildStructure() after native orchestrator to fill directory nodes + contains edges.
 * For full builds, passes changedFiles=null (full rebuild).
 * For incremental builds, passes the changed file list to scope the update.
 */
async function runPostNativeStructure(
  ctx: PipelineContext,
  allFileSymbols: Map<string, ExtractorOutput>,
  isFullBuild: boolean,
  changedFiles: string[] | undefined,
): Promise<number> {
  const structureStart = performance.now();
  try {
    const directories = new Set<string>();
    for (const relPath of allFileSymbols.keys()) {
      const parts = relPath.split('/');
      for (let i = 1; i < parts.length; i++) {
        directories.add(parts.slice(0, i).join('/'));
      }
    }

    const lineCountMap = new Map<string, number>();
    const cachedLineCounts = ctx.db
      .prepare(
        `SELECT n.name AS file, m.line_count
         FROM node_metrics m JOIN nodes n ON m.node_id = n.id
         WHERE n.kind = 'file'`,
      )
      .all() as Array<{ file: string; line_count: number }>;
    for (const row of cachedLineCounts) {
      lineCountMap.set(row.file, row.line_count);
    }

    // Full builds need null (rebuild everything). Incremental builds pass the
    // changed file list so buildStructure only updates those files' metrics
    // and contains edges — matching the JS pipeline's medium-incremental path.
    const changedFilePaths = isFullBuild || !changedFiles?.length ? null : changedFiles;
    const { buildStructure: buildStructureFn } = (await import(
      '../../../features/structure.js'
    )) as {
      buildStructure: (
        db: typeof ctx.db,
        fileSymbols: Map<string, ExtractorOutput>,
        rootDir: string,
        lineCountMap: Map<string, number>,
        directories: Set<string>,
        changedFiles: string[] | null,
      ) => void;
    };
    buildStructureFn(
      ctx.db,
      allFileSymbols,
      ctx.rootDir,
      lineCountMap,
      directories,
      changedFilePaths,
    );
    debug(
      `Structure phase completed after native orchestrator${changedFilePaths ? ` (${changedFilePaths.length} files)` : ' (full)'}`,
    );
  } catch (err) {
    warn(`Structure phase failed after native build: ${toErrorMessage(err)}`);
  }
  return performance.now() - structureStart;
}

/**
 * JS fallback for AST/complexity/CFG/dataflow analysis after native orchestrator.
 * Used when the Rust addon doesn't include analysis persistence (older addon
 * version) or when analysis failed on the Rust side.
 */
async function runPostNativeAnalysis(
  ctx: PipelineContext,
  allFileSymbols: Map<string, ExtractorOutput>,
  changedFiles: string[] | undefined,
): Promise<{ astMs: number; complexityMs: number; cfgMs: number; dataflowMs: number }> {
  const timing = { astMs: 0, complexityMs: 0, cfgMs: 0, dataflowMs: 0 };

  // Scope analysis fileSymbols to changed files only
  let analysisFileSymbols: Map<string, ExtractorOutput>;
  if (changedFiles && changedFiles.length > 0) {
    analysisFileSymbols = new Map();
    for (const f of changedFiles) {
      const entry = allFileSymbols.get(f);
      if (entry) analysisFileSymbols.set(f, entry);
    }
  } else {
    analysisFileSymbols = allFileSymbols;
  }

  // Reopen nativeDb for analysis features (suspend/resume WAL pattern).
  const native = loadNative();
  if (native?.NativeDatabase) {
    try {
      ctx.nativeDb = native.NativeDatabase.openReadWrite(ctx.dbPath);
      if (ctx.engineOpts) ctx.engineOpts.nativeDb = ctx.nativeDb;
    } catch {
      ctx.nativeDb = undefined;
      if (ctx.engineOpts) ctx.engineOpts.nativeDb = undefined;
    }
  }

  // Flush JS WAL pages once so Rust can see them, then no-op callbacks.
  // Previously each feature called wal_checkpoint(TRUNCATE) individually
  // (~68ms each × 3-4 features). One FULL checkpoint suffices.
  if (ctx.nativeDb && ctx.engineOpts) {
    ctx.db.pragma('wal_checkpoint(FULL)');
    ctx.engineOpts.suspendJsDb = () => {};
    ctx.engineOpts.resumeJsDb = () => {};
  }

  try {
    const { runAnalyses: runAnalysesFn } = (await import('../../../ast-analysis/engine.js')) as {
      runAnalyses: (
        db: BetterSqlite3Database,
        fileSymbols: Map<string, ExtractorOutput>,
        rootDir: string,
        opts: Record<string, unknown>,
        engineOpts?: Record<string, unknown>,
      ) => Promise<{ astMs?: number; complexityMs?: number; cfgMs?: number; dataflowMs?: number }>;
    };
    const result = await runAnalysesFn(
      ctx.db,
      analysisFileSymbols,
      ctx.rootDir,
      ctx.opts as Record<string, unknown>,
      ctx.engineOpts as unknown as Record<string, unknown> | undefined,
    );
    timing.astMs = result.astMs ?? 0;
    timing.complexityMs = result.complexityMs ?? 0;
    timing.cfgMs = result.cfgMs ?? 0;
    timing.dataflowMs = result.dataflowMs ?? 0;
  } catch (err) {
    warn(`Analysis phases failed after native build: ${toErrorMessage(err)}`);
  }

  // Close nativeDb after analyses — TRUNCATE checkpoint flushes all Rust
  // WAL writes so JS and external readers can see them. Runs once after
  // all analysis features complete (not per-feature).
  if (ctx.nativeDb) {
    try {
      ctx.nativeDb.exec('PRAGMA wal_checkpoint(TRUNCATE)');
    } catch {
      /* ignore checkpoint errors */
    }
    try {
      ctx.nativeDb.close();
    } catch {
      /* ignore close errors */
    }
    ctx.nativeDb = undefined;
    if (ctx.engineOpts) {
      ctx.engineOpts.nativeDb = undefined;
      ctx.engineOpts.suspendJsDb = undefined;
      ctx.engineOpts.resumeJsDb = undefined;
    }
  }

  return timing;
}

/** Format timing result from native orchestrator phases + JS post-processing. */
function formatNativeTimingResult(
  p: Record<string, number>,
  structurePatchMs: number,
  analysisTiming: { astMs: number; complexityMs: number; cfgMs: number; dataflowMs: number },
): BuildResult {
  return {
    phases: {
      setupMs: +(p.setupMs ?? 0).toFixed(1),
      collectMs: +(p.collectMs ?? 0).toFixed(1),
      detectMs: +(p.detectMs ?? 0).toFixed(1),
      parseMs: +(p.parseMs ?? 0).toFixed(1),
      insertMs: +(p.insertMs ?? 0).toFixed(1),
      resolveMs: +(p.resolveMs ?? 0).toFixed(1),
      edgesMs: +(p.edgesMs ?? 0).toFixed(1),
      structureMs: +((p.structureMs ?? 0) + structurePatchMs).toFixed(1),
      rolesMs: +(p.rolesMs ?? 0).toFixed(1),
      astMs: +(analysisTiming.astMs ?? 0).toFixed(1),
      complexityMs: +(analysisTiming.complexityMs ?? 0).toFixed(1),
      cfgMs: +(analysisTiming.cfgMs ?? 0).toFixed(1),
      dataflowMs: +(analysisTiming.dataflowMs ?? 0).toFixed(1),
      finalizeMs: +(p.finalizeMs ?? 0).toFixed(1),
    },
  };
}

/** Try the native build orchestrator. Returns a BuildResult on success, undefined to fall through to JS pipeline. */
async function tryNativeOrchestrator(
  ctx: PipelineContext,
): Promise<BuildResult | undefined | 'early-exit'> {
  const skipReason = shouldSkipNativeOrchestrator(ctx);
  if (skipReason) {
    debug(`Skipping native orchestrator: ${skipReason}`);
    return undefined;
  }

  // Open NativeDatabase on demand — deferred from setupPipeline to skip the
  // ~60ms cost on no-op/early-exit builds. Close the better-sqlite3 connection
  // first to avoid dual-connection WAL corruption.
  if (!ctx.nativeDb && ctx.nativeAvailable) {
    const native = loadNative();
    if (native?.NativeDatabase) {
      try {
        // Close better-sqlite3 before opening rusqlite to avoid WAL conflicts.
        // Uses raw close() instead of closeDb() intentionally — the advisory lock
        // is kept and transferred to the NativeDbProxy below, not released here.
        ctx.db.close();
        acquireAdvisoryLock(ctx.dbPath);
        ctx.nativeDb = native.NativeDatabase.openReadWrite(ctx.dbPath);
        ctx.nativeDb.initSchema();
        // Replace ctx.db with a NativeDbProxy so post-native JS fallback
        // (structure, analysis) can use it without reopening better-sqlite3.
        const proxy = new NativeDbProxy(ctx.nativeDb);
        proxy.__lockPath = `${ctx.dbPath}.lock`;
        ctx.db = proxy as unknown as typeof ctx.db;
        ctx.nativeFirstProxy = true;
      } catch (err) {
        warn(`NativeDatabase setup failed, falling back to JS: ${toErrorMessage(err)}`);
        try {
          ctx.nativeDb?.close();
        } catch (e) {
          debug(`tryNativeOrchestrator: close failed during fallback: ${toErrorMessage(e)}`);
        }
        ctx.nativeDb = undefined;
        ctx.nativeFirstProxy = false; // defensive: reset in case future refactors move the assignment above throwing lines
        releaseAdvisoryLock(`${ctx.dbPath}.lock`);
        // Reopen better-sqlite3 for JS pipeline fallback
        ctx.db = openDb(ctx.dbPath);
      }
    }
  }

  if (!ctx.nativeDb?.buildGraph) return undefined;

  const resultJson = ctx.nativeDb.buildGraph(
    ctx.rootDir,
    JSON.stringify(ctx.config),
    JSON.stringify(ctx.aliases),
    JSON.stringify(ctx.opts),
  );
  const result = JSON.parse(resultJson) as NativeOrchestratorResult;

  if (result.earlyExit) {
    info('No changes detected');
    // Even on no-op rebuilds, dropped-language files added since the last
    // full build are still missing from `nodes`/`file_hashes` (#1083), and
    // WASM-only files deleted from disk leave stale rows behind (#1073).
    // The orchestrator's file_collector skipped them, so its earlyExit
    // doesn't imply DB consistency. Run the gap repair before returning.
    const gap = detectDroppedLanguageGap(ctx);
    if (gap.missingAbs.length > 0 || gap.staleRel.length > 0) {
      await backfillNativeDroppedFiles(ctx, gap);
    }
    closeDbPair({ db: ctx.db, nativeDb: ctx.nativeDb });
    return 'early-exit';
  }

  // Log incremental status to match JS pipeline output
  const changed = result.changedCount ?? 0;
  const removed = result.removedCount ?? 0;
  if (!result.isFullBuild && (changed > 0 || removed > 0)) {
    info(`Incremental: ${changed} changed, ${removed} removed`);
  }

  const p = result.phases;

  // Sync build_meta so JS-side version/engine checks work on next build.
  // Use the binary's CARGO_PKG_VERSION (ctx.nativeBinaryVersion), not the
  // platform package.json version (ctx.engineVersion). The Rust side's
  // check_version_mismatch compares against CARGO_PKG_VERSION; writing
  // the package.json value would create a permanent mismatch whenever
  // the binary and platform package.json diverge — e.g., CI hot-swap
  // via ci-install-native.mjs (#1066) — forcing every subsequent build
  // to be a full rebuild.
  //
  // When the native addon doesn't expose engineVersion() (older addon),
  // fall back to CODEGRAPH_VERSION — same fallback used by both
  // checkEngineSchemaMismatch (read path) and persistBuildMetadata
  // (the JS-pipeline write path in finalize.ts). Using ctx.engineVersion
  // here would re-introduce the asymmetry this PR fixes for that case.
  const nativeVersionForMeta = ctx.nativeBinaryVersion || CODEGRAPH_VERSION;
  setBuildMeta(ctx.db, {
    engine: ctx.engineName,
    engine_version: nativeVersionForMeta,
    codegraph_version: nativeVersionForMeta,
    schema_version: String(ctx.schemaVersion),
    built_at: new Date().toISOString(),
  });

  info(
    `Native build orchestrator completed: ${result.nodeCount ?? 0} nodes, ${result.edgeCount ?? 0} edges, ${result.fileCount ?? 0} files`,
  );

  // ── Post-native structure + analysis ──────────────────────────────
  let analysisTiming = {
    astMs: +(p.astMs ?? 0),
    complexityMs: +(p.complexityMs ?? 0),
    cfgMs: +(p.cfgMs ?? 0),
    dataflowMs: +(p.dataflowMs ?? 0),
  };
  let structurePatchMs = 0;
  // Skip JS structure when the Rust pipeline's small-incremental fast path
  // already handled it. For full builds and large incrementals where Rust
  // skipped structure, we must run the JS fallback.
  const needsStructure = !result.structureHandled;
  // When the Rust addon doesn't include analysis persistence (older addon
  // version or analysis failed), fall back to JS-side analysis.
  const needsAnalysisFallback =
    !result.analysisComplete &&
    (ctx.opts.ast !== false ||
      ctx.opts.complexity !== false ||
      ctx.opts.cfg !== false ||
      ctx.opts.dataflow !== false);

  if (needsStructure || needsAnalysisFallback) {
    // When analysis fallback is needed, handoff to better-sqlite3 — the
    // analysis engine uses the suspend/resume WAL pattern that requires a
    // real better-sqlite3 connection, not the NativeDbProxy.
    if (needsAnalysisFallback && ctx.nativeFirstProxy) {
      closeNativeDb(ctx, 'pre-analysis-fallback');
      ctx.db = openDb(ctx.dbPath);
      ctx.nativeFirstProxy = false;
    } else if (!ctx.nativeFirstProxy && !handoffWalAfterNativeBuild(ctx)) {
      // DB reopen failed — return partial result
      return formatNativeTimingResult(p, 0, analysisTiming);
    }

    const fileSymbols = reconstructFileSymbolsFromDb(ctx);

    if (needsStructure) {
      structurePatchMs = await runPostNativeStructure(
        ctx,
        fileSymbols,
        !!result.isFullBuild,
        result.changedFiles,
      );
    }

    if (needsAnalysisFallback) {
      analysisTiming = await runPostNativeAnalysis(ctx, fileSymbols, result.changedFiles);
    }
  }

  // Engine parity: the native orchestrator silently drops files whose
  // Rust extractor/grammar is missing or fails (e.g. HCL, Scala, Swift on
  // stale native binaries). WASM handles those — backfill via WASM so both
  // engines process the same file set (#967).
  //
  // Detect the gap once (fs walk + 2 DB queries, ~20–30ms) and use it for
  // both gating and the backfill itself. On dirty incrementals/full builds
  // the orchestrator signals trigger backfill, so the walk happens once
  // (instead of redundantly inside backfill). On quiet incrementals we
  // still pay the walk so we can detect brand-new files in dropped-language
  // extensions — a gap that the orchestrator's `detect_removed_files`
  // filter (#1070) leaves open (#1083, #1091). The pre-check is cheap
  // because the expensive part (WASM re-parse of the missing set) is
  // gated below.
  const removedCount = result.removedCount ?? 0;
  const changedCount = result.changedCount ?? 0;
  const gap = detectDroppedLanguageGap(ctx);
  if (
    result.isFullBuild ||
    removedCount > 0 ||
    changedCount > 0 ||
    gap.missingAbs.length > 0 ||
    gap.staleRel.length > 0
  ) {
    await backfillNativeDroppedFiles(ctx, gap);
  }

  closeDbPair({ db: ctx.db, nativeDb: ctx.nativeDb });
  return formatNativeTimingResult(p, structurePatchMs, analysisTiming);
}

/** Files the native orchestrator silently dropped — the working set for backfill. */
interface DroppedLanguageGap {
  /** Relative paths (normalized) of files missing from `nodes` or `file_hashes`. */
  missingRel: string[];
  /** Absolute paths, aligned by index with `missingRel`. */
  missingAbs: string[];
  /**
   * Relative paths of WASM-only files present in DB but absent from disk (#1073).
   * Rust's `detect_removed_files` filter (#1070) skips these, so the JS-side
   * backfill must purge them. Always disjoint from `missingRel`.
   */
  staleRel: string[];
}

/**
 * Inputs to {@link computeWasmOnlyStaleFiles}. Sets are passed in so the helper
 * is pure and unit-testable independently of `getInstalledWasmExtensions` and
 * the `NATIVE_SUPPORTED_EXTENSIONS` global state.
 */
export interface WasmOnlyStaleFilesInput {
  /** Distinct `file` values from the `nodes` table. */
  existingNodes: ReadonlySet<string>;
  /** Distinct `file` values from the `file_hashes` table. */
  existingHashes: ReadonlySet<string>;
  /** Relative paths currently on disk (from `collectFilesUtil`). */
  expected: ReadonlySet<string>;
  /** Lowercased extensions whose WASM grammar is installed. */
  installedExts: ReadonlySet<string>;
  /** Extensions covered by the Rust addon — Rust owns deletion for these. */
  nativeSupported: ReadonlySet<string>;
}

/**
 * Compute the WASM-only files present in the DB but missing from disk (#1073).
 *
 * Returns relative paths that:
 *   - appear in `existingNodes` or `existingHashes` (in DB),
 *   - are absent from `expected` (not on disk),
 *   - have an extension installed for WASM, AND
 *   - have an extension NOT covered by `nativeSupported` — Rust's
 *     `purge_changed_files` handles deletion for natively-supported extensions
 *     via its own `detect_removed_files`, so the caller must not double-purge.
 *
 * Extensions are lowercased before lookup to match the registry and Rust's
 * `LanguageKind::from_extension` (which normalises case for the languages
 * where both cases are conventional, e.g. R's `.r` / `.R`).
 *
 * DB paths are forced to forward slashes before comparison with `expected`
 * (which is always normalised). The on-disk invariant is that DB rows are
 * written with forward slashes, but a stale row written by older code on
 * Windows could carry back-slashes — normalising here makes the comparison
 * platform-safe and prevents false-positive purges of live rows. We replace
 * `\\` explicitly (rather than calling `normalizePath`, which only touches
 * `path.sep`) so the defence works when running on POSIX against a DB that
 * was migrated from Windows.
 *
 * Exported for unit testing.
 */
export function computeWasmOnlyStaleFiles(input: WasmOnlyStaleFilesInput): string[] {
  const { existingNodes, existingHashes, expected, installedExts, nativeSupported } = input;
  const stale: string[] = [];
  const seen = new Set<string>();
  const consider = (rawRel: string): void => {
    const rel = rawRel.replace(/\\/g, '/');
    if (expected.has(rel) || seen.has(rel)) return;
    const ext = path.extname(rel).toLowerCase();
    if (nativeSupported.has(ext)) return;
    if (!installedExts.has(ext)) return;
    seen.add(rel);
    // Push the ORIGINAL raw path (not the normalised form) so the eventual
    // `DELETE FROM nodes WHERE file = ?` predicate in `purgeFilesData`
    // matches the actual stored row. The dedup `seen` set keeps the
    // normalised form so a file written once with `\` and once with `/`
    // is still treated as one entry — but the value the SQL sees has to
    // be byte-identical to what's on disk in the DB.
    stale.push(rawRel);
  };
  for (const rel of existingNodes) consider(rel);
  for (const rel of existingHashes) consider(rel);
  return stale;
}

/**
 * Group relative paths by their lowercased extension. Shape matches the bucket
 * type that `formatDropExtensionSummary` consumes, so callers can render a
 * log-friendly per-extension summary without going through `classifyNativeDrops`
 * when the reason is already known (e.g. the stale-purge path where every path
 * is guaranteed `unsupported-by-native`).
 */
function groupByExtension(relPaths: Iterable<string>): Map<string, string[]> {
  const buckets = new Map<string, string[]>();
  for (const rel of relPaths) {
    const ext = path.extname(rel).toLowerCase();
    let list = buckets.get(ext);
    if (!list) {
      list = [];
      buckets.set(ext, list);
    }
    list.push(rel);
  }
  return buckets;
}

/**
 * Detect files the native orchestrator silently dropped.
 *
 * Walks the filesystem and compares against `nodes` + `file_hashes`. A file
 * is "missing" if it's absent from EITHER table — both must be present for
 * the fast-skip pre-flight (#1054) to work, and the two can diverge (e.g.
 * legacy DBs where `nodes` was populated but `file_hashes` was not).
 *
 * Restricted to files with an installed WASM grammar; extensions in
 * `LANGUAGE_REGISTRY` without a shipped grammar (e.g. groovy on minimal
 * installs) can't be parsed by either engine, so they're not a native
 * regression — excluding them keeps the warn count in
 * `backfillNativeDroppedFiles` meaningful.
 *
 * Also detects WASM-only files deleted from disk (#1073). Rust's
 * `detect_removed_files` filter (#1070) skips files outside its supported
 * extensions, so deletions of WASM-only languages don't reach the native
 * purge path; the rest of the backfill only inserts rows, so without this
 * step stale `nodes`/`file_hashes` rows would linger across incremental
 * rebuilds until the next full rebuild.
 *
 * Cheap (no DB handoff, no parsing): used both to gate the backfill call
 * and as its working set. NativeDbProxy supports `.prepare().all()`, so
 * this works whether `ctx.db` is a proxy or a real better-sqlite3
 * connection — letting us skip the close-native / reopen-better-sqlite3
 * cost when there's nothing to backfill.
 */
function detectDroppedLanguageGap(ctx: PipelineContext): DroppedLanguageGap {
  const collected = collectFilesUtil(ctx.rootDir, [], ctx.config, new Set<string>());
  const expected = new Set(
    collected.files.map((f) => normalizePath(path.relative(ctx.rootDir, f))),
  );

  const existingNodeRows = ctx.db
    .prepare("SELECT DISTINCT file FROM nodes WHERE kind = 'file'")
    .all() as Array<{ file: string }>;
  const existingNodes = new Set(existingNodeRows.map((r) => r.file));

  let existingHashes = new Set<string>();
  try {
    const existingHashRows = ctx.db
      .prepare('SELECT DISTINCT file FROM file_hashes')
      .all() as Array<{ file: string }>;
    existingHashes = new Set(existingHashRows.map((r) => r.file));
  } catch (e) {
    // file_hashes table may not exist on legacy DBs; treat as fully missing
    // so the backfill writes rows on the upsert path below.
    debug(
      `detectDroppedLanguageGap: file_hashes read failed (table may not exist): ${toErrorMessage(e)}`,
    );
  }

  const installedExts = getInstalledWasmExtensions();
  const missingRel: string[] = [];
  const missingAbs: string[] = [];
  for (const rel of expected) {
    if (existingNodes.has(rel) && existingHashes.has(rel)) continue;
    const ext = path.extname(rel).toLowerCase();
    if (!installedExts.has(ext)) continue;
    missingRel.push(rel);
    missingAbs.push(path.join(ctx.rootDir, rel));
  }

  const staleRel = computeWasmOnlyStaleFiles({
    existingNodes,
    existingHashes,
    expected,
    installedExts,
    nativeSupported: NATIVE_SUPPORTED_EXTENSIONS,
  });

  return { missingRel, missingAbs, staleRel };
}

/**
 * Backfill files that the native orchestrator silently dropped during parse.
 * Falls back to WASM + inserts file/symbol nodes so engine counts match (#967).
 *
 * Also purges stale rows for WASM-only files deleted from disk (#1073), which
 * Rust's `detect_removed_files` filter (#1070) skips.
 *
 * Accepts a pre-computed `gap` from `detectDroppedLanguageGap` so the caller
 * can use the same scan for both gating and the actual backfill — avoiding
 * a redundant fs walk when the orchestrator's signals already triggered.
 */
async function backfillNativeDroppedFiles(
  ctx: PipelineContext,
  gap: DroppedLanguageGap,
): Promise<void> {
  const { missingRel, missingAbs, staleRel } = gap;
  if (missingAbs.length === 0 && staleRel.length === 0) return;

  // Now that we know there's work to do, hand off to better-sqlite3 (needed
  // for the INSERT path below).
  if (ctx.nativeFirstProxy) {
    closeNativeDb(ctx, 'pre-parity-backfill');
    ctx.db = openDb(ctx.dbPath);
    ctx.nativeFirstProxy = false;
  }

  const dbConn = ctx.db as unknown as BetterSqlite3Database;

  // Purge WASM-only files that were deleted from disk (#1073). Rust's
  // detect_removed_files skips them and the insert path below never visits
  // them, so without this their rows would persist across rebuilds until the
  // next full rebuild reset the DB.
  if (staleRel.length > 0) {
    // `computeWasmOnlyStaleFiles` guarantees every path here has an extension
    // outside NATIVE_SUPPORTED_EXTENSIONS, so `classifyNativeDrops` would
    // always bucket 100% into `unsupported-by-native`. Build the extension
    // summary directly to avoid a redundant classification pass.
    const staleByExt = groupByExtension(staleRel);
    info(
      `Detected ${staleRel.length} deleted WASM-only file(s) the native orchestrator skipped; purging stale rows: ${formatDropExtensionSummary(staleByExt)}`,
    );
    purgeFilesData(dbConn, staleRel);
  }

  if (missingAbs.length === 0) return;

  // Classify drops so users see per-extension reasons instead of just a count
  // (#1011). `unsupported-by-native` is a legitimate parser limit (no Rust
  // extractor); `native-extractor-failure` indicates a real native bug since
  // the language IS supported by the addon yet the file was dropped anyway.
  const { byReason, totals } = classifyNativeDrops(missingRel);
  if (totals['unsupported-by-native'] > 0) {
    info(
      `Native orchestrator skipped ${totals['unsupported-by-native']} file(s) in languages without a Rust extractor; backfilling via WASM: ${formatDropExtensionSummary(byReason['unsupported-by-native'])}`,
    );
  }
  if (totals['native-extractor-failure'] > 0) {
    warn(
      `Native orchestrator dropped ${totals['native-extractor-failure']} file(s) in natively-supported languages — likely a Rust extractor bug. Backfilling via WASM: ${formatDropExtensionSummary(byReason['native-extractor-failure'])}`,
    );
  }
  const wasmResults = await parseFilesWasmForBackfill(missingAbs, ctx.rootDir);

  const rows: unknown[][] = [];
  const exportKeys: unknown[][] = [];
  for (const [relPath, symbols] of wasmResults) {
    // File row — mirrors insertDefinitionsAndExports: qualified_name is null.
    rows.push([relPath, 'file', relPath, 0, null, null, null, null, null]);
    for (const def of symbols.definitions ?? []) {
      // Populate qualified_name/scope the same way the JS fallback does so
      // downstream queries (cross-file references, "go to definition") find
      // these symbols.
      const dotIdx = def.name.lastIndexOf('.');
      const scope = dotIdx !== -1 ? def.name.slice(0, dotIdx) : null;
      rows.push([
        def.name,
        def.kind,
        relPath,
        def.line,
        def.endLine ?? null,
        null,
        def.name,
        scope,
        def.visibility ?? null,
      ]);
    }
    // Exports: insert the row (INSERT OR IGNORE — a matching definition row
    // is a no-op) and queue a key for the second-pass exported=1 update, so
    // queries filtering on exported=1 find backfilled symbols (#970).
    for (const exp of symbols.exports ?? []) {
      rows.push([exp.name, exp.kind, relPath, exp.line, null, null, exp.name, null, null]);
      exportKeys.push([exp.name, exp.kind, relPath, exp.line]);
    }
  }
  const db = dbConn;
  batchInsertNodes(db, rows);

  // Mark exported symbols in batches — mirrors insertDefinitionsAndExports.
  if (exportKeys.length > 0) {
    const EXPORT_CHUNK = 500;
    const exportStmtCache = new Map<number, SqliteStatement>();
    for (let i = 0; i < exportKeys.length; i += EXPORT_CHUNK) {
      const end = Math.min(i + EXPORT_CHUNK, exportKeys.length);
      const chunkSize = end - i;
      let updateStmt = exportStmtCache.get(chunkSize);
      if (!updateStmt) {
        const conditions = Array.from(
          { length: chunkSize },
          () => '(name = ? AND kind = ? AND file = ? AND line = ?)',
        ).join(' OR ');
        updateStmt = db.prepare(`UPDATE nodes SET exported = 1 WHERE ${conditions}`);
        exportStmtCache.set(chunkSize, updateStmt);
      }
      const vals: unknown[] = [];
      for (let j = i; j < end; j++) {
        const k = exportKeys[j] as unknown[];
        vals.push(k[0], k[1], k[2], k[3]);
      }
      updateStmt.run(...vals);
    }
  }

  // Persist file_hashes rows for every backfilled file. The Rust orchestrator
  // only hashes files it parsed itself, so without this step files in
  // optional-language extensions (e.g. .clj when no Rust extractor exists)
  // would be missing from `file_hashes` — permanently breaking the JS-side
  // fast-skip pre-flight (#1054), which rejects on `collected file missing
  // from file_hashes` and forces every no-op rebuild back through the full
  // ~2s native pipeline (#1068).
  //
  // Iterates `missingRel` (every collected file the Rust orchestrator
  // dropped), not `wasmResults`, so files that produced zero symbols still
  // get a row.
  try {
    const upsertHash = db.prepare(
      'INSERT OR REPLACE INTO file_hashes (file, hash, mtime, size) VALUES (?, ?, ?, ?)',
    );
    const writeHashes = db.transaction(() => {
      for (let i = 0; i < missingRel.length; i++) {
        const relPath = missingRel[i];
        const absPath = missingAbs[i];
        if (!relPath || !absPath) continue;
        let code: string | null;
        try {
          code = readFileSafe(absPath);
        } catch (e) {
          debug(`backfillNativeDroppedFiles: read failed for ${relPath}: ${toErrorMessage(e)}`);
          continue;
        }
        if (code === null) continue;
        const stat = fileStat(absPath);
        const mtime = stat ? stat.mtime : 0;
        const size = stat ? stat.size : 0;
        upsertHash.run(relPath, fileHash(code), mtime, size);
      }
    });
    writeHashes();
  } catch (e) {
    debug(
      `backfillNativeDroppedFiles: file_hashes write failed (table may not exist): ${toErrorMessage(e)}`,
    );
  }

  // Free WASM parse trees from the inline backfill path (#1058).
  // `parseFilesWasmInline` sets `symbols._tree` (a live web-tree-sitter Tree
  // backed by WASM linear memory) on every result, but these symbols are
  // consumed locally for DB row construction and never added to
  // `ctx.allSymbols`, so the finalize-stage `releaseWasmTrees` sweep never
  // sees them. Without this, trees leak WASM memory until process exit —
  // bounded per run but cumulative across in-process integration tests.
  // Mirrors the cleanup discipline established for #931.
  for (const [, symbols] of wasmResults) {
    const tree = (symbols as { _tree?: { delete?: () => void } })._tree;
    if (tree && typeof tree.delete === 'function') {
      try {
        tree.delete();
      } catch {
        /* ignore cleanup errors */
      }
    }
    (symbols as { _tree?: unknown; _langId?: unknown })._tree = undefined;
    (symbols as { _tree?: unknown; _langId?: unknown })._langId = undefined;
  }
}

// ── Pipeline stages execution ───────────────────────────────────────────

async function runPipelineStages(ctx: PipelineContext): Promise<void> {
  // ── WASM / fallback dual-connection mode ─────────────────────────────
  // NativeDatabase is deferred — not opened during setup. collectFiles and
  // detectChanges only need better-sqlite3. If no files changed, we exit
  // early without ever opening the native connection, saving ~5ms.
  // If nativeDb was opened by tryNativeOrchestrator (which fell through),
  // suspend it now to avoid dual-connection WAL corruption during stages.
  if (ctx.db && ctx.nativeDb) {
    suspendNativeDb(ctx, 'pre-collect');
    // When nativeFirstProxy is true, ctx.db is a NativeDbProxy wrapping the
    // now-closed NativeDatabase. Replace it with a real better-sqlite3
    // connection so the JS pipeline stages can operate normally.
    if (ctx.nativeFirstProxy) {
      ctx.db = openDb(ctx.dbPath);
      ctx.nativeFirstProxy = false;
    }
  }

  await collectFiles(ctx);
  await detectChanges(ctx);

  if (ctx.earlyExit) return;

  await parseFiles(ctx);

  // For small incremental builds (≤smallFilesThreshold files), skip the nativeDb open/close
  // cycle for insertNodes — the WAL checkpoint + connection churn (~5-10ms)
  // exceeds the napi bulk-insert savings on a handful of files. The JS
  // fallback path inside insertNodes handles this case efficiently.
  const smallIncremental =
    !ctx.isFullBuild && ctx.allSymbols.size <= ctx.config.build.smallFilesThreshold;
  if (ctx.nativeAvailable && ctx.engineName === 'native' && !smallIncremental) {
    reopenNativeDb(ctx, 'insertNodes');
  }

  await insertNodes(ctx);

  // Close nativeDb after insertNodes — remaining pipeline stages use JS paths.
  if (ctx.nativeDb && ctx.db) {
    closeNativeDb(ctx, 'post-insertNodes');
    refreshJsDb(ctx);
  }

  await resolveImports(ctx);
  await buildEdges(ctx);
  await buildStructure(ctx);

  // Reopen nativeDb for feature modules (ast, cfg, complexity, dataflow).
  // Skip for small incremental builds — same rationale as insertNodes above.
  //
  // Perf: do ONE upfront FULL checkpoint to flush JS WAL pages so Rust
  // can see the latest rows, then make suspendJsDb/resumeJsDb no-ops.
  // Previously each feature called wal_checkpoint(TRUNCATE) individually
  // (~68ms each × 3-4 features = ~200-270ms overhead on incremental builds).
  if (ctx.nativeAvailable && !smallIncremental) {
    reopenNativeDb(ctx, 'analyses');
    if (ctx.nativeDb && ctx.engineOpts) {
      ctx.db.pragma('wal_checkpoint(FULL)');
      ctx.engineOpts.nativeDb = ctx.nativeDb;
      ctx.engineOpts.suspendJsDb = () => {};
      ctx.engineOpts.resumeJsDb = () => {};
    }
    if (!ctx.nativeDb && ctx.engineOpts) {
      ctx.engineOpts.nativeDb = undefined;
      ctx.engineOpts.suspendJsDb = undefined;
      ctx.engineOpts.resumeJsDb = undefined;
    }
  }

  await runAnalyses(ctx);

  // Release WASM trees deterministically on the success path — same cleanup
  // as the error-path catch block.  Without this, trees stay allocated until
  // GC collects ctx, holding WASM memory for the rest of the build (#931).
  if (ctx.allSymbols?.size > 0) {
    for (const [, symbols] of ctx.allSymbols) {
      const tree = symbols._tree as { delete?: () => void } | undefined;
      if (tree && typeof tree.delete === 'function') {
        try {
          tree.delete();
        } catch {
          /* ignore cleanup errors */
        }
      }
      symbols._tree = undefined;
    }
  }

  // Flush Rust WAL writes (AST, complexity, CFG, dataflow) so the JS
  // connection and any post-build readers can see them.  One TRUNCATE
  // here replaces the N per-feature resumeJsDb checkpoints (#checkpoint-opt).
  if (ctx.nativeDb) {
    try {
      ctx.nativeDb.exec('PRAGMA wal_checkpoint(TRUNCATE)');
    } catch (e) {
      debug(`post-analyses WAL checkpoint failed: ${toErrorMessage(e)}`);
    }
    refreshJsDb(ctx);
  }

  await finalize(ctx);
}

// ── Main entry point ────────────────────────────────────────────────────

/**
 * Build the dependency graph for a codebase.
 *
 * Signature and return value are identical to the original monolithic buildGraph().
 */
export async function buildGraph(
  rootDir: string,
  opts: BuildGraphOpts = {},
): Promise<BuildResult | undefined> {
  const ctx = new PipelineContext();
  ctx.buildStart = performance.now();
  ctx.opts = opts;
  ctx.rootDir = rootDir;

  try {
    setupPipeline(ctx);

    // ── JS-side fast-skip for native incremental (#1054) ──────────────
    // The Rust orchestrator's internal early-exit fires reliably locally
    // but not in CI, where every no-op rebuild was paying the full ~2s
    // pipeline cost. A read-only mtime+size check here matches WASM's
    // ~20ms early-exit and skips the orchestrator entirely when no
    // source files have changed. Tier-2 hashing is left to the native
    // side: any mismatch falls through and lets Rust's detect_changes
    // remain the source of truth.
    //
    // Diagnostic logging gated by CODEGRAPH_FAST_SKIP_DIAG (#1066) — when
    // any of the call-site guards short-circuit (forceFullRebuild,
    // engineName, scope, etc.) we log the reason so the bench gate run
    // produces observable output even if `detectNoChanges` is never
    // entered.
    const fastSkipDiag = process.env.CODEGRAPH_FAST_SKIP_DIAG === '1';
    if (fastSkipDiag) {
      const reasons: string[] = [];
      if (!ctx.nativeAvailable) reasons.push('nativeAvailable=false');
      if (ctx.engineName !== 'native') reasons.push(`engineName=${ctx.engineName}`);
      if (!ctx.incremental) reasons.push('incremental=false');
      if (ctx.forceFullRebuild) reasons.push('forceFullRebuild=true');
      if ((ctx.opts as Record<string, unknown>).scope) reasons.push('scope=set');
      if (reasons.length > 0) {
        info(`[fast-skip] false: pre-flight gate skipped — ${reasons.join(', ')}`);
      }
    }
    if (
      ctx.nativeAvailable &&
      ctx.engineName === 'native' &&
      ctx.incremental &&
      !ctx.forceFullRebuild &&
      !(ctx.opts as Record<string, unknown>).scope
    ) {
      try {
        await collectFiles(ctx);
        if (
          detectNoChanges(ctx.db, ctx.allFiles, ctx.rootDir, ctx.opts as Record<string, unknown>)
        ) {
          info('No changes detected. Graph is up to date.');
          writeJournalHeader(ctx.rootDir, Date.now());
          closeDb(ctx.db);
          return;
        }
      } catch (err) {
        // Pre-flight is best-effort — any failure falls through to the
        // orchestrator, which performs its own complete detection.
        // Reset ctx.allFiles so runPipelineStages re-collects under its own
        // engine state if we ended up partially populated before throwing.
        ctx.allFiles = undefined as unknown as string[];
        ctx.discoveredDirs = undefined as unknown as Set<string>;
        debug(`native fast-skip pre-flight failed: ${toErrorMessage(err)}`);
      }
    }

    // ── Rust orchestrator fast path (#695) ────────────────────────────
    // When available, run the entire build pipeline in Rust with zero
    // napi crossings (eliminates WAL dual-connection dance). Falls back
    // to the JS pipeline on failure or when native is unavailable.
    try {
      const nativeResult = await tryNativeOrchestrator(ctx);
      if (nativeResult === 'early-exit') return;
      if (nativeResult) return nativeResult;
    } catch (err) {
      warn(`Native build orchestrator failed, falling back to JS pipeline: ${toErrorMessage(err)}`);
      // The version gate in checkEngineSchemaMismatch was skipped because
      // nativeAvailable was true. Now that we're falling back to the JS
      // pipeline, perform the codegraph_version check here so a version
      // bump still promotes to a full rebuild (#928).
      if (ctx.incremental && !ctx.forceFullRebuild) {
        const prevVersion = getBuildMeta(ctx.db, 'codegraph_version');
        if (prevVersion && prevVersion !== CODEGRAPH_VERSION) {
          info(
            `Codegraph version changed (${prevVersion} → ${CODEGRAPH_VERSION}), promoting to full rebuild.`,
          );
          ctx.forceFullRebuild = true;
          // Re-check embeddings: the initial warnOnEmbeddingsWipe ran before
          // forceFullRebuild was set here, so the silent-data-loss guard
          // would otherwise miss this late-promotion path (#986 follow-up).
          warnOnEmbeddingsWipe(ctx);
        }
      }
    }

    await runPipelineStages(ctx);
  } catch (err) {
    if (!ctx.earlyExit) {
      // Release WASM trees before closing DB to prevent V8 crash during
      // GC cleanup of orphaned WASM objects (#931).
      if (ctx.allSymbols?.size > 0) {
        for (const [, symbols] of ctx.allSymbols) {
          const tree = symbols._tree as { delete?: () => void } | undefined;
          if (tree && typeof tree.delete === 'function') {
            try {
              tree.delete();
            } catch {
              /* ignore cleanup errors */
            }
          }
          symbols._tree = undefined;
        }
      }
      if (ctx.db) {
        closeDbPair({ db: ctx.db, nativeDb: ctx.nativeDb });
      }
    }
    throw err;
  }

  if (ctx.earlyExit) return;

  return formatTimingResult(ctx);
}
