klieo-provenance 3.5.0

//! `SqliteProvenanceRepository` — the first production
//! [`ProvenanceRepository`]: a sqlite-backed, append-only Merkle chain.
//!
//! Chain semantics are NOT reinvented here. `append` reconstructs a
//! single-entry [`ProvenanceChain`] seeded from the stored head, then
//! delegates to [`ProvenanceChain::append`] so the new entry's sequence,
//! `previous_hash_hex` linkage, and `entry_hash_hex` are minted by the
//! crate's own v2 hashing rules. A chain read back out of this store and
//! reassembled via [`reassemble_chain`] therefore passes
//! [`ProvenanceChain::verify`] — same rule the in-memory reference relies on.
//!
//! ### Blocking I/O on the async path
//!
//! rusqlite's `Connection` is not `Sync`, so it is held behind a
//! [`std::sync::Mutex`] inside an [`std::sync::Arc`]. Each async trait method
//! moves a clone of that `Arc` into [`tokio::task::spawn_blocking`], so the
//! synchronous sqlite work runs on Tokio's blocking pool rather than on an
//! async worker thread. The connection lock is acquired and released entirely
//! inside that blocking closure, which keeps the read-head-then-insert
//! single-writer invariant the append-only chain requires (the lock makes
//! each append atomic per scope) without ever stalling an async worker.

use std::path::Path;
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use rusqlite::{params, Connection, OptionalExtension};

use crate::chain::{ChainEntry, ProvenanceChain, ProvenanceEvent};
use crate::error::ProvenanceError;
use crate::repository::ProvenanceRepository;

/// Sentinel path that opens an ephemeral, per-connection in-memory database.
const IN_MEMORY_PATH: &str = ":memory:";

/// DDL for the single backing table. `recorded_at` is stored as the entry's
/// canonical RFC-3339 (ms-precision, `Z`) form so range-by-time compares are
/// index-friendly lexicographic string compares.
const SCHEMA_SQL: &str = "\
    CREATE TABLE IF NOT EXISTS provenance_chain (\
        scope        TEXT    NOT NULL,\
        sequence     INTEGER NOT NULL,\
        entry_json   TEXT    NOT NULL,\
        recorded_at  TEXT    NOT NULL,\
        PRIMARY KEY (scope, sequence)\
    );\
    CREATE INDEX IF NOT EXISTS idx_provenance_scope_time \
        ON provenance_chain(scope, recorded_at);";

/// SQLite-backed append-only provenance repository.
///
/// One row per [`ChainEntry`], keyed by `(scope, sequence)`. Every query
/// filters by `scope`, so chains for distinct scopes never interleave.
///
/// The shared `Arc<Mutex<Connection>>` is cloned into each method's
/// [`tokio::task::spawn_blocking`] closure, so the blocking sqlite work
/// never runs on an async worker thread.
pub struct SqliteProvenanceRepository {
    connection: Arc<Mutex<Connection>>,
}

impl SqliteProvenanceRepository {
    /// Open an ephemeral in-memory repository (test + single-process use).
    pub fn open_in_memory() -> Result<Self, ProvenanceError> {
        Self::open(IN_MEMORY_PATH)
    }

    /// Open (creating if absent) a sqlite database at `path` and ensure the
    /// schema exists.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, ProvenanceError> {
        let connection = Connection::open(path).map_err(repository_error)?;
        connection
            .execute_batch(SCHEMA_SQL)
            .map_err(repository_error)?;
        Ok(Self {
            connection: Arc::new(Mutex::new(connection)),
        })
    }

    /// Mint the next entry for `scope` and persist it atomically under the
    /// connection lock. Reads the stored head, reseeds a one-entry
    /// [`ProvenanceChain`], and delegates minting to [`ProvenanceChain::append`].
    fn append_locked(
        connection: &Connection,
        scope: &str,
        event: ProvenanceEvent,
    ) -> Result<ChainEntry, ProvenanceError> {
        let mut chain = ProvenanceChain::new(scope.to_string());
        if let Some(head) = read_head(connection, scope)? {
            chain.entries.push(head);
        }
        let entry = chain.append(event);
        insert_entry(connection, &entry)?;
        Ok(entry)
    }
}

/// Guard against a poisoned lock without panicking on the audit path.
fn lock_connection(
    connection: &Mutex<Connection>,
) -> Result<std::sync::MutexGuard<'_, Connection>, ProvenanceError> {
    connection
        .lock()
        .map_err(|_| ProvenanceError::Repository("connection mutex poisoned".to_string()))
}

#[async_trait]
impl ProvenanceRepository for SqliteProvenanceRepository {
    async fn append(
        &self,
        scope: &str,
        event: ProvenanceEvent,
    ) -> Result<ChainEntry, ProvenanceError> {
        let connection = self.connection.clone();
        let scope = scope.to_string();
        tokio::task::spawn_blocking(move || {
            let connection = lock_connection(&connection)?;
            SqliteProvenanceRepository::append_locked(&connection, &scope, event)
        })
        .await
        .map_err(join_error)?
    }

    async fn head(&self, scope: &str) -> Result<Option<ChainEntry>, ProvenanceError> {
        let connection = self.connection.clone();
        let scope = scope.to_string();
        tokio::task::spawn_blocking(move || {
            let connection = lock_connection(&connection)?;
            read_head(&connection, &scope)
        })
        .await
        .map_err(join_error)?
    }

    async fn list_range(
        &self,
        scope: &str,
        from_sequence: u64,
        to_sequence: u64,
    ) -> Result<Vec<ChainEntry>, ProvenanceError> {
        let connection = self.connection.clone();
        let scope = scope.to_string();
        tokio::task::spawn_blocking(move || {
            let connection = lock_connection(&connection)?;
            read_range_by_sequence(&connection, &scope, from_sequence, to_sequence)
        })
        .await
        .map_err(join_error)?
    }

    async fn list_by_time(
        &self,
        scope: &str,
        from: DateTime<Utc>,
        to: DateTime<Utc>,
    ) -> Result<Vec<ChainEntry>, ProvenanceError> {
        let connection = self.connection.clone();
        let scope = scope.to_string();
        tokio::task::spawn_blocking(move || {
            let connection = lock_connection(&connection)?;
            read_range_by_time(&connection, &scope, from, to)
        })
        .await
        .map_err(join_error)?
    }
}

/// Entries for `scope` whose sequence falls in `[from_sequence, to_sequence]`,
/// ordered by sequence.
fn read_range_by_sequence(
    connection: &Connection,
    scope: &str,
    from_sequence: u64,
    to_sequence: u64,
) -> Result<Vec<ChainEntry>, ProvenanceError> {
    let from = clamp_sequence(from_sequence);
    let to = clamp_sequence(to_sequence);
    let mut statement = connection
        .prepare(
            "SELECT entry_json FROM provenance_chain \
             WHERE scope = ?1 AND sequence >= ?2 AND sequence <= ?3 \
             ORDER BY sequence",
        )
        .map_err(repository_error)?;
    let rows = statement
        .query_map(params![scope, from, to], deserialize_row)
        .map_err(repository_error)?;
    collect_entries(rows)
}

/// Entries for `scope` whose `recorded_at` falls in the `[from, to]` window,
/// ordered by sequence. Compares canonical RFC-3339 strings so the
/// `(scope, recorded_at)` index serves the range.
fn read_range_by_time(
    connection: &Connection,
    scope: &str,
    from: DateTime<Utc>,
    to: DateTime<Utc>,
) -> Result<Vec<ChainEntry>, ProvenanceError> {
    let mut statement = connection
        .prepare(
            "SELECT entry_json FROM provenance_chain \
             WHERE scope = ?1 AND recorded_at >= ?2 AND recorded_at <= ?3 \
             ORDER BY sequence",
        )
        .map_err(repository_error)?;
    let from_canonical = ChainEntry::canonical_timestamp(from);
    let to_canonical = ChainEntry::canonical_timestamp(to);
    let rows = statement
        .query_map(
            params![scope, from_canonical, to_canonical],
            deserialize_row,
        )
        .map_err(repository_error)?;
    collect_entries(rows)
}

/// Highest-sequence entry for `scope`, or `None` when the chain is empty.
fn read_head(connection: &Connection, scope: &str) -> Result<Option<ChainEntry>, ProvenanceError> {
    connection
        .query_row(
            "SELECT entry_json FROM provenance_chain \
             WHERE scope = ?1 ORDER BY sequence DESC LIMIT 1",
            params![scope],
            deserialize_row,
        )
        .optional()
        .map_err(repository_error)?
        .transpose()
}

/// Persist a freshly-minted entry. The `(scope, sequence)` primary key makes
/// a duplicate sequence a hard error rather than a silent overwrite, so a
/// concurrent double-append surfaces instead of corrupting the chain.
fn insert_entry(connection: &Connection, entry: &ChainEntry) -> Result<(), ProvenanceError> {
    let entry_json = serde_json::to_string(entry).map_err(serialize_error)?;
    let sequence = clamp_sequence(entry.sequence);
    let recorded_at = ChainEntry::canonical_timestamp(entry.recorded_at);
    connection
        .execute(
            "INSERT INTO provenance_chain (scope, sequence, entry_json, recorded_at) \
             VALUES (?1, ?2, ?3, ?4)",
            params![entry.scope, sequence, entry_json, recorded_at],
        )
        .map_err(repository_error)?;
    Ok(())
}

/// `sequence` is a `u64` in the domain but sqlite integers are `i64`.
/// Clamp rather than wrap so an out-of-range value cannot silently alias a
/// different row (see the `neo4rs i64 wrap` lesson).
fn clamp_sequence(sequence: u64) -> i64 {
    i64::try_from(sequence).unwrap_or(i64::MAX)
}

/// Row mapper: a single `entry_json` column deserialised back to a
/// [`ChainEntry`]. Deserialisation failure is folded into the row's
/// `rusqlite::Result` so the caller's error path stays uniform.
fn deserialize_row(
    row: &rusqlite::Row<'_>,
) -> rusqlite::Result<Result<ChainEntry, ProvenanceError>> {
    let entry_json: String = row.get(0)?;
    Ok(serde_json::from_str(&entry_json).map_err(deserialize_error))
}

/// Drain a query's rows into a `Vec`, surfacing the first sqlite or
/// deserialisation error.
fn collect_entries(
    rows: impl Iterator<Item = rusqlite::Result<Result<ChainEntry, ProvenanceError>>>,
) -> Result<Vec<ChainEntry>, ProvenanceError> {
    let mut entries = Vec::new();
    for row in rows {
        entries.push(row.map_err(repository_error)??);
    }
    Ok(entries)
}

fn repository_error(error: rusqlite::Error) -> ProvenanceError {
    ProvenanceError::Repository(format!("sqlite: {error}"))
}

fn serialize_error(error: serde_json::Error) -> ProvenanceError {
    ProvenanceError::Repository(format!("serialise chain entry: {error}"))
}

fn deserialize_error(error: serde_json::Error) -> ProvenanceError {
    ProvenanceError::Repository(format!("deserialise chain entry: {error}"))
}

/// A `spawn_blocking` task failed to join (panicked or was cancelled). Surface
/// it as a repository error rather than re-panicking on the async path.
fn join_error(error: tokio::task::JoinError) -> ProvenanceError {
    ProvenanceError::Repository(format!("provenance sqlite task join: {error}"))
}

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

    /// Reassemble a contiguous run of persisted entries into a
    /// [`ProvenanceChain`] for end-to-end Merkle verification.
    fn reassemble_chain(scope: &str, entries: Vec<ChainEntry>) -> ProvenanceChain {
        ProvenanceChain {
            scope: scope.to_string(),
            entries,
        }
    }

    fn sample_event() -> ProvenanceEvent {
        ProvenanceEvent {
            kind: ProvenanceEventKind::AgentRunStarted,
            actor: "agent:hello".into(),
            resource_ref: "run-1".into(),
            payload_hash_hex: "ab".repeat(32),
            metadata: serde_json::json!({"agent_name": "hello"}),
        }
    }

    #[tokio::test]
    async fn append_then_head_and_range_roundtrip() {
        let repo = SqliteProvenanceRepository::open_in_memory().unwrap();
        let e1 = repo.append("scope-x", sample_event()).await.unwrap();
        let head = repo.head("scope-x").await.unwrap().unwrap();
        assert_eq!(head.id, e1.id);
        assert_eq!(head.entry_hash_hex, e1.entry_hash_hex);
        let range = repo.list_range("scope-x", 0, u64::MAX).await.unwrap();
        assert_eq!(range.len(), 1);
        assert_eq!(range[0].id, e1.id);
    }

    #[tokio::test]
    async fn second_append_chains_on_first() {
        let repo = SqliteProvenanceRepository::open_in_memory().unwrap();
        let e1 = repo.append("s", sample_event()).await.unwrap();
        let e2 = repo.append("s", sample_event()).await.unwrap();
        assert_ne!(e1.id, e2.id);
        assert_eq!(e2.sequence, e1.sequence + 1);
        assert_eq!(
            e2.previous_hash_hex, e1.entry_hash_hex,
            "second entry must link to the first via previous_hash_hex"
        );
        assert_eq!(repo.head("s").await.unwrap().unwrap().id, e2.id);
    }

    #[tokio::test]
    async fn scopes_are_isolated() {
        let repo = SqliteProvenanceRepository::open_in_memory().unwrap();
        repo.append("a", sample_event()).await.unwrap();
        assert!(repo.head("b").await.unwrap().is_none());
        assert!(repo.list_range("b", 0, u64::MAX).await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn empty_scope_head_is_none() {
        let repo = SqliteProvenanceRepository::open_in_memory().unwrap();
        assert!(repo.head("never-written").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn persisted_chain_passes_merkle_verification() {
        // The key security test: a chain minted through the sqlite repo and
        // read back must validate under the crate's own Merkle rules.
        let repo = SqliteProvenanceRepository::open_in_memory().unwrap();
        for _ in 0..3 {
            repo.append("verify-me", sample_event()).await.unwrap();
        }
        let entries = repo.list_range("verify-me", 0, u64::MAX).await.unwrap();
        assert_eq!(entries.len(), 3);
        let chain = reassemble_chain("verify-me", entries);
        chain
            .verify()
            .expect("persisted sqlite chain must pass Merkle verification");
        assert_eq!(chain.entries[0].sequence, 0);
        assert_eq!(chain.entries[2].sequence, 2);
        assert_eq!(chain.entries[2].chain_version, 2);
    }

    #[tokio::test]
    async fn list_by_time_filters_to_window() {
        let repo = SqliteProvenanceRepository::open_in_memory().unwrap();
        let entry = repo.append("timed", sample_event()).await.unwrap();
        let before = entry.recorded_at - chrono::Duration::seconds(1);
        let after = entry.recorded_at + chrono::Duration::seconds(1);

        let inside = repo.list_by_time("timed", before, after).await.unwrap();
        assert_eq!(inside.len(), 1);
        assert_eq!(inside[0].id, entry.id);

        let future_only = repo
            .list_by_time("timed", after, after + chrono::Duration::seconds(1))
            .await
            .unwrap();
        assert!(
            future_only.is_empty(),
            "entries after the time window must be excluded"
        );

        let past_only = repo
            .list_by_time("timed", before - chrono::Duration::seconds(1), before)
            .await
            .unwrap();
        assert!(
            past_only.is_empty(),
            "entries outside the past window must be excluded"
        );
    }

    #[test]
    fn clamp_sequence_saturates_at_i64_max() {
        assert_eq!(clamp_sequence(0), 0);
        assert_eq!(clamp_sequence(42), 42);
        assert_eq!(clamp_sequence(u64::MAX), i64::MAX);
    }

    #[tokio::test]
    async fn open_rejects_unwritable_path() {
        // A directory path is not a valid sqlite file target — the open must
        // surface a typed Repository error rather than panic.
        let result = SqliteProvenanceRepository::open("/this/path/does/not/exist/db.sqlite");
        match result {
            Err(ProvenanceError::Repository(_)) => {}
            Err(other) => panic!("expected Repository error, got {other:?}"),
            Ok(_) => panic!("expected open to fail on an unwritable path"),
        }
    }
}