acorn/io/database/

mod.rs

//! Database utilities for ACORN.
//!
//! This module provides database connection helpers guarded by a process-global
//! mutex so concurrent operations do not race while still respecting per-handle
//! database paths.
use crate::io::api::DatabasePersistence;
use crate::io::ApiResult;
use crate::prelude::{create_dir_all, env, Mutex, OnceLock, PathBuf};
use crate::util::constants::app::{APPLICATION, DEFAULT_CACHE_TTL_SECONDS, ORGANIZATION, QUALIFIER};
use crate::util::constants::env::{CACHE_TTL, DATABASE_PATH};
use crate::util::{to_rfc3339, Label};
use async_trait::async_trait;
use backend::{params, BackendRow, Connection, Params};
use color_eyre::eyre::eyre;
use core::marker::PhantomData;
use directories::ProjectDirs;
use jiff::{SignedDuration, Timestamp};
use tracing::{debug, info};

pub mod backend;
mod candidate;
pub(crate) mod macros;
pub mod schema;

pub(crate) use candidate::Provenance;
pub use candidate::{CandidateAction, CandidatePersistence, ResearchActivityCandidate};

use schema::Table;

static CONNECTION_LOCK: OnceLock<Mutex<()>> = OnceLock::new();

/// SQL parameters produced for dynamic queries.
pub type QueryParams = backend::ParamsFromIter<Vec<Box<dyn backend::ToSql>>>;
/// SQL query string paired with its bound parameters.
pub type SelectQuery = (String, QueryParams);
/// Result of a persist operation indicating whether data was downloaded or already existed.
#[derive(Clone, Debug)]
pub enum PersistStatus {
    /// Data was downloaded and persisted (contains row count).
    Downloaded(usize),
    /// Table already contained data, no action taken.
    AlreadyExists,
}
/// Write and access data in a database
#[async_trait]
pub trait Operations {
    /// Insert a value (row) into the database
    fn insert<R>(&self, value: R) -> ApiResult<usize>
    where
        R: Row;
    /// Save data to database
    async fn persist<P>(&self, data: Option<P>) -> ApiResult<usize>
    where
        P: DatabasePersistence + Send;
    /// Download and persist data for a table if the table is empty.
    /// Returns `PersistStatus::Downloaded(n)` if data was downloaded and persisted,
    /// or `PersistStatus::AlreadyExists` if the table already contained data.
    async fn populate(&self, table: impl TableSchemaProvider + Send) -> ApiResult<PersistStatus>;
    /// Select a value from the database
    fn select<T, P>(&self, value: P) -> ApiResult<Option<T>>
    where
        T: Into<Table>,
        P: Params;
}
/// Row metadata for database table mapping
pub trait Row: RowMetadata {
    /// Returns the table associated with this row type.
    fn table(&self) -> Table;
    /// Returns row field names generated by the `DatabaseRow` derive.
    fn fields() -> &'static [&'static str]
    where
        Self: Sized,
    {
        <Self as RowMetadata>::fields()
    }
    /// Inserts this row into its corresponding table.
    fn insert(self, _conn: &Connection) -> ApiResult<usize>
    where
        Self: Sized;
    /// Selects rows from the database matching this row's field values.
    fn select_all(&self, path: Option<PathBuf>) -> ApiResult<Vec<Self>>
    where
        Self: Sized,
        for<'row> Self: From<&'row BackendRow<'row>>,
    {
        let table = self.table();
        let name = table.name();
        let fields = <Self as RowMetadata>::fields().join(", ");
        let query = format!("SELECT {fields} FROM {name}");
        let (query, params) = self.build_select_query(&query);
        table.rows(&query, params, path.as_ref())
    }
    /// Selects the first row matching this row's filter values and predicate.
    fn select<F>(&self, path: Option<PathBuf>, predicate: F) -> ApiResult<Option<Self>>
    where
        Self: Sized,
        for<'row> Self: From<&'row BackendRow<'row>>,
        F: Fn(&Self) -> bool,
    {
        self.select_all(path).map(|rows| rows.into_iter().find(predicate))
    }
    /// Builds a SQL SELECT query with optional WHERE/ORDER BY clauses.
    fn build_select_query(&self, base: &str) -> SelectQuery;
    /// Get ID for next row (used when inserting new row)
    fn next_row_id(&self, conn: &Connection) -> ApiResult<i64>
    where
        Self: Sized,
    {
        let table = self.table().name();
        conn.query_row(&format!("SELECT COALESCE(MAX(id), 0) + 1 FROM {table}"), params![], |row| row.get(0))
            .map_err(|why| eyre!("=> {} Failed to determine next id for {table} — {why}", Label::fail()))
    }
}
/// Metadata generated by `DatabaseRow` derive for row structs.
pub trait RowMetadata {
    /// Returns the list of field names for the row.
    fn fields() -> &'static [&'static str]
    where
        Self: Sized;
}
/// Table schema provider for database operations
#[async_trait]
pub trait TableSchemaProvider: Copy + Into<Table> + 'static {
    /// Returns all variants for the table type.
    fn all() -> &'static [Self];
    /// Returns the CREATE TABLE SQL statement for this table
    fn create_statement(&self) -> String;
    /// Returns the table name as a string
    fn name(&self) -> &'static str;
    /// Populate the table with data (typically downloaded use API module)
    async fn populate(&self, path: Option<PathBuf>) -> ApiResult<usize>;
    /// Print all table rows
    fn print(self, path: Option<PathBuf>);
    /// Get table rows
    fn rows<R, P>(&self, query: &str, params: P, path: Option<&PathBuf>) -> ApiResult<Vec<R>>
    where
        P: Params,
        for<'row> R: Row + From<&'row BackendRow<'row>>;
}
/// Database handle for a given table schema provider type
#[derive(Clone, Debug)]
pub struct Database<T>
where
    T: TableSchemaProvider,
{
    path: Option<PathBuf>,
    marker: PhantomData<T>,
}
impl<T> Default for Database<T>
where
    T: TableSchemaProvider,
{
    fn default() -> Self {
        Self {
            path: None,
            marker: PhantomData,
        }
    }
}
#[async_trait]
impl<S> Operations for Database<S>
where
    S: TableSchemaProvider + Sync,
{
    fn insert<R>(&self, value: R) -> ApiResult<usize>
    where
        R: Row,
    {
        let row: R = value;
        self.with_connection(|conn| row.insert(conn))
    }
    async fn persist<P>(&self, data: Option<P>) -> ApiResult<usize>
    where
        P: DatabasePersistence + Send,
    {
        let database = Database::<Table>::from_path(self.path.clone());
        match data {
            | Some(rows) => rows.persist(database).await,
            | None => Err(eyre!("Failed to persist data")),
        }
    }
    async fn populate(&self, table: impl TableSchemaProvider + Send) -> ApiResult<PersistStatus> {
        match self.row_count(table) {
            | Ok(count) if count > 0 => Ok(PersistStatus::AlreadyExists),
            | Ok(_) => table.into().populate(self.path.clone()).await.map(PersistStatus::Downloaded),
            | Err(why) => Err(why),
        }
    }
    fn select<T, P>(&self, _value: P) -> ApiResult<Option<T>>
    where
        T: Into<Table>,
        P: Params,
    {
        Err(eyre!("Database query is not implemented"))
    }
}
impl<T> Database<T>
where
    T: TableSchemaProvider,
{
    /// Create a new database handle from an optional database path override.
    pub fn from_path(path: Option<PathBuf>) -> Self {
        Self { path, marker: PhantomData }
    }
    /// Executes a callback using the database path configured on this handle.
    pub fn with_connection<U>(&self, callback: impl FnOnce(&Connection) -> ApiResult<U>) -> ApiResult<U> {
        CONNECTION_LOCK
            .get_or_init(|| Mutex::new(()))
            .lock()
            .map_err(|why| eyre!("Failed to acquire database connection lock — {why}"))
            .and_then(|guard| {
                let result = resolve_database_path(self.path.as_ref())
                    .and_then(|db_path| Connection::open(db_path).map_err(|why| eyre!("Failed to open database connection — {why}")))
                    .and_then(|connection| callback(&connection));
                drop(guard);
                result
            })
    }
    /// Get the cache TTL value from environment or use default.
    pub fn cache_ttl(&self) -> SignedDuration {
        #[cfg(feature = "std")]
        {
            if let Ok(env_ttl) = dotenvy::var(CACHE_TTL) {
                if let Ok(seconds) = env_ttl.parse::<i64>() {
                    if seconds > 0 {
                        return SignedDuration::from_secs(seconds);
                    }
                }
            }
        }
        let default_ttl = i64::try_from(DEFAULT_CACHE_TTL_SECONDS).map_or(SignedDuration::MAX, SignedDuration::from_secs);
        if let Ok(env_ttl) = env::var(CACHE_TTL) {
            if let Ok(seconds) = env_ttl.parse::<i64>() {
                if seconds > 0 {
                    return SignedDuration::from_secs(seconds);
                }
            }
        }
        default_ttl
    }
    /// Create one table if it does not exist.
    pub fn migrate_table(&self, table: impl Into<Table>) -> ApiResult<()> {
        let table = table.into();
        self.with_connection(|conn| {
            let statement = table.create_statement();
            conn.execute(&statement, params![])
                .map_err(|why| eyre!("Failed to create table {} — {why}", table.name()))
                .map(|_| ())
        })
    }
    /// Create all tables defined by the variant provider type.
    pub fn migrate(&self) -> ApiResult<()> {
        T::all().iter().copied().try_fold((), |_, table| self.migrate_table(table)).inspect(|_| {
            debug!("{} Database migration", Label::pass());
        })
    }
    /// Delete all rows from a table.
    pub fn clear(&self, table: impl Into<Table>) -> ApiResult<usize> {
        let table = table.into();
        self.with_connection(|conn| {
            conn.execute(&format!("DELETE FROM {}", table.name()), params![]).or_else(|error| {
                if error.to_string().contains("no such table") {
                    Ok(0)
                } else {
                    Err(eyre!("Failed to clear {} — {error}", table.name()))
                }
            })
        })
    }
    /// Delete rows where a timestamp/text column is older than a cutoff value.
    pub fn delete_expired_before(&self, table: impl Into<Table>, column: &str, cutoff: &str) -> ApiResult<usize> {
        let table = table.into();
        let sql = format!("DELETE FROM {} WHERE {} < ?", table.name(), column);
        self.with_connection(|conn| {
            conn.execute(&sql, params![cutoff]).or_else(|error| {
                if error.to_string().contains("no such table") {
                    Ok(0)
                } else {
                    Err(eyre!("Failed to clean up {} — {error}", table.name()))
                }
            })
        })
    }
    /// Count rows in a table.
    pub fn row_count(&self, table: impl Into<Table>) -> ApiResult<usize> {
        let name = table.into().name();
        self.with_connection(|conn| {
            conn.query_row(&format!("SELECT COUNT(*) FROM {}", name), params![], |row| row.get::<_, i64>(0))
                .map_err(|why| eyre!("Failed to count rows in {} — {why}", name))
                .and_then(|count| usize::try_from(count).map_err(|why| eyre!("Failed to convert row count for {} — {why}", name)))
        })
    }
    /// Keep only the most recent `limit` rows using an id/order column pair.
    pub fn trim_to_recent(&self, table: impl Into<Table>, id_column: &str, order_column: &str, limit: usize) -> ApiResult<usize> {
        let table = table.into();
        let sql = format!(
            "DELETE FROM {table} WHERE {id_col} NOT IN (SELECT {id_col} FROM {table} ORDER BY {order_col} DESC LIMIT ?)",
            table = table.name(),
            id_col = id_column,
            order_col = order_column,
        );
        self.with_connection(|conn| {
            conn.execute(&sql, params![limit as i64])
                .map_err(|why| eyre!("Failed to trim {} — {why}", table.name()))
        })
    }
    /// Clean up expired link cache entries and trim validation history.
    pub fn cleanup_expired_cache(&self) -> ApiResult<usize> {
        let ttl = self.cache_ttl();
        let now = Timestamp::now();
        let cutoff = to_rfc3339(now.checked_sub(ttl).unwrap_or(now));
        self.delete_expired_before(Table::LinkCache, "expires_at", &cutoff)
            .and_then(|link_count| {
                self.trim_to_recent(Table::ValidationHistory, "id", "checked_at", 1000).map(|_| {
                    if link_count > 0 {
                        info!("=> {} Cleaned up {link_count} expired cache entries", Label::run());
                    }
                    link_count
                })
            })
    }
    /// Clear all cache-related tables.
    pub fn clear_cache(&self) -> ApiResult<usize> {
        self.clear(Table::LinkCache).and_then(|link_count| {
            self.clear(Table::ValidationHistory).map(|validation_count| {
                let total = link_count.saturating_add(validation_count);
                info!("=> {} Cleared {total} entries from cache tables", Label::pass());
                total
            })
        })
    }
    /// Clear (reset) all data from all tables.
    pub fn reset(&self) -> ApiResult<usize> {
        T::all()
            .iter()
            .copied()
            .try_fold(0usize, |total, table| self.clear(table).map(|count| total.saturating_add(count)))
            .inspect(|&total| {
                info!("=> {} Cleared {total} entries from all tables", Label::pass());
            })
    }
}
fn database_name() -> &'static str {
    #[cfg(feature = "duckdb")]
    {
        "acorn.duckdb"
    }
    #[cfg(not(feature = "duckdb"))]
    {
        "acorn.db"
    }
}
/// Resolve the effective database path from an explicit argument, environment variable, or project defaults.
pub fn resolve_database_path(path: Option<&PathBuf>) -> ApiResult<PathBuf> {
    let path = path
        .cloned()
        .or_else(|| env::var(DATABASE_PATH).ok().filter(|value| !value.trim().is_empty()).map(PathBuf::from));
    match path {
        | Some(path) => Ok(path),
        | None => ProjectDirs::from(QUALIFIER, ORGANIZATION, APPLICATION)
            .ok_or_else(|| eyre!("Failed to determine project directories"))
            .map(|directories| directories.cache_dir().join("database").join(database_name())),
    }
    .and_then(|path| {
        path.parent()
            .map(PathBuf::from)
            .ok_or_else(|| eyre!("Failed to determine database parent directory"))
            .and_then(|parent| {
                create_dir_all(&parent)
                    .map_err(|why| eyre!("Failed to create database directory: {why}"))
                    .map(|_| path)
            })
    })
}

#[cfg(test)]
mod tests;