innate_core/

migrate.rs

//! Schema migration runner — 4.0 → current chain.
//!
//! Each step is atomic: BEGIN IMMEDIATE … COMMIT. Any failure rolls back the
//! entire step and returns an error — no half-migrated state.

use std::path::Path;

use rusqlite::{Connection, OptionalExtension};

use crate::errors::{InnateError, Result};

// Embedded migration SQL ordered from lowest to highest target version.
const MIGRATIONS: &[(&str, &str, &str)] = &[
    ("4.0", "4.1", include_str!("migrations/4.0_to_4.1.sql")),
    ("4.1", "4.2", include_str!("migrations/4.1_to_4.2.sql")),
    ("4.2", "4.3", include_str!("migrations/4.2_to_4.3.sql")),
    ("4.3", "4.4", include_str!("migrations/4.3_to_4.4.sql")),
    ("4.4", "4.5", include_str!("migrations/4.4_to_4.5.sql")),
    ("4.5", "4.5.1", include_str!("migrations/4.5_to_4.5.1.sql")),
    (
        "4.5.1",
        "4.5.2",
        include_str!("migrations/4.5.1_to_4.5.2.sql"),
    ),
    ("4.5.2", "4.6", include_str!("migrations/4.5.2_to_4.6.sql")),
    ("4.6", "4.7", include_str!("migrations/4.6_to_4.7.sql")),
    ("4.7", "4.8", include_str!("migrations/4.7_to_4.8.sql")),
    ("4.8", "4.9", include_str!("migrations/4.8_to_4.9.sql")),
    ("4.9", "4.10", include_str!("migrations/4.9_to_4.10.sql")),
    ("4.10", "4.11", include_str!("migrations/4.10_to_4.11.sql")),
    ("4.11", "4.12", include_str!("migrations/4.11_to_4.12.sql")),
    ("4.12", "4.13", include_str!("migrations/4.12_to_4.13.sql")),
    ("4.13", "4.14", include_str!("migrations/4.13_to_4.14.sql")),
    ("4.14", "4.15", include_str!("migrations/4.14_to_4.15.sql")),
    ("4.15", "4.16", include_str!("migrations/4.15_to_4.16.sql")),
    ("4.16", "4.17", include_str!("migrations/4.16_to_4.17.sql")),
    ("4.17", "4.18", include_str!("migrations/4.17_to_4.18.sql")),
    ("4.18", "4.19", include_str!("migrations/4.18_to_4.19.sql")),
    ("4.19", "4.20", include_str!("migrations/4.19_to_4.20.sql")),
    ("4.20", "4.21", include_str!("migrations/4.20_to_4.21.sql")),
];

const TARGET: &str = "4.21";

/// The schema version this binary migrates to (single source of truth for callers
/// that want to report it, e.g. the CLI `migrate` "nothing to do" message).
pub fn target_version() -> &'static str {
    TARGET
}

/// Run all pending migrations on `db_path`. Idempotent if already at target.
/// Returns the list of migration steps executed.
pub fn run_migrations(db_path: impl AsRef<Path>) -> Result<Vec<String>> {
    let conn = Connection::open(db_path.as_ref())?;
    conn.execute_batch(
        "PRAGMA journal_mode=WAL;
         PRAGMA foreign_keys=ON;
         PRAGMA synchronous=NORMAL;
         -- Migration opens its own connection; without a busy_timeout a second
         -- process starting concurrently (e.g. the MCP server alongside a CLI
         -- command) would hit SQLITE_BUSY the instant a step's BEGIN IMMEDIATE
         -- collides. Match Storage's 5s wait so concurrent starts serialise
         -- instead of failing the migration outright.
         PRAGMA busy_timeout=5000;",
    )?;

    let current = schema_version(&conn)?;
    if current == TARGET {
        return Ok(vec![]);
    }

    let mut applied = vec![];
    let mut ver = current;

    for (from, to, sql) in MIGRATIONS {
        if ver_tuple(&ver) >= ver_tuple(to) {
            continue; // already at or beyond this step
        }
        if ver_tuple(&ver) < ver_tuple(from) {
            return Err(InnateError::Other(format!(
                "Migration gap: database at {ver}, expected {from}→{to}. \
                 Is the database from an unsupported version?"
            )));
        }
        let copy_last_used = *to == "4.12" && column_exists(&conn, "chunks", "last_used_at")?;
        // 方案 C:provenance 列在 4.14→4.15 条件添加(ALTER ADD COLUMN 无 IF NOT EXISTS)。
        let add_provenance =
            *to == "4.15" && !column_exists(&conn, "confidence_evidence", "provenance")?;
        // 混合检索:FTS5 词法通道。仅当 chunks 拥有索引列时建/灌(部分 schema
        // 测试夹具缺这些列 → 跳过,不致命;真实库恒满足)。DDL 自带 IF NOT EXISTS,
        // 回填 DELETE+INSERT,可重复执行。
        let add_fts = *to == "4.16"
            && column_exists(&conn, "chunks", "content")?
            && column_exists(&conn, "chunks", "trigger_desc")?
            && column_exists(&conn, "chunks", "skill_name")?;
        // agent 来源列(4.16→4.17)。ALTER ADD COLUMN 无 IF NOT EXISTS,故按表是否
        // 存在(借已知列探测)且 agent 列尚缺时条件添加,对部分 schema 夹具不致命。
        let add_agent_log = *to == "4.17"
            && column_exists(&conn, "episodic_log", "trace_id")?
            && !column_exists(&conn, "episodic_log", "agent")?;
        let add_agent_chunk = *to == "4.17"
            && column_exists(&conn, "chunks", "content")?
            && !column_exists(&conn, "chunks", "agent")?;
        // 关联实体索引回填(4.17→4.18):DDL 由本步 SQL 建表,存量 chunk 的实体由
        // Rust(extract_entities)回填。仅当 chunks 拥有内容列时执行(部分 schema
        // 测试夹具缺列 → 跳过,不致命;真实库恒满足)。可重复执行(INSERT OR IGNORE)。
        let backfill_entities = *to == "4.18"
            && column_exists(&conn, "chunks", "content")?
            && column_exists(&conn, "chunks", "trigger_desc")?;
        // 可观测性 ts 索引(4.18→4.19):CREATE INDEX 依赖既有表的 ts 列。部分迁移
        // 测试夹具只建简化 schema(缺这些表/列),故每个索引各自用 column_exists 守卫;
        // 真实库三表恒有 ts,索引照常建立。各索引独立判断(夹具可能只缺其中之一)。
        let add_ts_indexes = *to == "4.19";
        // Run the step atomically.
        conn.execute_batch("BEGIN IMMEDIATE")?;
        let r = conn.execute_batch(sql);
        match r {
            Ok(()) => {
                if add_fts {
                    if let Err(error) =
                        conn.execute_batch(include_str!("migrations/4.16_fts.sql"))
                    {
                        let _ = conn.execute_batch("ROLLBACK");
                        return Err(error.into());
                    }
                }
                if add_agent_log {
                    if let Err(error) =
                        conn.execute_batch("ALTER TABLE episodic_log ADD COLUMN agent TEXT")
                    {
                        let _ = conn.execute_batch("ROLLBACK");
                        return Err(error.into());
                    }
                }
                if add_agent_chunk {
                    if let Err(error) =
                        conn.execute_batch("ALTER TABLE chunks ADD COLUMN agent TEXT")
                    {
                        let _ = conn.execute_batch("ROLLBACK");
                        return Err(error.into());
                    }
                }
                if add_provenance {
                    if let Err(error) = conn.execute_batch(
                        "ALTER TABLE confidence_evidence
                         ADD COLUMN provenance TEXT NOT NULL DEFAULT 'observed'",
                    ) {
                        let _ = conn.execute_batch("ROLLBACK");
                        return Err(error.into());
                    }
                }
                if backfill_entities {
                    if let Err(error) = backfill_chunk_entities(&conn) {
                        let _ = conn.execute_batch("ROLLBACK");
                        return Err(error);
                    }
                }
                if add_ts_indexes {
                    for (table, col, idx) in [
                        ("episodic_log", "ts", "idx_log_ts"),
                        ("usage_trace", "ts", "idx_trace_ts"),
                        ("feedback_events", "ts", "idx_feedback_ts"),
                    ] {
                        if column_exists(&conn, table, col).unwrap_or(false) {
                            if let Err(error) = conn.execute_batch(&format!(
                                "CREATE INDEX IF NOT EXISTS {idx} ON {table}({col})"
                            )) {
                                let _ = conn.execute_batch("ROLLBACK");
                                return Err(error.into());
                            }
                        }
                    }
                }
                if copy_last_used {
                    if let Err(error) = conn.execute(
                        "UPDATE chunks
                             SET last_used_base=CASE
                               WHEN EXISTS (
                                 SELECT 1 FROM usage_trace u
                                 WHERE u.chunk_id=chunks.id AND u.event='used'
                               ) THEN NULL
                               ELSE last_used_at
                             END",
                        [],
                    ) {
                        let _ = conn.execute_batch("ROLLBACK");
                        return Err(error.into());
                    }
                }
                conn.execute_batch("COMMIT")?;
                applied.push(format!("{from}→{to}"));
                ver = to.to_string();
            }
            Err(e) => {
                let _ = conn.execute_batch("ROLLBACK");
                return Err(InnateError::Other(format!(
                    "Migration {from}→{to} failed: {e}"
                )));
            }
        }
    }

    if ver != TARGET {
        return Err(InnateError::Other(format!(
            "After all migrations, schema version is {ver}, expected {TARGET}."
        )));
    }

    Ok(applied)
}

/// Backfill `chunk_entities` from existing chunks using the deterministic
/// extractor. Runs inside the migration's open transaction. Idempotent
/// (`INSERT OR IGNORE` on the composite PK), so re-running is safe.
fn backfill_chunk_entities(conn: &Connection) -> Result<()> {
    let rows: Vec<(String, String, Option<String>)> = {
        let mut stmt = conn.prepare("SELECT id, content, trigger_desc FROM chunks")?;
        let mapped = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, Option<String>>(2)?,
            ))
        })?;
        mapped.collect::<rusqlite::Result<Vec<_>>>()?
    };
    let mut ins = conn.prepare(
        "INSERT OR IGNORE INTO chunk_entities (chunk_id, entity, etype, weight)
         VALUES (?1, ?2, ?3, 1.0)",
    )?;
    for (id, content, trigger) in rows {
        for e in crate::entities::extract_entities(&content, trigger.as_deref()) {
            ins.execute(rusqlite::params![id, e.entity, e.etype])?;
        }
    }
    Ok(())
}

fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool> {
    let sql = format!("SELECT COUNT(*) FROM pragma_table_info('{table}') WHERE name=?");
    Ok(conn.query_row(&sql, [column], |row| row.get::<_, i64>(0))? > 0)
}

fn schema_version(conn: &Connection) -> Result<String> {
    let has_meta: bool = conn.query_row(
        "SELECT count(*) FROM sqlite_master WHERE type='table' AND name='meta'",
        [],
        |r| r.get::<_, i64>(0),
    )? > 0;

    if !has_meta {
        return Err(InnateError::Other(
            "Database has no meta table — cannot migrate. \
             Use `innate` to create a fresh database."
                .into(),
        ));
    }

    let ver: Option<String> = conn
        .query_row(
            "SELECT value FROM meta WHERE key='schema_version'",
            [],
            |r| r.get(0),
        )
        .optional()?;

    ver.ok_or_else(|| InnateError::Other("meta table missing schema_version".into()))
}

fn ver_tuple(v: &str) -> (u32, u32, u32) {
    let parts: Vec<u32> = v.split('.').filter_map(|s| s.parse().ok()).collect();
    (
        parts.first().copied().unwrap_or(0),
        parts.get(1).copied().unwrap_or(0),
        parts.get(2).copied().unwrap_or(0),
    )
}