pond-db 0.19.4

//! Shared read-only SQLite plumbing for DB-backed adapters (opencode, openclaw,
//! hermes, nanoclaw, agy, devin, pi-coding-agent).
//!
//! Seam rule (CLAUDE.md "Seam boundaries"): this module carries only
//! cross-implementation infrastructure with two real callers and no
//! adapter-specific assumption. The adapter's own `NAME` threads through as a
//! `&'static str` purely for error attribution; nothing here knows a source
//! layout, table name, or record shape.

use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::path::{Path, PathBuf};

use base64::Engine as _;
use rusqlite::{Connection, OpenFlags, OptionalExtension, types::ValueRef};
use serde_json::{Value, json};

use super::AdapterError;

/// Event-channel bound; doubles as backpressure - the blocking reader parks on
/// `blocking_send` when the consumer lags.
pub(crate) const CHANNEL_CAP: usize = 256;

/// Read-side SQLite busy timeout - source DBs are written live (WAL) by their
/// owning process, so a short read burst must not stall on a checkpoint.
pub(crate) const DB_BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(5000);

/// Open a read-only connection (`READ_ONLY | URI`) with the shared busy timeout.
pub(crate) fn open_db(adapter: &'static str, path: &Path) -> Result<Connection, AdapterError> {
    let conn = Connection::open_with_flags(
        path,
        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI,
    )
    .map_err(|error| db_error(adapter, path, "open", &error))?;
    conn.busy_timeout(DB_BUSY_TIMEOUT)
        .map_err(|error| db_error(adapter, path, "set busy_timeout", &error))?;
    Ok(conn)
}

/// Get-or-open a cached connection for `path`.
pub(crate) fn connection<'a>(
    adapter: &'static str,
    conns: &'a mut HashMap<PathBuf, Connection>,
    path: &Path,
) -> Result<&'a Connection, AdapterError> {
    match conns.entry(path.to_path_buf()) {
        Entry::Occupied(entry) => Ok(entry.into_mut()),
        Entry::Vacant(entry) => Ok(entry.insert(open_db(adapter, path)?)),
    }
}

/// A DB-interaction failure (open, prepare, query, row read) classified as
/// [`AdapterError::io`] - these are substrate faults, not source-shape faults
/// (genuine shape errors route through each adapter's own parse errors).
pub(crate) fn db_error(
    adapter: &'static str,
    path: &Path,
    op: &str,
    error: &rusqlite::Error,
) -> AdapterError {
    AdapterError::io(
        adapter,
        path.display().to_string(),
        std::io::Error::other(format!("sqlite {op} failed: {error}")),
    )
}

/// A blocking-task panic is a pond bug, not bad source data, so it fails the
/// whole run rather than skipping a session.
pub(crate) fn join_error(adapter: &'static str, join: tokio::task::JoinError) -> AdapterError {
    AdapterError::io(
        adapter,
        "blocking read task",
        std::io::Error::other(join.to_string()),
    )
}

/// Does this table exist? Probe `sqlite_master` before preparing a SELECT
/// against an optional table, so its absence is a clean control-flow signal
/// rather than a swallowed prepare error.
///
/// Shared because the harnesses pond reads reshape their schemas between
/// releases, so "which tables are present" is load-bearing for more than one
/// adapter: openclaw tells three storage eras apart by the table SET, and agy
/// checks for its `steps` table before peeking a conversation's watermark
/// (absent means empty, unreadable means opaque). It lives here for the same
/// reason [`open_db`] and [`row_to_json`] do - one behaviour, one place,
/// rather than each adapter carrying a copy that drifts.
///
/// `LIMIT 1` rather than `count(*)`: both scan `sqlite_master` (no index on
/// `name`, so the query plans read the same), but the limit emits a counter
/// and halts on the first match while `count(*)` always runs the scan out.
/// This asks whether a row exists, not how many.
pub(crate) fn has_table(conn: &Connection, table: &str) -> rusqlite::Result<bool> {
    conn.query_row(
        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1 LIMIT 1",
        [table],
        |_| Ok(()),
    )
    .optional()
    .map(|row| row.is_some())
}

/// SQLite affinity of each column in a mirrored SELECT list, so a row decodes to
/// the right JSON scalar. `Real` covers float columns (hermes's epoch-second
/// timestamps); openclaw's columns are all `Str`/`Int`.
#[derive(Clone, Copy)]
pub(crate) enum ColKind {
    Str,
    Int,
    Real,
}

/// The column names of a `(name, kind)` list, in order, comma-joined for a SELECT.
pub(crate) fn columns_sql(columns: &[(&str, ColKind)]) -> String {
    columns
        .iter()
        .map(|(name, _)| *name)
        .collect::<Vec<_>>()
        .join(", ")
}

/// Rebuild a row as a JSON object, column names kept verbatim, null columns
/// omitted (spec.md#model-lossless-projection - every non-null column
/// recoverable). One `(name, kind)` list drives both the SELECT and this decode,
/// so the projection can never drift from the query.
pub(crate) fn row_to_json(
    row: &rusqlite::Row,
    columns: &[(&str, ColKind)],
) -> rusqlite::Result<Value> {
    let mut map = serde_json::Map::new();
    for (idx, (name, kind)) in columns.iter().enumerate() {
        match kind {
            ColKind::Str => {
                if let Some(value) = row.get::<_, Option<String>>(idx)? {
                    map.insert((*name).to_owned(), json!(value));
                }
            }
            ColKind::Int => {
                if let Some(value) = row.get::<_, Option<i64>>(idx)? {
                    map.insert((*name).to_owned(), json!(value));
                }
            }
            ColKind::Real => {
                if let Some(value) = row.get::<_, Option<f64>>(idx)? {
                    map.insert((*name).to_owned(), json!(value));
                }
            }
        }
    }
    Ok(Value::Object(map))
}

/// One cell as lossless JSON: a blob (or TEXT that is not UTF-8) is base64
/// under a key naming the encoding, so it can never read back as text. Shared
/// by the adapters that mirror whole rows (`SELECT *`), agy and devin.
pub(crate) fn value_json(value: ValueRef<'_>) -> Value {
    let base64 = |bytes: &[u8]| base64::engine::general_purpose::STANDARD.encode(bytes);
    match value {
        ValueRef::Null => Value::Null,
        ValueRef::Integer(value) => json!(value),
        ValueRef::Real(value) => json!(value),
        ValueRef::Text(bytes) => match std::str::from_utf8(bytes) {
            Ok(text) => json!(text),
            Err(_) => json!({ "text_base64": base64(bytes) }),
        },
        ValueRef::Blob(bytes) => json!({ "base64": base64(bytes) }),
    }
}

/// Send one yield through a blocking read channel, returning `false` from the
/// caller (stop reading this source) when the consumer dropped the receiver.
macro_rules! emit {
    ($tx:expr, $item:expr) => {
        if $tx.blocking_send($item).is_err() {
            return false;
        }
    };
}
pub(crate) use emit;