memlay 0.1.4

//! Worktree-local SQLite index (PRD §12). The database is a disposable cache
//! derived from canonical record files and repository content; it is never
//! the source of truth and can always be rebuilt.

pub mod files;

use crate::config::Config;
use crate::errors::{err, ErrorCode};
use crate::gitx::Repo;
use crate::memgraph::MemoryGraph;
use crate::records::store::LoadedRecord;
use crate::team::LayerMap;
use anyhow::{Context, Result};
use rusqlite::Connection;
use std::path::PathBuf;

pub const SCHEMA_VERSION: i64 = 2;

pub struct Index {
    pub conn: Connection,
    pub db_path: PathBuf,
}

impl Index {
    /// Open (creating/migrating if needed) the worktree-local index.
    pub fn open(repo: &Repo) -> Result<Index> {
        let dir = repo.state_dir();
        std::fs::create_dir_all(&dir)?;
        let db_path = dir.join("index.sqlite");
        Self::open_at(db_path)
    }

    pub fn open_at(db_path: PathBuf) -> Result<Index> {
        let conn =
            Connection::open(&db_path).with_context(|| format!("opening {}", db_path.display()))?;
        conn.busy_timeout(std::time::Duration::from_millis(5_000))?;
        conn.pragma_update(None, "journal_mode", "WAL")?;
        conn.pragma_update(None, "foreign_keys", "ON")?;
        conn.pragma_update(None, "synchronous", "NORMAL")?;
        let mut index = Index { conn, db_path };
        index.migrate()?;
        Ok(index)
    }

    fn migrate(&mut self) -> Result<()> {
        let version: i64 = self
            .conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap_or(0);
        if version == SCHEMA_VERSION {
            return Ok(());
        }
        if version > SCHEMA_VERSION {
            return Err(err(
                ErrorCode::IndexCorrupt,
                format!(
                    "index schema version {version} is newer than this memlay build supports ({SCHEMA_VERSION}); run 'memlay index rebuild'"
                ),
            ));
        }
        let tx = self.conn.transaction()?;
        if version > 0 {
            // The index is a disposable cache (PRD §12): older schemas are
            // recreated rather than migrated in place.
            tx.execute_batch(DROP_SQL)?;
        }
        tx.execute_batch(SCHEMA_SQL)?;
        tx.pragma_update(None, "user_version", SCHEMA_VERSION)?;
        tx.commit()?;
        Ok(())
    }

    pub fn integrity_check(&self) -> Result<()> {
        let ok: String = self
            .conn
            .query_row("PRAGMA integrity_check", [], |r| r.get(0))
            .map_err(|e| {
                err(
                    ErrorCode::IndexCorrupt,
                    format!("integrity check failed: {e}"),
                )
            })?;
        if ok != "ok" {
            return Err(err(
                ErrorCode::IndexCorrupt,
                format!("integrity check: {ok}"),
            ));
        }
        Ok(())
    }

    #[cfg_attr(not(test), allow(dead_code))]
    pub fn meta_get(&self, key: &str) -> Option<String> {
        self.conn
            .query_row("SELECT value FROM meta WHERE key = ?1", [key], |r| r.get(0))
            .ok()
    }

    #[cfg_attr(not(test), allow(dead_code))]
    pub fn meta_set(&self, key: &str, value: &str) -> Result<()> {
        self.conn.execute(
            "INSERT INTO meta(key, value) VALUES (?1, ?2)
             ON CONFLICT(key) DO UPDATE SET value = excluded.value",
            [key, value],
        )?;
        Ok(())
    }

    /// Replace the record-derived tables from the currently loaded memory.
    /// Records are few relative to code; a transactional full refresh keeps
    /// the cache trivially consistent with the canonical files.
    pub fn refresh_records(
        &mut self,
        records: &[LoadedRecord],
        graph: &MemoryGraph,
        layers: &LayerMap,
        provenance: &std::collections::HashMap<String, crate::gitx::CommitInfo>,
    ) -> Result<()> {
        let tx = self.conn.transaction()?;
        tx.execute_batch(
            "DELETE FROM record_scopes; DELETE FROM record_evidence;
             DELETE FROM record_relations; DELETE FROM current_heads;
             DELETE FROM semantic_conflicts; DELETE FROM records;
             DELETE FROM records_fts;",
        )?;
        {
            let mut ins_record = tx.prepare(
                "INSERT INTO records(id, key, canonical_key, kind, op, summary, rationale,
                    confidence, created_at, writer, human, agent, session, pr, issue,
                    layer, file, valid, head, conflicted, intro_commit, intro_author, intro_at)
                 VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22,?23)",
            )?;
            let mut ins_scope = tx.prepare(
                "INSERT INTO record_scopes(record_id, scope_type, value) VALUES (?1,?2,?3)",
            )?;
            let mut ins_evidence = tx.prepare(
                "INSERT INTO record_evidence(record_id, etype, value) VALUES (?1,?2,?3)",
            )?;
            let mut ins_relation = tx.prepare(
                "INSERT INTO record_relations(record_id, relation, target) VALUES (?1,?2,?3)",
            )?;
            let mut ins_fts = tx.prepare(
                "INSERT INTO records_fts(record_id, key, summary, rationale, details) VALUES (?1,?2,?3,?4,?5)",
            )?;

            for lr in records {
                let r = &lr.record;
                let canonical = graph.resolve_key(&r.key);
                let state = graph.keys.get(&canonical);
                let is_head = state.map(|s| s.head_ids.contains(&r.id)).unwrap_or(false);
                let conflicted = state.map(|s| s.conflicted).unwrap_or(false);
                let id = r.id.to_string();
                ins_record.execute(rusqlite::params![
                    id,
                    r.key,
                    canonical,
                    r.kind.as_str(),
                    r.op.as_str(),
                    r.summary,
                    r.rationale,
                    r.confidence.as_str(),
                    r.created_at.to_rfc3339(),
                    r.writer,
                    r.human,
                    r.agent,
                    r.session,
                    r.pr,
                    r.issue,
                    layers.layer_of(&lr.rel_path).as_str(),
                    lr.rel_path,
                    lr.is_valid() as i64,
                    is_head as i64,
                    conflicted as i64,
                    provenance.get(&lr.rel_path).map(|p| p.oid.clone()),
                    provenance
                        .get(&lr.rel_path)
                        .map(|p| format!("{} <{}>", p.author_name, p.author_email)),
                    provenance.get(&lr.rel_path).map(|p| p.author_date.clone()),
                ])?;
                for p in &r.paths {
                    ins_scope.execute(rusqlite::params![id, "path", p])?;
                }
                for s in &r.symbols {
                    ins_scope.execute(rusqlite::params![id, "symbol", s])?;
                }
                for t in &r.tags {
                    ins_scope.execute(rusqlite::params![id, "tag", t])?;
                }
                for e in &r.evidence {
                    ins_evidence.execute(rusqlite::params![id, e.etype.as_str(), e.value])?;
                }
                for u in &r.supersedes {
                    ins_relation.execute(rusqlite::params![id, "supersedes", u.to_string()])?;
                }
                for u in &r.related {
                    ins_relation.execute(rusqlite::params![id, "related", u.to_string()])?;
                }
                if lr.is_valid() {
                    ins_fts.execute(rusqlite::params![
                        id,
                        r.key,
                        r.summary,
                        r.rationale.clone().unwrap_or_default(),
                        r.details.join(" "),
                    ])?;
                }
            }

            let mut ins_head = tx.prepare(
                "INSERT INTO current_heads(canonical_key, kind, head_ids, active, conflicted)
                 VALUES (?1,?2,?3,?4,?5)",
            )?;
            for (key, state) in &graph.keys {
                let ids: Vec<String> = state.head_ids.iter().map(|u| u.to_string()).collect();
                ins_head.execute(rusqlite::params![
                    key,
                    state.kind.as_str(),
                    ids.join(","),
                    state.active as i64,
                    state.conflicted as i64,
                ])?;
            }
            let mut ins_conflict = tx.prepare(
                "INSERT INTO semantic_conflicts(canonical_key, kind, head_ids, alias_induced)
                 VALUES (?1,?2,?3,?4)",
            )?;
            for c in &graph.conflicts {
                let ids: Vec<String> = c.head_ids.iter().map(|u| u.to_string()).collect();
                ins_conflict.execute(rusqlite::params![
                    c.canonical_key,
                    c.kind.as_str(),
                    ids.join(","),
                    c.alias_induced as i64,
                ])?;
            }
        }
        tx.commit()?;
        Ok(())
    }

    /// Rebuild the database beside the old one and atomically replace it
    /// (PRD §12.4). The caller re-runs indexing afterwards.
    pub fn rebuild(repo: &Repo) -> Result<Index> {
        let dir = repo.state_dir();
        std::fs::create_dir_all(&dir)?;
        let final_path = dir.join("index.sqlite");
        let new_path = dir.join("index.sqlite.rebuild");
        let _ = std::fs::remove_file(&new_path);
        {
            let new_index = Index::open_at(new_path.clone())?;
            new_index.integrity_check()?;
            // Close cleanly before the swap.
            drop(new_index);
        }
        let _ = std::fs::remove_file(dir.join("index.sqlite-wal"));
        let _ = std::fs::remove_file(dir.join("index.sqlite-shm"));
        if final_path.exists() {
            std::fs::remove_file(&final_path)
                .with_context(|| "removing old index (is another memlay process running?)")?;
        }
        std::fs::rename(&new_path, &final_path)?;
        Index::open_at(final_path)
    }

    pub fn stats(&self) -> Result<IndexStats> {
        let count = |sql: &str| -> i64 { self.conn.query_row(sql, [], |r| r.get(0)).unwrap_or(0) };
        Ok(IndexStats {
            files: count("SELECT COUNT(*) FROM files"),
            symbols: count("SELECT COUNT(*) FROM symbols"),
            records: count("SELECT COUNT(*) FROM records"),
            heads: count("SELECT COUNT(*) FROM current_heads"),
            conflicts: count("SELECT COUNT(*) FROM semantic_conflicts"),
            parse_failures: count("SELECT COUNT(*) FROM files WHERE parse_status = 'error'"),
            db_bytes: std::fs::metadata(&self.db_path)
                .map(|m| m.len())
                .unwrap_or(0),
            languages: {
                let mut stmt = self.conn.prepare(
                    "SELECT language, COUNT(*) FROM files GROUP BY language ORDER BY language",
                )?;
                let rows = stmt
                    .query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?)))?
                    .filter_map(|r| r.ok())
                    .collect();
                rows
            },
        })
    }
}

#[derive(Debug, serde::Serialize)]
pub struct IndexStats {
    pub files: i64,
    pub symbols: i64,
    pub records: i64,
    pub heads: i64,
    pub conflicts: i64,
    pub parse_failures: i64,
    pub db_bytes: u64,
    pub languages: Vec<(String, i64)>,
}

/// Refresh everything derivable from the current checkout: records into the
/// index plus incremental file discovery.
pub fn update_all(repo: &Repo, cfg: &Config) -> Result<Index> {
    let loaded = crate::records::store::load_all(&repo.root)?;
    let graph = crate::memgraph::build(&loaded.records);
    let layers = crate::team::layer_map(repo, cfg);
    let provenance = repo.records_provenance();
    let mut index = Index::open(repo)?;
    index.refresh_records(&loaded.records, &graph, &layers, &provenance)?;
    files::update_files(&mut index, repo, cfg)?;
    files::parse_pending(&mut index, repo, cfg)?;
    crate::retrieval::update_stale(&mut index)?;
    Ok(index)
}

const SCHEMA_SQL: &str = r#"
CREATE TABLE IF NOT EXISTS meta (
    key   TEXT PRIMARY KEY,
    value TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS files (
    path         TEXT PRIMARY KEY,
    language     TEXT NOT NULL,
    content_hash TEXT NOT NULL,
    git_blob_oid TEXT,
    size         INTEGER NOT NULL,
    origin       TEXT NOT NULL DEFAULT 'worktree',
    indexed_at   TEXT NOT NULL,
    parse_status TEXT NOT NULL DEFAULT 'pending',
    parse_error  TEXT
);

CREATE TABLE IF NOT EXISTS symbols (
    id            INTEGER PRIMARY KEY,
    name          TEXT NOT NULL,
    qualified_name TEXT NOT NULL,
    kind          TEXT NOT NULL,
    path          TEXT NOT NULL,
    start_line    INTEGER NOT NULL,
    end_line      INTEGER NOT NULL,
    signature     TEXT,
    content_hash  TEXT NOT NULL,
    is_test       INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_symbols_name ON symbols(name);
CREATE INDEX IF NOT EXISTS idx_symbols_qualified ON symbols(qualified_name);
CREATE INDEX IF NOT EXISTS idx_symbols_path ON symbols(path);

CREATE TABLE IF NOT EXISTS symbol_edges (
    src_path  TEXT NOT NULL,
    edge_type TEXT NOT NULL,
    dst       TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_edges_src ON symbol_edges(src_path);
CREATE INDEX IF NOT EXISTS idx_edges_dst ON symbol_edges(dst);

CREATE TABLE IF NOT EXISTS records (
    id            TEXT PRIMARY KEY,
    key           TEXT NOT NULL,
    canonical_key TEXT NOT NULL,
    kind          TEXT NOT NULL,
    op            TEXT NOT NULL,
    summary       TEXT NOT NULL,
    rationale     TEXT,
    confidence    TEXT NOT NULL,
    created_at    TEXT NOT NULL,
    writer        TEXT NOT NULL,
    human         TEXT,
    agent         TEXT,
    session       TEXT,
    pr            TEXT,
    issue         TEXT,
    layer         TEXT NOT NULL,
    file          TEXT NOT NULL,
    valid         INTEGER NOT NULL,
    head          INTEGER NOT NULL,
    conflicted    INTEGER NOT NULL,
    stale         INTEGER NOT NULL DEFAULT 0,
    stale_reason  TEXT,
    intro_commit  TEXT,
    intro_author  TEXT,
    intro_at      TEXT
);
CREATE INDEX IF NOT EXISTS idx_records_key ON records(canonical_key);
CREATE INDEX IF NOT EXISTS idx_records_kind ON records(kind);

CREATE TABLE IF NOT EXISTS record_scopes (
    record_id  TEXT NOT NULL,
    scope_type TEXT NOT NULL,
    value      TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_scopes_value ON record_scopes(value);
CREATE INDEX IF NOT EXISTS idx_scopes_record ON record_scopes(record_id);

CREATE TABLE IF NOT EXISTS record_evidence (
    record_id TEXT NOT NULL,
    etype     TEXT NOT NULL,
    value     TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_evidence_record ON record_evidence(record_id);

CREATE TABLE IF NOT EXISTS record_relations (
    record_id TEXT NOT NULL,
    relation  TEXT NOT NULL,
    target    TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_relations_record ON record_relations(record_id);
CREATE INDEX IF NOT EXISTS idx_relations_target ON record_relations(target);

CREATE TABLE IF NOT EXISTS current_heads (
    canonical_key TEXT PRIMARY KEY,
    kind          TEXT NOT NULL,
    head_ids      TEXT NOT NULL,
    active        INTEGER NOT NULL,
    conflicted    INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS semantic_conflicts (
    canonical_key TEXT NOT NULL,
    kind          TEXT NOT NULL,
    head_ids      TEXT NOT NULL,
    alias_induced INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS stale_records (
    record_id TEXT PRIMARY KEY,
    reason    TEXT NOT NULL
);

CREATE VIRTUAL TABLE IF NOT EXISTS records_fts USING fts5(
    record_id UNINDEXED, key, summary, rationale, details
);
CREATE VIRTUAL TABLE IF NOT EXISTS files_fts USING fts5(path);
CREATE VIRTUAL TABLE IF NOT EXISTS symbols_fts USING fts5(
    symbol_id UNINDEXED, name, qualified_name, path, parts
);
"#;

const DROP_SQL: &str = r#"
DROP TABLE IF EXISTS meta;
DROP TABLE IF EXISTS files;
DROP TABLE IF EXISTS symbols;
DROP TABLE IF EXISTS symbol_edges;
DROP TABLE IF EXISTS records;
DROP TABLE IF EXISTS record_scopes;
DROP TABLE IF EXISTS record_evidence;
DROP TABLE IF EXISTS record_relations;
DROP TABLE IF EXISTS current_heads;
DROP TABLE IF EXISTS semantic_conflicts;
DROP TABLE IF EXISTS stale_records;
DROP TABLE IF EXISTS records_fts;
DROP TABLE IF EXISTS files_fts;
DROP TABLE IF EXISTS symbols_fts;
"#;

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn schema_creates_and_fts5_available() {
        let tmp = tempfile::tempdir().unwrap();
        let index = Index::open_at(tmp.path().join("index.sqlite")).unwrap();
        index.integrity_check().unwrap();
        // FTS5 must be usable (bundled SQLite).
        index
            .conn
            .execute(
                "INSERT INTO records_fts(record_id, key, summary, rationale, details)
                 VALUES ('x', 'a.b', 'webhook retries', '', '')",
                [],
            )
            .unwrap();
        let hits: i64 = index
            .conn
            .query_row(
                "SELECT COUNT(*) FROM records_fts WHERE records_fts MATCH 'webhook'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(hits, 1);
    }

    #[test]
    fn meta_round_trip_and_reopen() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("index.sqlite");
        {
            let index = Index::open_at(path.clone()).unwrap();
            index.meta_set("k", "v1").unwrap();
            index.meta_set("k", "v2").unwrap();
        }
        let index = Index::open_at(path).unwrap();
        assert_eq!(index.meta_get("k").as_deref(), Some("v2"));
    }
}