grit-core 0.2.4

//! Lossless JSONL export/import of the full graph + oplog (Design
//! Invariant 8). This format ships in v0.1 and is a permanent compatibility
//! surface: one JSON object per line, first line is a header, every other
//! line is `{"t": "<record type>", ...}`.
//!
//! Embeddings are deliberately absent: they are recomputable local state
//! tagged with a model id (Design Invariant 5); `embedding_meta` is exported
//! so the importing side knows what to re-embed with.

use std::io::{BufRead, Write};
use std::path::Path;

use rusqlite::{Connection, params};
use serde::{Deserialize, Serialize};
use serde_json::Value as Json;

use crate::Grit;
use crate::error::{Error, Result};
use crate::migrate::SCHEMA_VERSION;

/// Header line of an export stream.
#[derive(Debug, Serialize, Deserialize)]
struct Header {
    grit_export: u32,
    schema_version: i64,
}

/// One data line of an export stream.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "t", rename_all = "snake_case")]
enum Record {
    Node {
        id: String,
        kind: String,
        name: String,
        summary: String,
        attrs: Json,
        group_id: String,
        created_at: i64,
        expired_at: Option<i64>,
        merged_into: Option<String>,
        hlc: String,
    },
    Edge {
        id: String,
        src: String,
        dst: String,
        rel: String,
        fact: String,
        attrs: Json,
        group_id: String,
        valid_at: Option<i64>,
        invalid_at: Option<i64>,
        created_at: i64,
        expired_at: Option<i64>,
        hlc: String,
    },
    Episode {
        id: String,
        source: String,
        /// Source-kind tag; absent in pre-v3 exports.
        #[serde(default)]
        kind: String,
        content: String,
        occurred_at: i64,
        group_id: String,
        created_at: i64,
        hlc: String,
    },
    Mention {
        episode_id: String,
        target_id: String,
    },
    EdgeInvalidation {
        edge_id: String,
        invalid_at: i64,
        recorded_at: i64,
        hlc: String,
    },
    NodeUpdate {
        node_id: String,
        field: String,
        value: String,
        hlc: String,
    },
    Purged {
        id: String,
        merged_into: Option<String>,
    },
    Oplog {
        id: String,
        hlc: String,
        device_id: String,
        op: Json,
        applied_at: i64,
    },
    EmbeddingMeta {
        model_id: String,
        dim: i64,
        model_version: String,
    },
}

/// Counts of what an import inserted.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct ImportStats {
    /// Node rows inserted.
    pub nodes: usize,
    /// Edge rows inserted.
    pub edges: usize,
    /// Episode rows inserted.
    pub episodes: usize,
    /// Mention rows inserted.
    pub mentions: usize,
    /// Oplog entries inserted.
    pub oplog: usize,
}

impl Grit {
    /// Stream the entire database (graph tables + oplog + embedding metadata,
    /// not the recomputable vectors) as JSONL. The inverse of
    /// [`import_jsonl`].
    ///
    /// # Example
    /// ```no_run
    /// # use grit_core::{Grit, Options};
    /// # let g = Grit::open("memory.db", Options::new("laptop"))?;
    /// let mut out = Vec::new();
    /// g.export_jsonl(&mut out)?;
    /// # Ok::<(), grit_core::Error>(())
    /// ```
    pub fn export_jsonl<W: Write>(&self, mut out: W) -> Result<()> {
        let conn = self.read();
        // One read transaction: all table dumps come from the same committed
        // snapshot — a write landing mid-export cannot tear the stream.
        let tx = conn.unchecked_transaction()?;
        let mut line = |record: &dyn erased::Line| -> Result<()> {
            out.write_all(record.to_json_line()?.as_bytes())?;
            out.write_all(b"\n")?;
            Ok(())
        };
        line(&Header {
            grit_export: 1,
            schema_version: SCHEMA_VERSION,
        })?;

        let mut stmt = tx.prepare(
            "SELECT id, kind, name, summary, attrs, group_id, created_at, expired_at,
                    merged_into, hlc
             FROM nodes ORDER BY id",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok(Record::Node {
                id: r.get(0)?,
                kind: r.get(1)?,
                name: r.get(2)?,
                summary: r.get(3)?,
                attrs: parse_json(r.get::<_, String>(4)?)?,
                group_id: r.get(5)?,
                created_at: r.get(6)?,
                expired_at: r.get(7)?,
                merged_into: r.get(8)?,
                hlc: r.get(9)?,
            })
        })?;
        for row in rows {
            line(&row?)?;
        }

        let mut stmt = tx.prepare(
            "SELECT id, src, dst, rel, fact, attrs, group_id, valid_at, invalid_at,
                    created_at, expired_at, hlc
             FROM edges ORDER BY id",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok(Record::Edge {
                id: r.get(0)?,
                src: r.get(1)?,
                dst: r.get(2)?,
                rel: r.get(3)?,
                fact: r.get(4)?,
                attrs: parse_json(r.get::<_, String>(5)?)?,
                group_id: r.get(6)?,
                valid_at: r.get(7)?,
                invalid_at: r.get(8)?,
                created_at: r.get(9)?,
                expired_at: r.get(10)?,
                hlc: r.get(11)?,
            })
        })?;
        for row in rows {
            line(&row?)?;
        }

        let mut stmt = tx.prepare(
            "SELECT id, source, kind, content, occurred_at, group_id, created_at, hlc
             FROM episodes ORDER BY id",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok(Record::Episode {
                id: r.get(0)?,
                source: r.get(1)?,
                kind: r.get(2)?,
                content: r.get(3)?,
                occurred_at: r.get(4)?,
                group_id: r.get(5)?,
                created_at: r.get(6)?,
                hlc: r.get(7)?,
            })
        })?;
        for row in rows {
            line(&row?)?;
        }

        let mut stmt = conn
            .prepare("SELECT episode_id, target_id FROM mentions ORDER BY episode_id, target_id")?;
        let rows = stmt.query_map([], |r| {
            Ok(Record::Mention {
                episode_id: r.get(0)?,
                target_id: r.get(1)?,
            })
        })?;
        for row in rows {
            line(&row?)?;
        }

        let mut stmt = tx.prepare(
            "SELECT edge_id, invalid_at, recorded_at, hlc FROM edge_invalidations
             ORDER BY edge_id, invalid_at",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok(Record::EdgeInvalidation {
                edge_id: r.get(0)?,
                invalid_at: r.get(1)?,
                recorded_at: r.get(2)?,
                hlc: r.get(3)?,
            })
        })?;
        for row in rows {
            line(&row?)?;
        }

        let mut stmt = tx.prepare(
            "SELECT node_id, field, value, hlc FROM node_updates
             ORDER BY node_id, field",
        )?;
        let rows = stmt.query_map([], |r| {
            Ok(Record::NodeUpdate {
                node_id: r.get(0)?,
                field: r.get(1)?,
                value: r.get(2)?,
                hlc: r.get(3)?,
            })
        })?;
        for row in rows {
            line(&row?)?;
        }

        let mut stmt = tx.prepare("SELECT id, merged_into FROM purged ORDER BY id")?;
        let rows = stmt.query_map([], |r| {
            Ok(Record::Purged {
                id: r.get(0)?,
                merged_into: r.get(1)?,
            })
        })?;
        for row in rows {
            line(&row?)?;
        }

        let mut stmt =
            tx.prepare("SELECT id, hlc, device_id, op, applied_at FROM oplog ORDER BY seq")?;
        let rows = stmt.query_map([], |r| {
            Ok(Record::Oplog {
                id: r.get(0)?,
                hlc: r.get(1)?,
                device_id: r.get(2)?,
                op: parse_json(r.get::<_, String>(3)?)?,
                applied_at: r.get(4)?,
            })
        })?;
        for row in rows {
            line(&row?)?;
        }

        let mut stmt =
            tx.prepare("SELECT model_id, dim, model_version FROM embedding_meta WHERE id = 1")?;
        let rows = stmt.query_map([], |r| {
            Ok(Record::EmbeddingMeta {
                model_id: r.get(0)?,
                dim: r.get(1)?,
                model_version: r.get(2)?,
            })
        })?;
        for row in rows {
            line(&row?)?;
        }
        Ok(())
    }
}

/// Load a JSONL export into a **fresh** database file at `db_path` (errors if
/// the file already contains data — imports never merge). FTS mirrors are
/// rebuilt by the schema triggers as rows insert.
pub fn import_jsonl(db_path: impl AsRef<Path>, input: impl BufRead) -> Result<ImportStats> {
    crate::vecext::register_sqlite_vec();
    let mut conn = Connection::open(db_path)?;
    crate::configure(&conn)?;
    crate::migrate::migrate(&mut conn)?;

    let existing: i64 = conn.query_row(
        "SELECT (SELECT COUNT(*) FROM oplog) + (SELECT COUNT(*) FROM nodes)",
        [],
        |r| r.get(0),
    )?;
    if existing > 0 {
        return Err(Error::Import(
            "refusing to import into a non-empty database".into(),
        ));
    }

    let mut lines = input.lines();
    let header_line = lines
        .next()
        .ok_or_else(|| Error::Import("empty import stream".into()))??;
    let header: Header = serde_json::from_str(&header_line)
        .map_err(|e| Error::Import(format!("bad header: {e}")))?;
    if header.schema_version > SCHEMA_VERSION {
        return Err(Error::SchemaTooNew {
            found: header.schema_version,
            supported: SCHEMA_VERSION,
        });
    }

    let tx = conn.transaction()?;
    let mut stats = ImportStats::default();
    for (lineno, text) in lines.enumerate() {
        let text = text?;
        if text.trim().is_empty() {
            continue;
        }
        let record: Record = serde_json::from_str(&text)
            .map_err(|e| Error::Import(format!("line {}: {e}", lineno + 2)))?;
        let check_uuid = |field: &str, value: &str| -> Result<()> {
            uuid::Uuid::parse_str(value).map(|_| ()).map_err(|_| {
                Error::Import(format!(
                    "line {}: {field} is not a uuid: {value:?}",
                    lineno + 2
                ))
            })
        };
        match record {
            Record::Node {
                id,
                kind,
                name,
                summary,
                attrs,
                group_id,
                created_at,
                expired_at,
                merged_into,
                hlc,
            } => {
                check_uuid("id", &id)?;
                if let Some(m) = &merged_into {
                    check_uuid("merged_into", m)?;
                }
                tx.execute(
                    "INSERT INTO nodes (id, kind, name, summary, attrs, group_id,
                                        created_at, expired_at, merged_into, hlc)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
                    params![
                        id,
                        kind,
                        name,
                        summary,
                        attrs.to_string(),
                        group_id,
                        created_at,
                        expired_at,
                        merged_into,
                        hlc
                    ],
                )?;
                stats.nodes += 1;
            }
            Record::Edge {
                id,
                src,
                dst,
                rel,
                fact,
                attrs,
                group_id,
                valid_at,
                invalid_at,
                created_at,
                expired_at,
                hlc,
            } => {
                check_uuid("id", &id)?;
                check_uuid("src", &src)?;
                check_uuid("dst", &dst)?;
                tx.execute(
                    "INSERT INTO edges (id, src, dst, rel, fact, attrs, group_id, valid_at,
                                        invalid_at, created_at, expired_at, hlc)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
                    params![
                        id,
                        src,
                        dst,
                        rel,
                        fact,
                        attrs.to_string(),
                        group_id,
                        valid_at,
                        invalid_at,
                        created_at,
                        expired_at,
                        hlc
                    ],
                )?;
                stats.edges += 1;
            }
            Record::Episode {
                id,
                source,
                kind,
                content,
                occurred_at,
                group_id,
                created_at,
                hlc,
            } => {
                check_uuid("id", &id)?;
                tx.execute(
                    "INSERT INTO episodes (id, source, kind, content, occurred_at, group_id,
                                           created_at, hlc)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
                    params![
                        id,
                        source,
                        kind,
                        content,
                        occurred_at,
                        group_id,
                        created_at,
                        hlc
                    ],
                )?;
                stats.episodes += 1;
            }
            Record::Mention {
                episode_id,
                target_id,
            } => {
                check_uuid("episode_id", &episode_id)?;
                check_uuid("target_id", &target_id)?;
                tx.execute(
                    "INSERT INTO mentions (episode_id, target_id) VALUES (?1, ?2)",
                    params![episode_id, target_id],
                )?;
                stats.mentions += 1;
            }
            Record::EdgeInvalidation {
                edge_id,
                invalid_at,
                recorded_at,
                hlc,
            } => {
                check_uuid("edge_id", &edge_id)?;
                tx.execute(
                    "INSERT INTO edge_invalidations (edge_id, invalid_at, recorded_at, hlc)
                     VALUES (?1, ?2, ?3, ?4)",
                    params![edge_id, invalid_at, recorded_at, hlc],
                )?;
            }
            Record::NodeUpdate {
                node_id,
                field,
                value,
                hlc,
            } => {
                check_uuid("node_id", &node_id)?;
                if !matches!(field.as_str(), "name" | "summary" | "kind" | "attrs") {
                    return Err(Error::Import(format!(
                        "line {}: unknown node_updates field: {field:?}",
                        lineno + 2
                    )));
                }
                tx.execute(
                    "INSERT INTO node_updates (node_id, field, value, hlc)
                     VALUES (?1, ?2, ?3, ?4)",
                    params![node_id, field, value, hlc],
                )?;
            }
            Record::Purged { id, merged_into } => {
                check_uuid("id", &id)?;
                if let Some(m) = &merged_into {
                    check_uuid("merged_into", m)?;
                }
                tx.execute(
                    "INSERT INTO purged (id, merged_into) VALUES (?1, ?2)",
                    params![id, merged_into],
                )?;
            }
            Record::Oplog {
                id,
                hlc,
                device_id,
                op,
                applied_at,
            } => {
                check_uuid("id", &id)?;
                tx.execute(
                    "INSERT INTO oplog (id, hlc, device_id, op, applied_at)
                     VALUES (?1, ?2, ?3, ?4, ?5)",
                    params![id, hlc, device_id, op.to_string(), applied_at],
                )?;
                stats.oplog += 1;
            }
            Record::EmbeddingMeta {
                model_id,
                dim,
                model_version,
            } => {
                tx.execute(
                    "INSERT INTO embedding_meta (id, model_id, dim, model_version)
                     VALUES (1, ?1, ?2, ?3)",
                    params![model_id, dim, model_version],
                )?;
            }
        }
    }
    tx.commit()?;
    Ok(stats)
}

fn parse_json(s: String) -> rusqlite::Result<Json> {
    serde_json::from_str(&s).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(e))
    })
}

/// Tiny object-safety shim so `line()` above can take both `Header` and
/// `Record` without generics-in-closure gymnastics.
mod erased {
    use serde::Serialize;

    use crate::error::Result;

    pub(super) trait Line {
        fn to_json_line(&self) -> Result<String>;
    }

    impl<T: Serialize> Line for T {
        fn to_json_line(&self) -> Result<String> {
            Ok(serde_json::to_string(self)?)
        }
    }
}