// Package store_sqlite is the on-disk, SQLite-backed implementation of
// graph.Store. It uses the pure-Go modernc.org/sqlite driver so the
// binary stays CGO-free on this code path, and satisfies the same
// conformance suite as the in-memory store (see
// internal/graph/storetest).
//
// Hot queries are precompiled as prepared statements in Open and
// closed in Close. Writes serialize through a single Go-side mutex
// because SQLite already serialises writers internally and an explicit
// mutex sidesteps SQLITE_BUSY contention when the conformance suite
// fans out 8 concurrent writers; reads still run concurrently under
// WAL mode.
//
// Meta maps are encoded as JSON (see meta_json.go); an empty / nil Meta
// is stored as NULL so the common case adds no row weight beyond the
// column header.
//
// EdgeIdentityRevisions is tracked in memory (atomic counter) -- it
// mirrors the in-memory store's monotonic "provenance churn" signal
// and does not need to survive process restarts (the in-memory store
// resets it on every New(), so the contract is per-process).
package store_sqlite

import (
	"database/sql"
	"errors"
	"fmt"
	"iter"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/zzet/gortex/internal/graph"

	_ "modernc.org/sqlite"
)

// Store is the SQLite-backed graph.Store implementation.
type Store struct {
	db *sql.DB

	// dbPath is the on-disk SQLite file path, retained for size
	// telemetry — the WAL high-water mark surfaces in daemon_health so a
	// runaway -wal is observable rather than silently filling the disk.
	dbPath string

	// writeMu serialises every mutation. SQLite serialises writers
	// internally; doing the same on the Go side turns SQLITE_BUSY
	// contention into clean lock-wait and keeps the conformance
	// concurrency test predictable.
	writeMu sync.Mutex

	// resolveMu is the resolver-coordination mutex returned by
	// ResolveMutex. Held by cross-repo / temporal / external resolver
	// passes to keep their edge mutations from interleaving. Separate
	// from writeMu so the resolver can hold it across multiple writes
	// without blocking unrelated steady-state mutations.
	resolveMu sync.Mutex

	edgeIdentityRevs atomic.Int64

	// wiped records that Open dropped an incompatible on-disk DB and
	// recreated it empty (a schema-version mismatch that an in-place ALTER
	// could not satisfy). Surfaced via NeedsRebuild so the daemon forces a
	// full re-index on warm restart instead of an incremental reconcile,
	// rather than relying on the side effect that a total wipe also empties
	// file_mtimes.
	wiped bool

	// WAL-checkpoint loop lifecycle. In WAL mode a COMMIT only appends to
	// the -wal file; pages move into the main DB (and the WAL becomes
	// reusable) at a checkpoint. SQLite's default passive auto-checkpoint
	// reuses the WAL in place and never shrinks the file, so under steady
	// writes with ever-present readers (the pooled connections here, plus
	// any other process holding the store open) the -wal ratchets up to a
	// large high-water mark and stays there. runCheckpointLoop periodically
	// runs `PRAGMA wal_checkpoint(TRUNCATE)` to drain the log into the DB
	// and shrink the file back down. nil for in-memory stores (no WAL).
	stopCheckpoint chan struct{} // closed by Close to stop the loop
	checkpointDone chan struct{} // closed by the loop when it returns
	stopOnce       sync.Once     // makes stopCheckpointLoop idempotent

	// bundles is the content-addressed package-scoped cache over
	// SearchSymbolBundles: a query serves cached Node + in/out edges for
	// packages whose content fingerprint is unchanged and skips the node
	// + edge fan-out for them. nil until SetBundleFingerprints is first
	// called (the daemon wires it from the analysis pass); a nil cache
	// makes SearchSymbolBundles fall through to the uncached path.
	bundles *bundleCache

	// memEst memoises AllRepoMemoryEstimates for a short TTL. That query
	// is two COUNT(*) … GROUP BY repo scans; on a large graph under
	// enrichment write load the pure-Go modernc sqlite count is
	// pathologically slow, and the daemon status path can call it
	// repeatedly. The TTL (and the mutex held across the recompute)
	// collapses a burst of status polls onto a single scan; a few
	// seconds of staleness is irrelevant for an advisory estimate.
	memEstMu  sync.Mutex
	memEstVal map[string]graph.RepoMemoryEstimate
	memEstAt  time.Time

	// Bulk-load fast path (graph.BulkLoader). Non-nil only between
	// BeginBulkLoad and FlushBulk, and only on a first/empty cold index.
	// database/sql PRAGMAs are connection-local, so the fast path pins one
	// connection (bulkConn) carrying synchronous=OFF + an enlarged page
	// cache and routes every bulk write through it; bulkPrevSync /
	// bulkPrevCacheSize hold the values FlushBulk restores before the
	// connection returns to the pool. All three are guarded by writeMu.
	bulkConn          *sql.Conn
	bulkPrevSync      int64
	bulkPrevCacheSize int64

	// Prepared statements (compiled once in Open, closed in Close).
	stmtInsertNode         *sql.Stmt
	stmtGetNode            *sql.Stmt
	stmtGetNodeByQual      *sql.Stmt
	stmtFindByName         *sql.Stmt
	stmtFindByNameInRepo   *sql.Stmt
	stmtFileNodes          *sql.Stmt
	stmtRepoNodes          *sql.Stmt
	stmtAllNodes           *sql.Stmt
	stmtNodeCount          *sql.Stmt
	stmtRepoPrefixes       *sql.Stmt
	stmtRepoStatsNodes     *sql.Stmt
	stmtRepoStatsEdges     *sql.Stmt
	stmtRepoNodeCount      *sql.Stmt
	stmtRepoEdgeCount      *sql.Stmt
	stmtAllRepoCountsNodes *sql.Stmt
	stmtAllRepoCountsEdges *sql.Stmt
	stmtStatsByKind        *sql.Stmt
	stmtStatsByLanguage    *sql.Stmt

	stmtInsertEdge       *sql.Stmt
	stmtOutEdges         *sql.Stmt
	stmtOutEdgesLight    *sql.Stmt
	stmtInEdges          *sql.Stmt
	stmtRepoEdges        *sql.Stmt
	stmtAllEdges         *sql.Stmt
	stmtEdgeCount        *sql.Stmt
	stmtRemoveEdge       *sql.Stmt
	stmtUpdateEdgeOrigin *sql.Stmt
	stmtUpdateEdgeAttrs  *sql.Stmt
	stmtSelectEdgeOrigin *sql.Stmt
	stmtDeleteEdgeByKey  *sql.Stmt
	stmtEdgeExists       *sql.Stmt

	stmtSelectFileNodeIDs *sql.Stmt
	stmtSelectRepoNodeIDs *sql.Stmt
	stmtDeleteNodeByFile  *sql.Stmt
	stmtDeleteNodeByRepo  *sql.Stmt
}

// Compile-time assertion: *Store satisfies graph.Store.
var _ graph.Store = (*Store)(nil)

// ResolveMutex returns the resolver-coordination mutex. Held by
// cross-repo / temporal / external resolver passes to serialise edge
// mutations. Separate from writeMu (which protects per-statement
// write serialisation against SQLITE_BUSY) so the resolver can hold
// it across multi-write batches without blocking unrelated steady-
// state mutations on the same store.
func (s *Store) ResolveMutex() *sync.Mutex { return &s.resolveMu }

// NeedsRebuild reports that Open dropped an incompatible on-disk database and
// recreated it empty, so the daemon's warm-restart path should force a full
// re-index (bypassing an incremental reconcile that would carry stale state)
// — see cmd/gortex.storeNeedsRebuild, the capability probe this satisfies.
func (s *Store) NeedsRebuild() bool { return s.wiped }

// Open opens (or creates) the SQLite database at path, runs the schema
// migration, and prepares hot statements. The DB is opened with WAL
// journaling and synchronous=NORMAL -- the same durability/throughput
// tradeoff every embedded-SQLite app uses for write-heavy workloads.
//
// Pass ":memory:" for an ephemeral in-process database (handy for
// tests when you don't need on-disk persistence).
//
// By default Open will NOT destroy an incompatible on-disk database: if the
// stored schema version requires a rebuild (a newer build's DB, or an older
// one crossing a rebuild migration) it returns ErrSchemaRebuildRequired and
// leaves the file untouched. Pass WithRebuild to permit the drop-and-recreate
// — only a caller that holds exclusive access to the store may do so (see
// WithRebuild).
func Open(path string, opts ...Option) (*Store, error) {
	var o openOptions
	for _, opt := range opts {
		opt(&o)
	}
	return openWith(path, currentSchemaVersion, schemaMigrations, o.allowRebuild)
}

// Option configures Open.
type Option func(*openOptions)

type openOptions struct {
	allowRebuild bool
}

// WithRebuild permits Open to drop and recreate an on-disk database whose
// schema version is incompatible (a newer build's, or an older one crossing a
// migration that an in-place ALTER cannot satisfy).
//
// The caller MUST hold exclusive cross-process access to the store file —
// removing a SQLite file another process has open silently splits its state.
// The daemon satisfies this: it takes an exclusive flock on <store>.lock for
// the writable on-disk sqlite lifecycle and passes this option only in that
// branch (see serverstack.NewSharedServer / OpenBackend). Without it, a wipe
// plan yields ErrSchemaRebuildRequired and the file is left intact, so a
// caller that does not hold the lock cannot corrupt a live store.
func WithRebuild() Option { return func(o *openOptions) { o.allowRebuild = true } }

// ErrSchemaRebuildRequired is returned by Open when an on-disk database needs a
// destructive rebuild but the caller did not pass WithRebuild (i.e. cannot
// prove it holds the store lock).
var ErrSchemaRebuildRequired = errors.New("store_sqlite: on-disk schema is incompatible and must be rebuilt; reopen with WithRebuild while holding the store lock")

// openWith is Open parameterised by the target schema version, migration
// registry, and rebuild permission so tests can drive the baseline / in-place
// / rebuild arms without mutating package globals. Open passes the package
// defaults (currentSchemaVersion, schemaMigrations) and the WithRebuild flag.
func openWith(path string, current int, migrations []schemaMigration, allowRebuild bool) (*Store, error) {
	// Pragmas: WAL + synchronous=NORMAL is the standard write-heavy
	// embedded tradeoff. cache_size(-32768) gives each pooled connection a
	// 32 MiB page cache; temp_store(MEMORY) keeps GROUP BY / ORDER BY scratch
	// off disk; mmap_size(256 MiB) lets reads fault pages straight from the
	// OS page cache instead of copying through SQLite's. These materially
	// speed the resolver/query phases on a large graph.
	//
	// journal_size_limit(64 MiB) caps the -wal high-water mark: after any
	// checkpoint SQLite truncates the WAL back down to this size instead of
	// leaving it at whatever it grew to. Without it the WAL only ratchets
	// up (a passive checkpoint reuses the file in place, never shrinking
	// it), which is how a 535 MB DB ends up with an 11 GB -wal. This bounds
	// the file even between the explicit TRUNCATE checkpoints runCheckpointLoop
	// issues, and even if that loop is not running.
	dsn := path + "?_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)&_pragma=busy_timeout(5000)&_pragma=foreign_keys(OFF)&_pragma=cache_size(-32768)&_pragma=temp_store(MEMORY)&_pragma=mmap_size(268435456)&_pragma=journal_size_limit(67108864)"
	db, err := sql.Open("sqlite", dsn)
	if err != nil {
		return nil, fmt.Errorf("sqlite open: %w", err)
	}
	// Pool up to NumCPU connections so the resolver's parallel
	// worker fan-out (NumCPU goroutines doing FindNodesByName /
	// GetNode / GetOutEdges concurrently) doesn't serialise through
	// a single connection — the dominant gap between the SQLite and
	// bbolt backends on the bench's resolver stage was exactly that.
	// SQLite's WAL mode allows concurrent readers across multiple
	// connections; writes still serialise via writeMu on the Go
	// side, then via SQLite's internal write lock. Every connection
	// the pool opens picks up the journal-mode / synchronous /
	// busy-timeout pragmas from the DSN above, so we don't need to
	// pin one connection to "remember" them.
	db.SetMaxOpenConns(runtime.NumCPU())

	// Reconcile the on-disk schema version before applying schemaSQL. The graph
	// store is a rebuildable cache, so an incompatible (older needing a rebuild
	// step, or newer) DB is dropped and reindexed rather than migrated in place
	// (see schema_version.go). The daemon holds an exclusive store.lock around
	// Open, so wiping the file here cannot race another process.
	stored, err := readUserVersion(db)
	if err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("sqlite read schema version: %w", err)
	}
	plan := planSchemaMigrationWith(stored, current, migrations)
	didWipe := false
	if plan.wipe && !isMemoryPath(path) {
		// Refuse the destructive rebuild unless the caller proved it holds
		// exclusive access (WithRebuild). This keeps the file safe even if a
		// future caller reaches a wipe plan without the daemon's store lock.
		if !allowRebuild {
			_ = db.Close()
			return nil, ErrSchemaRebuildRequired
		}
		if err := db.Close(); err != nil {
			return nil, fmt.Errorf("sqlite close for rebuild: %w", err)
		}
		if err := removeStoreFiles(path); err != nil {
			return nil, fmt.Errorf("sqlite rebuild: %w", err)
		}
		db, err = sql.Open("sqlite", dsn)
		if err != nil {
			return nil, fmt.Errorf("sqlite reopen for rebuild: %w", err)
		}
		db.SetMaxOpenConns(runtime.NumCPU())
		didWipe = true
	}

	if _, err := db.Exec(schemaSQL); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("sqlite schema: %w", err)
	}
	// Add the promoted node columns to databases created before they
	// existed (CREATE TABLE IF NOT EXISTS won't alter an existing table).
	// Must run before the droppable-index loop below — nodes_semantic_pending
	// references a promoted column — and before prepare(), whose node INSERT
	// references them too.
	if err := ensureNodeColumns(db); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("sqlite node columns: %w", err)
	}
	// nodes.is_stub generated column — see ensureNodeGeneratedColumns for why
	// this is a separate function from ensureNodeColumns above.
	if err := ensureNodeGeneratedColumns(db); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("sqlite node generated columns: %w", err)
	}
	// Same treatment for the edges table's is_unresolved generated column —
	// must run before the droppable-index loop below, which creates an index
	// over it.
	if err := ensureEdgeColumns(db); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("sqlite edge columns: %w", err)
	}
	// Create the droppable secondary indexes from the shared set so their
	// initial-creation DDL is byte-identical to the DDL the bulk-load fast
	// path rebuilds them with (BeginBulkLoad drops these, FlushBulk
	// recreates them — see bulk_load.go). Kept out of schemaSQL so the two
	// sites cannot drift.
	for _, idx := range bulkDroppableIndexes {
		if _, err := db.Exec(idx.ddl); err != nil {
			_ = db.Close()
			return nil, fmt.Errorf("sqlite create index %s: %w", idx.name, err)
		}
	}
	// edges_external is a partial index over exactly the external-call
	// terminals, so ExternalCallCandidateEdges scans a tiny index instead
	// of the full edges table. Built from the shared predicate const (not
	// inlined in schemaSQL) so the index WHERE and the query WHERE stay
	// byte-identical — SQLite only uses a partial index when the query's
	// WHERE matches the index's.
	if _, err := db.Exec(`CREATE INDEX IF NOT EXISTS edges_external ON edges(kind) WHERE ` + externalCallTargetPredicate); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("sqlite edges_external index: %w", err)
	}
	// Backfill the FTS rowid sidecar for databases built before it existed,
	// so the first incremental UpsertSymbolFTS on an already-indexed symbol
	// can do its O(log n) docid delete instead of leaking a duplicate row.
	// One-time; a no-op once the map is populated or the FTS index is empty.
	if err := backfillSymbolFTSRowidMap(db); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("sqlite fts rowid backfill: %w", err)
	}

	// Apply any in-place migration steps, then stamp the current schema version.
	// Fresh and pre-versioning (stored==0) stores run the in-place steps too —
	// they are idempotent and no-op on an empty or already-clean store — so the
	// first in-place migration ships without forcing every non-daemon Open to
	// pass WithRebuild. A wipe plan carries no in-place steps, and after a wipe
	// the store is empty and the daemon's normal indexing repopulates it.
	if plan.stamp {
		if err := applyInPlaceMigrations(db, plan.inPlace); err != nil {
			_ = db.Close()
			return nil, fmt.Errorf("sqlite schema migrate: %w", err)
		}
		if err := setUserVersion(db, current); err != nil {
			_ = db.Close()
			return nil, fmt.Errorf("sqlite stamp schema version: %w", err)
		}
	}

	s := &Store{db: db, dbPath: path, wiped: didWipe}
	// Initialise the bundle cache at construction so its pointer is
	// never written after Open — concurrent SearchSymbolBundles reads
	// and SetBundleFingerprints writes then race only on the cache's
	// own mutex-guarded maps, not on the Store field. The cache stays
	// inert (every lookup a miss) until the daemon supplies fingerprints.
	s.bundles = newBundleCache()
	if err := s.prepare(); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("sqlite prepare: %w", err)
	}
	// In-memory databases have no WAL file to drain, so the periodic
	// checkpoint is pointless there (and would leak a goroutine per
	// short-lived test store). Only run it for on-disk stores.
	if !strings.Contains(path, ":memory:") {
		s.stopCheckpoint = make(chan struct{})
		s.checkpointDone = make(chan struct{})
		go s.runCheckpointLoop(walCheckpointInterval)
	}
	return s, nil
}

// walCheckpointInterval is how often runCheckpointLoop drains the WAL into
// the main DB and truncates the -wal file. Five minutes keeps the file
// bounded under steady writes without making the checkpoint itself a hot
// path; journal_size_limit in the DSN bounds growth between ticks.
const walCheckpointInterval = 5 * time.Minute

// runCheckpointLoop issues a TRUNCATE checkpoint every interval until Close
// stops it. Best-effort: a checkpoint that can't fully complete because a
// reader or writer holds the WAL just truncates what it can and retries on
// the next tick.
func (s *Store) runCheckpointLoop(interval time.Duration) {
	defer close(s.checkpointDone)
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-s.stopCheckpoint:
			return
		case <-ticker.C:
			_ = s.CheckpointWAL()
		}
	}
}

// CheckpointWAL runs `PRAGMA wal_checkpoint(TRUNCATE)`: it flushes the
// write-ahead log into the main database file and shrinks the -wal back to
// zero. A passive checkpoint (SQLite's default) only reuses the WAL in
// place and never reclaims the space; TRUNCATE is the mode that does.
// Exposed so a daemon shutdown path or an operator command can force a
// drain; the background loop calls it on a timer. Not held under writeMu —
// SQLite coordinates checkpoints against writers internally, and blocking
// steady-state writes on a maintenance op is the wrong tradeoff.
func (s *Store) CheckpointWAL() error {
	_, err := s.db.Exec("PRAGMA wal_checkpoint(TRUNCATE)")
	return err
}

// stopCheckpointLoop signals the background loop to exit and waits for it,
// so callers can be sure no checkpoint is in flight before closing s.db.
// Idempotent: safe to call from Close more than once.
func (s *Store) stopCheckpointLoop() {
	s.stopOnce.Do(func() {
		if s.stopCheckpoint != nil {
			close(s.stopCheckpoint)
			<-s.checkpointDone
		}
	})
}

// Close closes every prepared statement and the underlying *sql.DB. It
// first stops the WAL-checkpoint loop and issues one final TRUNCATE
// checkpoint so the -wal file is drained and shrunk on graceful shutdown
// rather than lingering at its high-water mark until the next open.
func (s *Store) Close() error {
	s.stopCheckpointLoop()
	if s.checkpointDone != nil { // on-disk store: drain the WAL one last time
		_ = s.CheckpointWAL()
	}
	stmts := []*sql.Stmt{
		s.stmtInsertNode, s.stmtGetNode, s.stmtGetNodeByQual,
		s.stmtFindByName, s.stmtFindByNameInRepo,
		s.stmtFileNodes, s.stmtRepoNodes,
		s.stmtAllNodes, s.stmtNodeCount, s.stmtRepoPrefixes,
		s.stmtRepoStatsNodes, s.stmtRepoStatsEdges,
		s.stmtRepoNodeCount, s.stmtRepoEdgeCount,
		s.stmtAllRepoCountsNodes, s.stmtAllRepoCountsEdges,
		s.stmtStatsByKind, s.stmtStatsByLanguage,
		s.stmtInsertEdge, s.stmtOutEdges, s.stmtOutEdgesLight, s.stmtInEdges,
		s.stmtRepoEdges,
		s.stmtAllEdges, s.stmtEdgeCount, s.stmtRemoveEdge,
		s.stmtUpdateEdgeOrigin, s.stmtUpdateEdgeAttrs, s.stmtSelectEdgeOrigin, s.stmtDeleteEdgeByKey,
		s.stmtEdgeExists,
		s.stmtSelectFileNodeIDs, s.stmtSelectRepoNodeIDs,
		s.stmtDeleteNodeByFile, s.stmtDeleteNodeByRepo,
	}
	for _, st := range stmts {
		if st != nil {
			_ = st.Close()
		}
	}
	return s.db.Close()
}

func (s *Store) prepare() error {
	var err error
	prep := func(out **sql.Stmt, q string) {
		if err != nil {
			return
		}
		var st *sql.Stmt
		st, err = s.db.Prepare(q)
		if err != nil {
			err = fmt.Errorf("prepare %q: %w", q, err)
			return
		}
		*out = st
	}

	const nodeCols = lookupNodeCols

	prep(&s.stmtInsertNode,
		`INSERT OR REPLACE INTO nodes (`+nodeCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`)
	prep(&s.stmtGetNode,
		`SELECT `+nodeCols+` FROM nodes WHERE id = ?`)
	prep(&s.stmtGetNodeByQual,
		`SELECT `+nodeCols+` FROM nodes WHERE qual_name = ? LIMIT 1`)
	prep(&s.stmtFindByName,
		`SELECT `+nodeCols+` FROM nodes WHERE name = ?`)
	prep(&s.stmtFindByNameInRepo,
		`SELECT `+nodeCols+` FROM nodes WHERE name = ? AND repo_prefix = ?`)
	prep(&s.stmtFileNodes,
		`SELECT `+nodeCols+` FROM nodes WHERE file_path = ?`)
	prep(&s.stmtRepoNodes,
		`SELECT `+nodeCols+` FROM nodes WHERE repo_prefix = ?`)
	prep(&s.stmtAllNodes,
		`SELECT `+nodeCols+` FROM nodes`)
	prep(&s.stmtNodeCount,
		`SELECT COUNT(*) FROM nodes`)
	prep(&s.stmtRepoPrefixes,
		`SELECT DISTINCT repo_prefix FROM nodes WHERE repo_prefix <> ''`)

	prep(&s.stmtRepoStatsNodes,
		`SELECT repo_prefix, kind, language, COUNT(*) FROM nodes WHERE repo_prefix <> '' GROUP BY repo_prefix, kind, language`)
	prep(&s.stmtRepoStatsEdges,
		`SELECT n.repo_prefix, COUNT(*)
		 FROM edges e
		 JOIN nodes n ON n.id = e.from_id
		 WHERE n.repo_prefix <> ''
		 GROUP BY n.repo_prefix`)
	prep(&s.stmtRepoNodeCount,
		`SELECT COUNT(*) FROM nodes WHERE repo_prefix = ?`)
	prep(&s.stmtRepoEdgeCount,
		`SELECT COUNT(*)
		 FROM edges e
		 JOIN nodes n ON n.id = e.from_id
		 WHERE n.repo_prefix = ?`)
	prep(&s.stmtAllRepoCountsNodes,
		`SELECT repo_prefix, COUNT(*) FROM nodes WHERE repo_prefix <> '' GROUP BY repo_prefix`)
	prep(&s.stmtAllRepoCountsEdges,
		`SELECT n.repo_prefix, COUNT(*)
		 FROM edges e
		 JOIN nodes n ON n.id = e.from_id
		 WHERE n.repo_prefix <> ''
		 GROUP BY n.repo_prefix`)

	prep(&s.stmtStatsByKind,
		`SELECT kind, COUNT(*) FROM nodes GROUP BY kind`)
	prep(&s.stmtStatsByLanguage,
		`SELECT language, COUNT(*) FROM nodes GROUP BY language`)

	const edgeCols = `from_id, to_id, kind, file_path, line, confidence, confidence_label, origin, tier, cross_repo, meta, resolve_terminal, resolve_terminal_reason`

	prep(&s.stmtInsertEdge,
		`INSERT OR IGNORE INTO edges (`+edgeCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)`)
	prep(&s.stmtOutEdges,
		`SELECT `+edgeCols+` FROM edges WHERE from_id = ?`)
	// edgeColsLight is the package-level meta-less projection (store_light_edges.go),
	// shared with AllEdgesLight so this prepared statement and the whole-graph scan
	// can never drift apart.
	prep(&s.stmtOutEdgesLight,
		`SELECT `+edgeColsLight+` FROM edges WHERE from_id = ?`)
	prep(&s.stmtInEdges,
		`SELECT `+edgeCols+` FROM edges WHERE to_id = ?`)
	prep(&s.stmtRepoEdges,
		`SELECT e.from_id, e.to_id, e.kind, e.file_path, e.line,
		        e.confidence, e.confidence_label, e.origin, e.tier,
		        e.cross_repo, e.meta, e.resolve_terminal, e.resolve_terminal_reason
		   FROM edges e
		   JOIN nodes n ON n.id = e.from_id
		  WHERE n.repo_prefix = ?`)
	prep(&s.stmtAllEdges,
		`SELECT `+edgeCols+` FROM edges`)
	prep(&s.stmtEdgeCount,
		`SELECT COUNT(*) FROM edges`)
	prep(&s.stmtRemoveEdge,
		`DELETE FROM edges WHERE from_id = ? AND to_id = ? AND kind = ?`)

	prep(&s.stmtSelectEdgeOrigin,
		`SELECT origin FROM edges WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ?`)
	prep(&s.stmtUpdateEdgeOrigin,
		`UPDATE edges SET origin = ?, tier = ? WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ?`)
	prep(&s.stmtUpdateEdgeAttrs,
		`UPDATE edges SET confidence = ?, confidence_label = ?, origin = ?, tier = ?, meta = ?, resolve_terminal = ?, resolve_terminal_reason = ? WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ?`)
	prep(&s.stmtDeleteEdgeByKey,
		`DELETE FROM edges WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ?`)
	prep(&s.stmtEdgeExists,
		`SELECT 1 FROM edges WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ? LIMIT 1`)

	prep(&s.stmtSelectFileNodeIDs,
		`SELECT id FROM nodes WHERE file_path = ?`)
	prep(&s.stmtSelectRepoNodeIDs,
		`SELECT id FROM nodes WHERE repo_prefix = ?`)
	prep(&s.stmtDeleteNodeByFile,
		`DELETE FROM nodes WHERE file_path = ?`)
	prep(&s.stmtDeleteNodeByRepo,
		`DELETE FROM nodes WHERE repo_prefix = ?`)

	return err
}

// encodeMeta / decodeMeta live in meta_json.go (JSON codec + the
// metaWire typed DTO + the legacy-gob dual-read fallback).

// -- row scanners ---------------------------------------------------------

func scanNode(scanner interface {
	Scan(...any) error
}) (*graph.Node, error) {
	var (
		n        graph.Node
		metaBlob []byte
		p        promotedNodeMeta
	)
	err := scanner.Scan(
		&n.ID, &n.Kind, &n.Name, &n.QualName, &n.FilePath,
		&n.StartLine, &n.EndLine, &n.StartColumn, &n.EndColumn, &n.Language,
		&n.RepoPrefix, &n.WorkspaceID, &n.ProjectID,
		&p.sig, &p.vis, &p.doc, &p.external, &p.returnType,
		&p.isAsync, &p.isStatic, &p.isAbstract, &p.isExported, &p.updatedAt,
		&p.dataClass, &p.semanticType, &p.semanticSource, &metaBlob,
	)
	if err != nil {
		return nil, err
	}
	if len(metaBlob) > 0 {
		m, derr := decodeMeta(metaBlob)
		if derr != nil {
			return nil, derr
		}
		n.Meta = m
	}
	// Restore the promoted columns into Meta. They are authoritative for
	// rows written after the promotion; a NULL column (legacy gob rows)
	// is left alone so the blob-carried value survives.
	restorePromotedMeta(&n, p)
	return &n, nil
}

// scanNodeLight scans the same columns as scanNode minus the trailing meta
// blob — no decodeMeta call, so no JSON/gob parse per row. Promoted columns
// still restore into Meta via restorePromotedMeta, so any caller that only
// reads a promoted key (signature, visibility, ..., semantic_type) sees the
// exact values scanNode would produce; only non-promoted content still
// living in the row's blob is absent. See graph.LightNodeReader: a node
// from this scan must never be round-tripped back through AddNode/AddBatch.
func scanNodeLight(scanner interface {
	Scan(...any) error
}) (*graph.Node, error) {
	var (
		n graph.Node
		p promotedNodeMeta
	)
	err := scanner.Scan(
		&n.ID, &n.Kind, &n.Name, &n.QualName, &n.FilePath,
		&n.StartLine, &n.EndLine, &n.StartColumn, &n.EndColumn, &n.Language,
		&n.RepoPrefix, &n.WorkspaceID, &n.ProjectID,
		&p.sig, &p.vis, &p.doc, &p.external, &p.returnType,
		&p.isAsync, &p.isStatic, &p.isAbstract, &p.isExported, &p.updatedAt,
		&p.dataClass, &p.semanticType, &p.semanticSource,
	)
	if err != nil {
		return nil, err
	}
	restorePromotedMeta(&n, p)
	return &n, nil
}

func scanEdge(scanner interface {
	Scan(...any) error
}) (*graph.Edge, error) {
	var (
		e         graph.Edge
		metaBlob  []byte
		crossRepo int64
		p         promotedEdgeMeta
	)
	err := scanner.Scan(
		&e.From, &e.To, &e.Kind, &e.FilePath, &e.Line,
		&e.Confidence, &e.ConfidenceLabel, &e.Origin, &e.Tier,
		&crossRepo, &metaBlob, &p.resolveTerminal, &p.resolveTerminalReason,
	)
	if err != nil {
		return nil, err
	}
	e.CrossRepo = crossRepo != 0
	if len(metaBlob) > 0 {
		m, derr := decodeMeta(metaBlob)
		if derr != nil {
			return nil, derr
		}
		e.Meta = m
	}
	// Restore the promoted columns into Meta. They are authoritative for
	// rows written after the promotion; a NULL column (pre-promotion rows)
	// is left alone so any blob-carried value survives.
	restorePromotedEdgeMeta(&e, p)
	return &e, nil
}

// scanEdgeLight scans an edge WITHOUT decoding its meta blob -- for hot
// read paths (dataflow call-target lookup) that read only endpoints,
// kind, and line. Skipping the meta column avoids the JSON decode + map
// allocation that dominates large edge scans on this backend; the
// returned edge's Meta is nil.
func scanEdgeLight(scanner interface {
	Scan(...any) error
}) (*graph.Edge, error) {
	var (
		e         graph.Edge
		crossRepo int64
	)
	err := scanner.Scan(
		&e.From, &e.To, &e.Kind, &e.FilePath, &e.Line,
		&e.Confidence, &e.ConfidenceLabel, &e.Origin, &e.Tier,
		&crossRepo,
	)
	if err != nil {
		return nil, err
	}
	e.CrossRepo = crossRepo != 0
	return &e, nil
}

// -- writes ---------------------------------------------------------------

// AddNode inserts or replaces a node. Idempotent on the id column --
// re-adding the same id with new content does a last-write-wins
// update, matching the in-memory store's behaviour.
func (s *Store) AddNode(n *graph.Node) {
	if n == nil || n.ID == "" {
		return
	}
	// Cross-daemon proxy nodes are volatile remote-derived state and
	// must never reach disk. The durable writer is the single gate —
	// neither the resolver mint path nor the hydrator carries its own
	// "don't persist" branch. A dropped proxy node is re-minted on
	// demand after a restart.
	if graph.IsProxyNode(n) {
		return
	}
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	if err := s.insertNodeLocked(s.stmtInsertNode, n); err != nil {
		// graph.Store.AddNode has no error channel; the in-memory
		// store can't fail either. We swallow the error here for API
		// parity; surface as a panic only on a clearly catastrophic
		// failure (closed DB), not on a transient busy.
		panicOnFatal(err)
	}
}

func (s *Store) insertNodeLocked(stmt *sql.Stmt, n *graph.Node) error {
	p, blobMeta := extractPromotedMeta(n.Meta)
	metaBlob, err := encodeMeta(blobMeta)
	if err != nil {
		return err
	}
	_, err = stmt.Exec(
		n.ID, string(n.Kind), n.Name, n.QualName, n.FilePath,
		n.StartLine, n.EndLine, n.StartColumn, n.EndColumn, n.Language,
		n.RepoPrefix, n.WorkspaceID, n.ProjectID,
		p.sig, p.vis, p.doc, p.external, p.returnType,
		p.isAsync, p.isStatic, p.isAbstract, p.isExported, p.updatedAt,
		p.dataClass, p.semanticType, p.semanticSource, metaBlob,
	)
	return err
}

// AddEdge inserts an edge. Idempotent on the logical edge key (from,
// to, kind, file_path, line) -- a second AddEdge with the same key is
// a no-op (INSERT OR IGNORE), matching the in-memory store's "stored
// pointer replaced in place" semantics. Origin upgrades on a re-add
// are NOT applied through this path; use SetEdgeProvenance for that
// (matches the in-memory store: AddEdge replaces the *Edge pointer,
// but the conformance suite only verifies dedup-by-key, not pointer
// replacement, and the in-memory store also routes provenance
// upgrades through SetEdgeProvenance).
func (s *Store) AddEdge(e *graph.Edge) {
	if e == nil {
		return
	}
	// An edge to/from a cross-daemon proxy node is volatile and never
	// persisted (the proxy node itself is dropped at AddNode).
	if graph.IsProxyID(e.From) || graph.IsProxyID(e.To) {
		return
	}
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	if err := s.insertEdgeLocked(s.stmtInsertEdge, e); err != nil {
		panicOnFatal(err)
	}
}

func (s *Store) insertEdgeLocked(stmt *sql.Stmt, e *graph.Edge) error {
	p, blobMeta := extractPromotedEdgeMeta(e.Meta)
	metaBlob, err := encodeMeta(blobMeta)
	if err != nil {
		return err
	}
	var crossRepo int64
	if e.CrossRepo {
		crossRepo = 1
	}
	_, err = stmt.Exec(
		e.From, e.To, string(e.Kind), e.FilePath, e.Line,
		e.Confidence, e.ConfidenceLabel, e.Origin, e.Tier,
		crossRepo, metaBlob, p.resolveTerminal, p.resolveTerminalReason,
	)
	return err
}

// AddBatch inserts nodes and edges in a single transaction -- the
// 10-100x speedup vs per-statement commits at indexing scale.
func (s *Store) AddBatch(nodes []*graph.Node, edges []*graph.Edge) {
	if len(nodes) == 0 && len(edges) == 0 {
		return
	}
	s.writeMu.Lock()
	defer s.writeMu.Unlock()

	tx, err := s.beginWrite()
	if err != nil {
		panicOnFatal(err)
		return
	}
	commit := false
	defer func() {
		if !commit {
			_ = tx.Rollback()
		}
	}()

	insertNode := tx.Stmt(s.stmtInsertNode)
	defer insertNode.Close()
	insertEdge := tx.Stmt(s.stmtInsertEdge)
	defer insertEdge.Close()

	for _, n := range nodes {
		if n == nil || n.ID == "" {
			continue
		}
		// Cross-daemon proxy nodes never reach disk.
		if graph.IsProxyNode(n) {
			continue
		}
		if err := s.insertNodeLocked(insertNode, n); err != nil {
			panicOnFatal(err)
			return
		}
	}
	for _, e := range edges {
		if e == nil {
			continue
		}
		// An edge to or from a proxy node is volatile remote-derived
		// state too; never persist it (it would dangle on reload since
		// the proxy node itself is dropped).
		if graph.IsProxyID(e.From) || graph.IsProxyID(e.To) {
			continue
		}
		if err := s.insertEdgeLocked(insertEdge, e); err != nil {
			panicOnFatal(err)
			return
		}
	}

	if err := tx.Commit(); err != nil {
		panicOnFatal(err)
		return
	}
	commit = true
}

// SetEdgeProvenance mutates an existing edge's origin in-place and
// bumps the identity-revision counter when the origin actually
// changes. Returns true iff a change was applied. Mirrors the
// in-memory store's "delete-then-insert of identity" semantics.
func (s *Store) SetEdgeProvenance(e *graph.Edge, newOrigin string) bool {
	if e == nil {
		return false
	}
	s.writeMu.Lock()
	defer s.writeMu.Unlock()

	// Look up the stored origin -- the caller-supplied *Edge may be a
	// detached copy whose Origin already matches newOrigin even though
	// the row still has the old value.
	var storedOrigin string
	row := s.stmtSelectEdgeOrigin.QueryRow(e.From, e.To, string(e.Kind), e.FilePath, e.Line)
	if err := row.Scan(&storedOrigin); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false
		}
		panicOnFatal(err)
		return false
	}
	if storedOrigin == newOrigin {
		return false
	}
	newTier := e.Tier
	if newTier != "" {
		newTier = graph.ResolvedBy(newOrigin)
	}
	if _, err := s.stmtUpdateEdgeOrigin.Exec(newOrigin, newTier, e.From, e.To, string(e.Kind), e.FilePath, e.Line); err != nil {
		panicOnFatal(err)
		return false
	}
	// Reflect the change on the caller's struct, mirroring the
	// in-memory store which mutates the in-graph *Edge in place.
	e.Origin = newOrigin
	if e.Tier != "" {
		e.Tier = newTier
	}
	s.edgeIdentityRevs.Add(1)
	return true
}

// PersistEdgeAttributes durably rewrites the mutable attribute columns
// (confidence, confidence_label, origin, tier, meta) of the edge row
// identified by e's full logical key. It is the disk-backend counterpart
// to the in-memory store's "mutate the live *Edge in place" behaviour: a
// pass that confirms an edge's full provenance bundle (not just origin)
// calls this so the confidence / label / meta survive a reload. A missing
// row is a silent no-op (UPDATE ... WHERE matches nothing).
func (s *Store) PersistEdgeAttributes(e *graph.Edge) {
	if e == nil {
		return
	}
	p, blobMeta := extractPromotedEdgeMeta(e.Meta)
	metaBlob, err := encodeMeta(blobMeta)
	if err != nil {
		panicOnFatal(err)
		return
	}
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	if _, err := s.stmtUpdateEdgeAttrs.Exec(
		e.Confidence, e.ConfidenceLabel, e.Origin, e.Tier, metaBlob,
		p.resolveTerminal, p.resolveTerminalReason,
		e.From, e.To, string(e.Kind), e.FilePath, e.Line,
	); err != nil {
		panicOnFatal(err)
	}
}

// Compile-time assertion: *Store satisfies the batched meta persister.
var _ graph.EdgeMetaBatchPersister = (*Store)(nil)

// PersistEdgeAttributesBatch is the batched form of PersistEdgeAttributes:
// it rewrites the mutable attribute columns (confidence, confidence_label,
// origin, tier, meta) for every edge in the batch, chunking the writes into
// reindexChunkSize-row transactions re-using one prepared statement. The
// resolver's terminal-skip stamping calls it to persist a Meta flag across a
// large slice of edges without paying a BEGIN/COMMIT per edge. A row with no
// matching key is a silent no-op (UPDATE ... WHERE matches nothing).
func (s *Store) PersistEdgeAttributesBatch(edges []*graph.Edge) {
	if len(edges) == 0 {
		return
	}
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	for i := 0; i < len(edges); i += reindexChunkSize {
		end := minInt(i+reindexChunkSize, len(edges))
		chunk := edges[i:end]
		tx, err := s.db.Begin()
		if err != nil {
			panicOnFatal(err)
			return
		}
		updStmt := tx.Stmt(s.stmtUpdateEdgeAttrs)
		for _, e := range chunk {
			if e == nil {
				continue
			}
			p, blobMeta := extractPromotedEdgeMeta(e.Meta)
			metaBlob, err := encodeMeta(blobMeta)
			if err != nil {
				_ = tx.Rollback()
				panicOnFatal(err)
				return
			}
			if _, err := updStmt.Exec(
				e.Confidence, e.ConfidenceLabel, e.Origin, e.Tier, metaBlob,
				p.resolveTerminal, p.resolveTerminalReason,
				e.From, e.To, string(e.Kind), e.FilePath, e.Line,
			); err != nil {
				_ = tx.Rollback()
				panicOnFatal(err)
				return
			}
		}
		if err := tx.Commit(); err != nil {
			panicOnFatal(err)
			return
		}
	}
}

// ReindexEdge updates the stored row after e.To has been mutated from
// oldTo to e.To. Implemented as delete-old + insert-new under the
// same write lock (SQLite's UNIQUE constraint on (from,to,kind,file,
// line) makes "UPDATE to_id" a one-shot, but the delete+insert form
// keeps semantics identical when the new (from,to,...) key happens to
// already exist -- the INSERT OR IGNORE drops the dup, just like the
// in-memory store's bucket-replace).
func (s *Store) ReindexEdge(e *graph.Edge, oldTo string) {
	if e == nil || oldTo == e.To {
		return
	}
	s.writeMu.Lock()
	defer s.writeMu.Unlock()

	if _, err := s.stmtDeleteEdgeByKey.Exec(e.From, oldTo, string(e.Kind), e.FilePath, e.Line); err != nil {
		panicOnFatal(err)
		return
	}
	if err := s.insertEdgeLocked(s.stmtInsertEdge, e); err != nil {
		panicOnFatal(err)
		return
	}
}

// reindexChunkSize bounds the number of edge re-binds per BEGIN/COMMIT.
// Same shape as the bbolt sibling: large enough to amortise the
// per-tx overhead (BEGIN+COMMIT plus WAL fsync) but small enough that
// the WAL doesn't balloon and a crash mid-batch only loses ≤chunk
// mutations.
const reindexChunkSize = 5000

// ReindexEdges chunks the batch into reindexChunkSize-mutation
// transactions and runs each through prepared statements re-used
// across the chunk. Per-edge ReindexEdge was the resolver hot path
// (10k+ calls = 10k+ BEGIN/COMMIT pairs); this collapses them to two.
func (s *Store) ReindexEdges(batch []graph.EdgeReindex) {
	if len(batch) == 0 {
		return
	}
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	for i := 0; i < len(batch); i += reindexChunkSize {
		end := minInt(i+reindexChunkSize, len(batch))
		chunk := batch[i:end]
		tx, err := s.db.Begin()
		if err != nil {
			panicOnFatal(err)
			return
		}
		delStmt := tx.Stmt(s.stmtDeleteEdgeByKey)
		insStmt := tx.Stmt(s.stmtInsertEdge)
		for _, r := range chunk {
			if r.Edge == nil || r.OldTo == r.Edge.To {
				continue
			}
			if _, err := delStmt.Exec(r.Edge.From, r.OldTo, string(r.Edge.Kind), r.Edge.FilePath, r.Edge.Line); err != nil {
				_ = tx.Rollback()
				panicOnFatal(err)
				return
			}
			if err := s.insertEdgeLocked(insStmt, r.Edge); err != nil {
				_ = tx.Rollback()
				panicOnFatal(err)
				return
			}
		}
		if err := tx.Commit(); err != nil {
			panicOnFatal(err)
			return
		}
	}
}

// SetEdgeProvenanceBatch chunks origin promotions into one BEGIN/
// COMMIT per chunk and bumps the in-process revision counter once
// per actual change, matching the per-edge SetEdgeProvenance's
// semantics. Returns the total number of edges whose Origin changed.
func (s *Store) SetEdgeProvenanceBatch(batch []graph.EdgeProvenanceUpdate) int {
	if len(batch) == 0 {
		return 0
	}
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	totalChanged := 0
	for i := 0; i < len(batch); i += reindexChunkSize {
		end := minInt(i+reindexChunkSize, len(batch))
		chunk := batch[i:end]
		tx, err := s.db.Begin()
		if err != nil {
			panicOnFatal(err)
			return totalChanged
		}
		selStmt := tx.Stmt(s.stmtSelectEdgeOrigin)
		updStmt := tx.Stmt(s.stmtUpdateEdgeOrigin)
		chunkChanged := 0
		for _, u := range chunk {
			if u.Edge == nil {
				continue
			}
			var storedOrigin string
			row := selStmt.QueryRow(u.Edge.From, u.Edge.To, string(u.Edge.Kind), u.Edge.FilePath, u.Edge.Line)
			if err := row.Scan(&storedOrigin); err != nil {
				if errors.Is(err, sql.ErrNoRows) {
					continue
				}
				_ = tx.Rollback()
				panicOnFatal(err)
				return totalChanged
			}
			if storedOrigin == u.NewOrigin {
				continue
			}
			newTier := u.Edge.Tier
			if newTier != "" {
				newTier = graph.ResolvedBy(u.NewOrigin)
			}
			if _, err := updStmt.Exec(u.NewOrigin, newTier, u.Edge.From, u.Edge.To, string(u.Edge.Kind), u.Edge.FilePath, u.Edge.Line); err != nil {
				_ = tx.Rollback()
				panicOnFatal(err)
				return totalChanged
			}
			u.Edge.Origin = u.NewOrigin
			if u.Edge.Tier != "" {
				u.Edge.Tier = newTier
			}
			chunkChanged++
		}
		if err := tx.Commit(); err != nil {
			panicOnFatal(err)
			return totalChanged
		}
		if chunkChanged > 0 {
			s.edgeIdentityRevs.Add(int64(chunkChanged))
		}
		totalChanged += chunkChanged
	}
	return totalChanged
}

func minInt(a, b int) int {
	if a < b {
		return a
	}
	return b
}

// RemoveEdge deletes every edge between (from, to) with the given
// kind. Returns true iff at least one row was deleted.
func (s *Store) RemoveEdge(from, to string, kind graph.EdgeKind) bool {
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	res, err := s.stmtRemoveEdge.Exec(from, to, string(kind))
	if err != nil {
		panicOnFatal(err)
		return false
	}
	n, err := res.RowsAffected()
	if err != nil {
		panicOnFatal(err)
		return false
	}
	return n > 0
}

// EvictFile removes every node anchored to filePath and every edge
// that touches one of those nodes. Returns (nodesRemoved,
// edgesRemoved).
func (s *Store) EvictFile(filePath string) (nodesRemoved, edgesRemoved int) {
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	return s.evictByScopeLocked(s.stmtSelectFileNodeIDs, s.stmtDeleteNodeByFile, filePath)
}

// EvictRepo removes every node in repoPrefix and every edge that
// touches one. Returns (nodesRemoved, edgesRemoved).
func (s *Store) EvictRepo(repoPrefix string) (nodesRemoved, edgesRemoved int) {
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	return s.evictByScopeLocked(s.stmtSelectRepoNodeIDs, s.stmtDeleteNodeByRepo, repoPrefix)
}

// evictByScopeLocked is the shared body of EvictFile / EvictRepo --
// collect the affected node IDs, delete every edge touching one of
// them, then delete the nodes themselves.
func (s *Store) evictByScopeLocked(selectIDs, deleteNodes *sql.Stmt, scope string) (int, int) {
	rows, err := selectIDs.Query(scope)
	if err != nil {
		panicOnFatal(err)
		return 0, 0
	}
	var ids []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			_ = rows.Close()
			panicOnFatal(err)
			return 0, 0
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		panicOnFatal(err)
		return 0, 0
	}
	_ = rows.Close()
	if len(ids) == 0 {
		return 0, 0
	}

	// Delete every edge touching one of these nodes. A DELETE-per-node
	// (… WHERE from_id = ? OR to_id = ?) is one statement round-trip and
	// WAL commit per node — hundreds of them when evicting a large file on
	// the per-edit reindex path. Instead delete in chunked IN batches
	// keyed on from_id then to_id, so each chunk is one index-driven
	// DELETE; the chunk size stays under SQLite's bound-variable limit.
	var edgesRemoved int
	const evictEdgeChunk = 900
	for _, col := range []string{"from_id", "to_id"} {
		for start := 0; start < len(ids); start += evictEdgeChunk {
			end := minInt(start+evictEdgeChunk, len(ids))
			chunk := ids[start:end]
			placeholders := strings.TrimSuffix(strings.Repeat("?,", len(chunk)), ",")
			args := make([]any, len(chunk))
			for i, id := range chunk {
				args[i] = id
			}
			res, err := s.db.Exec(`DELETE FROM edges WHERE `+col+` IN (`+placeholders+`)`, args...)
			if err != nil {
				panicOnFatal(err)
				return 0, edgesRemoved
			}
			if n, err := res.RowsAffected(); err == nil {
				edgesRemoved += int(n)
			}
		}
	}

	res, err := deleteNodes.Exec(scope)
	if err != nil {
		panicOnFatal(err)
		return 0, edgesRemoved
	}
	n, err := res.RowsAffected()
	if err != nil {
		panicOnFatal(err)
		return 0, edgesRemoved
	}
	return int(n), edgesRemoved
}

// -- reads ---------------------------------------------------------------

func (s *Store) GetNode(id string) *graph.Node {
	row := s.stmtGetNode.QueryRow(id)
	n, err := scanNode(row)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		panicOnFatal(err)
		return nil
	}
	return n
}

func (s *Store) GetNodeByQualName(qualName string) *graph.Node {
	if qualName == "" {
		return nil
	}
	row := s.stmtGetNodeByQual.QueryRow(qualName)
	n, err := scanNode(row)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		panicOnFatal(err)
		return nil
	}
	return n
}

func (s *Store) FindNodesByName(name string) []*graph.Node {
	return s.queryNodes(s.stmtFindByName, name)
}

func (s *Store) FindNodesByNameInRepo(name, repoPrefix string) []*graph.Node {
	return s.queryNodes(s.stmtFindByNameInRepo, name, repoPrefix)
}

func (s *Store) GetFileNodes(filePath string) []*graph.Node {
	return s.queryNodes(s.stmtFileNodes, filePath)
}

func (s *Store) GetRepoNodes(repoPrefix string) []*graph.Node {
	return s.queryNodes(s.stmtRepoNodes, repoPrefix)
}

func (s *Store) AllNodes() []*graph.Node {
	return s.queryNodes(s.stmtAllNodes)
}

func (s *Store) queryNodes(stmt *sql.Stmt, args ...any) []*graph.Node {
	rows, err := stmt.Query(args...)
	if err != nil {
		panicOnFatal(err)
		return nil
	}
	defer rows.Close()
	var out []*graph.Node
	for rows.Next() {
		n, err := scanNode(rows)
		if err != nil {
			panicOnFatal(err)
			return out
		}
		out = append(out, n)
	}
	return out
}

// GetRepoNonContentNodes is the graph.NonContentNodeReader fast path: a
// SQL-level enumeration that drops CONTENT (data_class="content") section
// nodes, so the code-oriented passes never materialise a content-heavy
// repo's hundreds of thousands of sections. data_class is a promoted node
// column for rows written by the flat codec; legacy JSON rows (no column)
// fall back to json_extract, guarded by json_valid so the flat / gob blobs
// — which are not JSON — are skipped without error. The NULL-safe
// `IS NOT 'content'` keeps every node whose data_class is absent or carries
// any other value. An empty repoPrefix spans all repos.
func (s *Store) GetRepoNonContentNodes(repoPrefix string) []*graph.Node {
	const filter = `COALESCE(data_class, CASE WHEN json_valid(CAST(meta AS TEXT)) THEN json_extract(CAST(meta AS TEXT), '$.data_class') END) IS NOT 'content'`
	if repoPrefix == "" {
		return s.scanNodeQuery(`SELECT ` + lookupNodeCols + ` FROM nodes WHERE ` + filter)
	}
	return s.scanNodeQuery(`SELECT `+lookupNodeCols+` FROM nodes WHERE repo_prefix = ? AND `+filter, repoPrefix)
}

// GetRepoNodesLight is the graph.LightNodeReader fast path: repo_prefix
// still uses the nodes_by_repo index, but the meta column is left out of
// the projection entirely, so a repo's already-enriched majority never
// crosses the driver boundary as a blob to decode. See LightNodeReader's
// doc for the read-only-use invariant this projection depends on.
func (s *Store) GetRepoNodesLight(repoPrefix string) []*graph.Node {
	rows, err := s.db.Query(`SELECT `+lookupNodeColsLight+` FROM nodes WHERE repo_prefix = ?`, repoPrefix)
	if err != nil {
		panicOnFatal(err)
		return nil
	}
	defer rows.Close()
	var out []*graph.Node
	for rows.Next() {
		n, err := scanNodeLight(rows)
		if err != nil {
			panicOnFatal(err)
			return out
		}
		out = append(out, n)
	}
	return out
}

// scanNodeQuery runs an ad-hoc node SELECT (columns = lookupNodeCols) and
// scans its rows into nodes — for the few non-hot enumerations that need a
// WHERE clause the prepared statements don't cover.
func (s *Store) scanNodeQuery(query string, args ...any) []*graph.Node {
	rows, err := s.db.Query(query, args...)
	if err != nil {
		panicOnFatal(err)
		return nil
	}
	defer rows.Close()
	var out []*graph.Node
	for rows.Next() {
		n, err := scanNode(rows)
		if err != nil {
			panicOnFatal(err)
			return out
		}
		out = append(out, n)
	}
	return out
}

func (s *Store) GetOutEdges(nodeID string) []*graph.Edge {
	return s.queryEdges(s.stmtOutEdges, nodeID)
}

// EdgeExists reports whether an edge with exactly this identity is present --
// (from, to, kind, file_path, line) is the edges UNIQUE key, so this is a
// single indexed point lookup: no row decode, no Meta gob, no per-edge
// allocation, unlike GetOutEdges. The resolver's liveness guard
// (edgeStillLive) calls this once per applied edge on the cold/full pass; the
// difference from scanning + gob-decoding all of `from`'s out-edges is a
// dominant share of resolve cost on a large graph.
func (s *Store) EdgeExists(from, to string, kind graph.EdgeKind, filePath string, line int) bool {
	var one int
	err := s.stmtEdgeExists.QueryRow(from, to, string(kind), filePath, line).Scan(&one)
	if err == sql.ErrNoRows {
		return false
	}
	if err != nil {
		panicOnFatal(err)
		return false
	}
	return true
}

// GetOutEdgesLight returns a node's out-edges without decoding the
// per-edge Meta blob -- for hot dataflow lookups that need only
// endpoints/kind/line. The returned edges have a nil Meta.
func (s *Store) GetOutEdgesLight(nodeID string) []*graph.Edge {
	return s.queryEdgesLight(s.stmtOutEdgesLight, nodeID)
}

func (s *Store) GetInEdges(nodeID string) []*graph.Edge {
	return s.queryEdges(s.stmtInEdges, nodeID)
}

// GetOutEdgesForNodes fetches the out-edges of many nodes in one batched query
// (chunked) instead of a round-trip per node. The single-file resolve path
// walks every node of the edited file, which is an N+1 query storm on a disk
// backend; this collapses it to one query per chunk. Edges are grouped by
// their from_id; nodes with no out-edges are absent from the map.
func (s *Store) GetOutEdgesForNodes(ids []string) map[string][]*graph.Edge {
	out := make(map[string][]*graph.Edge, len(ids))
	if len(ids) == 0 {
		return out
	}
	seen := make(map[string]struct{}, len(ids))
	uniq := make([]string, 0, len(ids))
	for _, id := range ids {
		if id == "" {
			continue
		}
		if _, ok := seen[id]; ok {
			continue
		}
		seen[id] = struct{}{}
		uniq = append(uniq, id)
	}
	for i := 0; i < len(uniq); i += lookupChunkSize {
		end := minInt(i+lookupChunkSize, len(uniq))
		chunk := uniq[i:end]
		ph := make([]string, len(chunk))
		args := make([]any, len(chunk))
		for j, id := range chunk {
			ph[j] = "?"
			args[j] = id
		}
		q := `SELECT ` + lookupEdgeCols + ` FROM edges WHERE from_id IN (` + strings.Join(ph, ",") + `)`
		for _, e := range s.queryEdgesSQL(q, args...) {
			out[e.From] = append(out[e.From], e)
		}
	}
	return out
}

func (s *Store) AllEdges() []*graph.Edge {
	return s.queryEdges(s.stmtAllEdges)
}

// GetRepoEdges returns every edge whose source node has the given
// RepoPrefix. The pre-Store idiom — GetRepoNodes(r) followed by
// GetOutEdges(n.ID) per node — was O(repo_nodes) prepared-statement
// invocations, which on a multi-repo workspace dominated the
// per-repo extractor passes. A single JOIN over edges/nodes keyed
// on n.repo_prefix runs as one prepared statement and hits the
// existing repo_prefix index.
func (s *Store) GetRepoEdges(repoPrefix string) []*graph.Edge {
	if repoPrefix == "" {
		return nil
	}
	return s.queryEdges(s.stmtRepoEdges, repoPrefix)
}

func (s *Store) queryEdges(stmt *sql.Stmt, args ...any) []*graph.Edge {
	rows, err := stmt.Query(args...)
	if err != nil {
		panicOnFatal(err)
		return nil
	}
	defer rows.Close()
	var out []*graph.Edge
	for rows.Next() {
		e, err := scanEdge(rows)
		if err != nil {
			panicOnFatal(err)
			return out
		}
		out = append(out, e)
	}
	return out
}

// queryEdgesLight mirrors queryEdges but scans each row without its
// meta blob (scanEdgeLight), leaving Meta nil. Only for callers that
// never read edge Meta.
func (s *Store) queryEdgesLight(stmt *sql.Stmt, args ...any) []*graph.Edge {
	rows, err := stmt.Query(args...)
	if err != nil {
		panicOnFatal(err)
		return nil
	}
	defer rows.Close()
	var out []*graph.Edge
	for rows.Next() {
		e, err := scanEdgeLight(rows)
		if err != nil {
			panicOnFatal(err)
			return out
		}
		out = append(out, e)
	}
	return out
}

// -- counts and stats -----------------------------------------------------

func (s *Store) NodeCount() int {
	var n int
	if err := s.stmtNodeCount.QueryRow().Scan(&n); err != nil {
		panicOnFatal(err)
		return 0
	}
	return n
}

func (s *Store) EdgeCount() int {
	var n int
	if err := s.stmtEdgeCount.QueryRow().Scan(&n); err != nil {
		panicOnFatal(err)
		return 0
	}
	return n
}

func (s *Store) Stats() graph.GraphStats {
	st := graph.GraphStats{
		ByKind:     map[string]int{},
		ByLanguage: map[string]int{},
	}
	st.TotalNodes = s.NodeCount()
	st.TotalEdges = s.EdgeCount()

	rows, err := s.stmtStatsByKind.Query()
	if err != nil {
		panicOnFatal(err)
		return st
	}
	for rows.Next() {
		var kind string
		var n int
		if err := rows.Scan(&kind, &n); err != nil {
			_ = rows.Close()
			panicOnFatal(err)
			return st
		}
		st.ByKind[kind] = n
	}
	_ = rows.Close()

	rows, err = s.stmtStatsByLanguage.Query()
	if err != nil {
		panicOnFatal(err)
		return st
	}
	for rows.Next() {
		var lang string
		var n int
		if err := rows.Scan(&lang, &n); err != nil {
			_ = rows.Close()
			panicOnFatal(err)
			return st
		}
		st.ByLanguage[lang] = n
	}
	_ = rows.Close()
	return st
}

func (s *Store) RepoStats() map[string]graph.GraphStats {
	out := map[string]graph.GraphStats{}
	rows, err := s.stmtRepoStatsNodes.Query()
	if err != nil {
		panicOnFatal(err)
		return out
	}
	for rows.Next() {
		var repo, kind, lang string
		var n int
		if err := rows.Scan(&repo, &kind, &lang, &n); err != nil {
			_ = rows.Close()
			panicOnFatal(err)
			return out
		}
		st, ok := out[repo]
		if !ok {
			st = graph.GraphStats{ByKind: map[string]int{}, ByLanguage: map[string]int{}}
		}
		st.TotalNodes += n
		st.ByKind[kind] += n
		st.ByLanguage[lang] += n
		out[repo] = st
	}
	_ = rows.Close()

	rows, err = s.stmtRepoStatsEdges.Query()
	if err != nil {
		panicOnFatal(err)
		return out
	}
	for rows.Next() {
		var repo string
		var n int
		if err := rows.Scan(&repo, &n); err != nil {
			_ = rows.Close()
			panicOnFatal(err)
			return out
		}
		st, ok := out[repo]
		if !ok {
			st = graph.GraphStats{ByKind: map[string]int{}, ByLanguage: map[string]int{}}
		}
		st.TotalEdges = n
		out[repo] = st
	}
	_ = rows.Close()
	return out
}

func (s *Store) RepoPrefixes() []string {
	rows, err := s.stmtRepoPrefixes.Query()
	if err != nil {
		panicOnFatal(err)
		return nil
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var p string
		if err := rows.Scan(&p); err != nil {
			panicOnFatal(err)
			return out
		}
		out = append(out, p)
	}
	return out
}

// -- provenance verification ---------------------------------------------

func (s *Store) EdgeIdentityRevisions() int {
	return int(s.edgeIdentityRevs.Load())
}

// VerifyEdgeIdentities is a no-op for the SQL backend: the in-memory
// store's invariant is "the same *Edge pointer lives in both
// adjacency views". The SQL store has a single row per edge, so the
// invariant is trivially satisfied -- no walk can find a divergence
// to report.
func (s *Store) VerifyEdgeIdentities() error { return nil }

// -- memory estimation (advisory) ----------------------------------------

// perRowByteEstimate is a deliberately rough per-row byte cost --
// the disk backend doesn't have an in-memory footprint to report, so
// the contract (per Store interface comment) is "return what you can
// compute and callers treat the result as advisory". The conformance
// test only checks NodeCount.
const (
	perNodeByteEstimate = 256
	perEdgeByteEstimate = 128
)

func (s *Store) RepoMemoryEstimate(repoPrefix string) graph.RepoMemoryEstimate {
	var est graph.RepoMemoryEstimate
	var n, e int
	if err := s.stmtRepoNodeCount.QueryRow(repoPrefix).Scan(&n); err != nil {
		panicOnFatal(err)
		return est
	}
	if err := s.stmtRepoEdgeCount.QueryRow(repoPrefix).Scan(&e); err != nil {
		panicOnFatal(err)
		return est
	}
	est.NodeCount = n
	est.EdgeCount = e
	est.NodeBytes = uint64(n) * perNodeByteEstimate
	est.EdgeBytes = uint64(e) * perEdgeByteEstimate
	return est
}

// memEstTTL bounds how long AllRepoMemoryEstimates serves a memoised
// result before recomputing. The estimate is advisory (status display),
// so a few seconds of staleness is fine, and the TTL keeps a burst of
// status polls from each triggering a full COUNT … GROUP BY scan.
const memEstTTL = 3 * time.Second

func cloneRepoMemEstimates(m map[string]graph.RepoMemoryEstimate) map[string]graph.RepoMemoryEstimate {
	out := make(map[string]graph.RepoMemoryEstimate, len(m))
	for k, v := range m {
		out[k] = v
	}
	return out
}

func (s *Store) AllRepoMemoryEstimates() map[string]graph.RepoMemoryEstimate {
	// Hold memEstMu across the recompute so a burst of concurrent status
	// polls collapses onto one scan: the first caller computes and
	// caches, the rest block briefly and then hit the fresh cache.
	s.memEstMu.Lock()
	defer s.memEstMu.Unlock()
	if s.memEstVal != nil && time.Since(s.memEstAt) < memEstTTL {
		return cloneRepoMemEstimates(s.memEstVal)
	}

	out := map[string]graph.RepoMemoryEstimate{}
	rows, err := s.stmtAllRepoCountsNodes.Query()
	if err != nil {
		panicOnFatal(err)
		return out
	}
	for rows.Next() {
		var repo string
		var n int
		if err := rows.Scan(&repo, &n); err != nil {
			_ = rows.Close()
			panicOnFatal(err)
			return out
		}
		est := out[repo]
		est.NodeCount = n
		est.NodeBytes = uint64(n) * perNodeByteEstimate
		out[repo] = est
	}
	_ = rows.Close()

	rows, err = s.stmtAllRepoCountsEdges.Query()
	if err != nil {
		panicOnFatal(err)
		return out
	}
	for rows.Next() {
		var repo string
		var n int
		if err := rows.Scan(&repo, &n); err != nil {
			_ = rows.Close()
			panicOnFatal(err)
			return out
		}
		est := out[repo]
		est.EdgeCount = n
		est.EdgeBytes = uint64(n) * perEdgeByteEstimate
		out[repo] = est
	}
	_ = rows.Close()

	// Cache only on the full-success path — the early error returns above
	// leave a partial `out` and must not poison the cache.
	s.memEstVal = out
	s.memEstAt = time.Now()
	return cloneRepoMemEstimates(out)
}

// -- helpers --------------------------------------------------------------

// panicOnFatal turns truly catastrophic SQLite errors (closed DB,
// schema mismatch, disk-full at insert time) into a panic so callers
// see them, while letting expected sql.ErrNoRows / busy / no-affected
// callers stay quiet. The graph.Store interface deliberately does not
// surface errors -- it mirrors the in-memory store's "everything
// succeeds" contract -- so a fatal storage failure cannot be ignored.
//
// Caller contract: on a teardown-race error panicOnFatal RETURNS rather than
// panicking, so a caller that keeps using the query result after it returns
// MUST nil-check first. `rows, err := db.Query(...); panicOnFatal(err)` leaves
// rows == nil on a swallowed error, and the subsequent rows.Close() /
// rows.Next() would SIGSEGV — the aggregator reads early-return their empty
// value on nil rows for exactly this reason. In one line: fatal panics; a
// teardown-race read returns empty.
func panicOnFatal(err error) {
	if err == nil {
		return
	}
	if errors.Is(err, sql.ErrNoRows) {
		return
	}
	// A closed statement / database / connection is a teardown race, not
	// data corruption: Close() shuts the store (daemon shutdown, restart,
	// or store swap) while an in-flight reader -- e.g. a deferred
	// parallel-enrich goroutine still holding a cached *sql.Stmt -- runs a
	// query. Crashing the whole daemon over a benign shutdown race is
	// strictly worse than the read returning empty (or a winding-down write
	// being dropped), so treat these as non-fatal.
	if errors.Is(err, sql.ErrConnDone) || isStoreClosedErr(err) {
		return
	}
	panic(fmt.Errorf("store_sqlite: %w", err))
}

// isStoreClosedErr reports whether err is the database/sql sentinel for a
// closed prepared statement or a closed database -- string-matched because
// database/sql does not export these as typed errors.
func isStoreClosedErr(err error) bool {
	msg := err.Error()
	return strings.Contains(msg, "statement is closed") ||
		strings.Contains(msg, "database is closed")
}

// -- predicate-shaped reads ---------------------------------------------
//
// Each method runs one indexed SELECT and streams rows back via the
// iter.Seq[T] yield callback. Stops cleanly when yield returns false.
// Heavier than the equivalent bolt path (sql parsing + driver row
// materialisation) but cuts the resolver's wasted full-table scans
// down to "match-only" cardinality, which is the whole point.

// All three predicate iterators here MATERIALISE the query result
// into a slice before yielding, then iterate the slice. This avoids
// a deadlock peculiar to the SQLite backend's single-connection
// pool: a streaming rows-cursor holds THE connection, and any
// callback in the yield body that re-enters the store (e.g. GetNode
// to resolve an edge's caller) blocks forever waiting on the same
// connection. Materialise-then-yield releases the connection before
// the body runs, so re-entrant store calls work.
//
// The "predicate-shaped" win still holds: the indexed SELECT only
// fetches matching rows, not the whole table. We give up streaming
// memory savings (we still build a Go slice of *Edge / *Node) but
// keep the structural advantage that the row count flowing through
// scanEdge is proportional to the result, not the table.

// EdgesByKind: indexed SELECT on the (kind) column.
func (s *Store) EdgesByKind(kind graph.EdgeKind) iter.Seq[*graph.Edge] {
	return func(yield func(*graph.Edge) bool) {
		out := s.queryEdgesSQL(`
SELECT from_id, to_id, kind, file_path, line, confidence, confidence_label, origin, tier, cross_repo, meta, resolve_terminal, resolve_terminal_reason
FROM edges WHERE kind = ?`, string(kind))
		for _, e := range out {
			if !yield(e) {
				return
			}
		}
	}
}

// NodesByKind: indexed SELECT on the (kind) column.
func (s *Store) NodesByKind(kind graph.NodeKind) iter.Seq[*graph.Node] {
	return func(yield func(*graph.Node) bool) {
		out := s.queryNodesSQL(`SELECT `+lookupNodeCols+` FROM nodes WHERE kind = ?`, string(kind))
		for _, n := range out {
			if !yield(n) {
				return
			}
		}
	}
}

// EdgesWithUnresolvedTarget yields edges whose target is an unresolved stub
// in either form graph.IsUnresolvedTarget recognises. Filters on the
// is_unresolved generated column (see isUnresolvedColumnDDL) rather than
// re-deriving the to_id pattern match in SQL: measured 2.7x faster than the
// equivalent to_id-based OR query on a real 26-repo store (7.96s -> 2.95s for
// the same 847,684-row result) because the boolean index's bookmark lookups
// land in ascending rowid order, unlike a to_id-ordered index's.
//
// Gate-owned fn-value placeholders (graph.FnValuePlaceholderMarker,
// `unresolved::fnvalue::<name>`) are excluded on top of is_unresolved: the
// master resolver can never bind them, so they are pure pending-set bloat here
// (a live store held millions). The bare form is dropped by the range predicate
// — which rides edges_by_to(to_id) — using the ':;' range end from
// isUnresolvedColumnDDL's idiom (';' == ':'+1); the multi-repo COPY-rewrite form
// is dropped by the NOT LIKE, matching IsFnValuePlaceholder's infix shape.
func (s *Store) EdgesWithUnresolvedTarget() iter.Seq[*graph.Edge] {
	return func(yield func(*graph.Edge) bool) {
		out := s.queryEdgesSQL(`
SELECT from_id, to_id, kind, file_path, line, confidence, confidence_label, origin, tier, cross_repo, meta, resolve_terminal, resolve_terminal_reason
FROM edges WHERE is_unresolved = 1
  AND NOT (to_id >= 'unresolved::fnvalue::' AND to_id < 'unresolved::fnvalue:;')
  AND to_id NOT LIKE '%::unresolved::fnvalue::%'`)
		for _, e := range out {
			if !yield(e) {
				return
			}
		}
	}
}

// queryEdgesSQL runs an edge-shaped SELECT, materialises the rows
// into a slice, and closes the rows-cursor before returning —
// releasing the underlying sql.Conn so the predicate-iterator's
// callback body is free to make re-entrant store calls without
// deadlocking on the MaxOpenConns=1 pool. Companion to the existing
// queryEdges helper that takes a *sql.Stmt; this one takes a raw
// SQL string so the predicate iterators can pass inline queries.
func (s *Store) queryEdgesSQL(q string, args ...any) []*graph.Edge {
	rows, err := s.db.Query(q, args...)
	if err != nil {
		return nil
	}
	defer rows.Close()
	var out []*graph.Edge
	for rows.Next() {
		e, err := scanEdge(rows)
		if err != nil || e == nil {
			continue
		}
		out = append(out, e)
	}
	return out
}

// queryNodesSQL is the node-shaped sibling of queryEdgesSQL.
func (s *Store) queryNodesSQL(q string, args ...any) []*graph.Node {
	rows, err := s.db.Query(q, args...)
	if err != nil {
		return nil
	}
	defer rows.Close()
	var out []*graph.Node
	for rows.Next() {
		n, err := scanNode(rows)
		if err != nil || n == nil {
			continue
		}
		out = append(out, n)
	}
	return out
}

// lookupChunkSize bounds the IN-list parameter count per SQL query.
// SQLite's default SQLITE_MAX_VARIABLE_NUMBER is 32766 in modern
// builds, but staying well under that keeps query plans stable and
// avoids surprising the parser on monster lists.
const lookupChunkSize = 5000

// GetNodesByIDs collapses N per-id SELECTs into ⌈N/chunk⌉ queries
// of the form `SELECT … FROM nodes WHERE id IN (?, ?, …)`. The
// resolver fires hundreds of thousands of these on a large pass;
// chunking turns hundreds of seconds into single-digit seconds.
func (s *Store) GetNodesByIDs(ids []string) map[string]*graph.Node {
	if len(ids) == 0 {
		return nil
	}
	// Dedupe + skip empty up front to keep the chunk loop honest.
	seen := make(map[string]struct{}, len(ids))
	uniq := make([]string, 0, len(ids))
	for _, id := range ids {
		if id == "" {
			continue
		}
		if _, ok := seen[id]; ok {
			continue
		}
		seen[id] = struct{}{}
		uniq = append(uniq, id)
	}
	out := make(map[string]*graph.Node, len(uniq))
	const nodeCols = lookupNodeCols
	for i := 0; i < len(uniq); i += lookupChunkSize {
		end := minInt(i+lookupChunkSize, len(uniq))
		chunk := uniq[i:end]
		placeholders := strings.Repeat(",?", len(chunk))[1:]
		q := `SELECT ` + nodeCols + ` FROM nodes WHERE id IN (` + placeholders + `)`
		args := make([]any, len(chunk))
		for j, id := range chunk {
			args[j] = id
		}
		for _, n := range s.queryNodesSQL(q, args...) {
			if n != nil {
				out[n.ID] = n
			}
		}
	}
	return out
}

// FindNodesByNames collapses N per-name FindNodesByName queries into
// one `SELECT … FROM nodes WHERE name IN (…)` plus an in-Go bucket
// by name. The (name) index makes the SELECT seek-driven, and the
// caller sees the same map[name][]*Node it would have built by
// calling FindNodesByName N times.
func (s *Store) FindNodesByNames(names []string) map[string][]*graph.Node {
	if len(names) == 0 {
		return nil
	}
	seen := make(map[string]struct{}, len(names))
	uniq := make([]string, 0, len(names))
	for _, name := range names {
		if name == "" {
			continue
		}
		if _, ok := seen[name]; ok {
			continue
		}
		seen[name] = struct{}{}
		uniq = append(uniq, name)
	}
	out := make(map[string][]*graph.Node, len(uniq))
	const nodeCols = lookupNodeCols
	for i := 0; i < len(uniq); i += lookupChunkSize {
		end := minInt(i+lookupChunkSize, len(uniq))
		chunk := uniq[i:end]
		placeholders := strings.Repeat(",?", len(chunk))[1:]
		q := `SELECT ` + nodeCols + ` FROM nodes WHERE name IN (` + placeholders + `)`
		args := make([]any, len(chunk))
		for j, name := range chunk {
			args[j] = name
		}
		for _, n := range s.queryNodesSQL(q, args...) {
			if n == nil {
				continue
			}
			out[n.Name] = append(out[n.Name], n)
		}
	}
	return out
}

// -- BulkLoader implementation -------------------------------------------

// BeginBulkLoad / FlushBulk (the graph.BulkLoader bracket) live in
// bulk_load.go. The bracket exists so the indexer's in-memory shadow
// swap activates — the resolver and its post-resolve passes run against
// an in-memory *Graph at nanosecond latency, and the final drain dumps
// the resolved graph to sqlite in one shot. On a first/empty cold index
// the bracket additionally engages a bulk-persist fast path (dropped
// secondary indexes + synchronous=OFF on a pinned connection).