hanzo-database 1.1.13

//! # Hanzo DB - Multi-Backend Database Abstraction
//!
//! Production-ready database abstraction for Hanzo Node, supporting:
//! - LanceDB for vector search and multimodal storage
//! - DuckDB for analytics and OLAP queries
//! - PostgreSQL for relational data
//! - Redis for caching
//! - SQLite for lightweight deployments
//!
//! Features:
//! - Unified interface across all backends
//! - Automatic backend selection based on workload
//! - Connection pooling and transaction support
//! - Migration between backends

use anyhow::{Context, Result};
use async_trait::async_trait;
use log::{debug, error, info, warn};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::RwLock;
use thiserror::Error;

/// Database error types
#[derive(Debug, Error)]
pub enum HanzoDbError {
    #[error("Connection error: {0}")]
    ConnectionError(String),

    #[error("Query error: {0}")]
    QueryError(String),

    #[error("Schema error: {0}")]
    SchemaError(String),

    #[error("Invalid schema: {0}")]
    InvalidSchema(String),

    #[error("Invalid data: {0}")]
    InvalidData(String),

    #[error("Transaction error: {0}")]
    TransactionError(String),

    #[error("Migration error: {0}")]
    MigrationError(String),

    #[error("Not implemented: {0}")]
    NotImplemented(String),
}

pub mod models;

#[cfg(feature = "backend-lancedb")]
pub mod vector_search;

// Migration module (only with migration feature and SQLite backend)
#[cfg(all(feature = "migration", feature = "backend-sqlite"))]
pub mod migration;

// Backend modules
pub mod backends;

// Re-exports
pub use models::*;

#[cfg(feature = "backend-lancedb")]
pub use vector_search::*;

/// Database backend type
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DatabaseBackend {
    /// LanceDB for vector operations and multimodal data
    LanceDB,
    /// DuckDB for analytics and OLAP
    DuckDB,
    /// PostgreSQL for relational data
    PostgreSQL,
    /// Redis for caching
    Redis,
    /// SQLite for lightweight deployments
    SQLite,
}

impl DatabaseBackend {
    /// Select optimal backend for workload type
    pub fn for_workload(workload: WorkloadType) -> Self {
        match workload {
            WorkloadType::VectorSearch => Self::LanceDB,
            WorkloadType::Analytics => Self::DuckDB,
            WorkloadType::Transactional => Self::PostgreSQL,
            WorkloadType::Cache => Self::Redis,
            WorkloadType::Embedded => Self::SQLite,
        }
    }
}

/// Workload type for backend selection
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WorkloadType {
    /// Vector similarity search
    VectorSearch,
    /// Analytical queries (OLAP)
    Analytics,
    /// Transactional operations (OLTP)
    Transactional,
    /// High-speed caching
    Cache,
    /// Embedded/lightweight operations
    Embedded,
}

/// Unified database configuration
#[derive(Debug, Clone)]
pub struct HanzoDbConfig {
    /// Selected backend
    pub backend: DatabaseBackend,
    /// Database path (for file-based backends)
    pub path: Option<PathBuf>,
    /// Connection URL (for network backends)
    pub url: Option<String>,
    /// Connection pool size
    pub pool_size: usize,
    /// Enable write-ahead logging
    pub enable_wal: bool,
    /// Cache size in bytes
    pub cache_size: Option<usize>,
    /// Enable compression
    pub enable_compression: bool,
}

impl Default for HanzoDbConfig {
    fn default() -> Self {
        Self {
            backend: DatabaseBackend::LanceDB,
            path: Some(PathBuf::from("./storage/hanzo-db")),
            url: None,
            pool_size: 16,
            enable_wal: true,
            cache_size: Some(64 * 1024 * 1024), // 64MB
            enable_compression: true,
        }
    }
}

/// Unified database trait
#[async_trait]
pub trait HanzoDatabase: Send + Sync {
    /// Initialize the database
    async fn init(&self) -> Result<()>;
    
    /// Create a table
    async fn create_table(&self, name: &str, schema: TableSchema) -> Result<()>;
    
    /// Insert data
    async fn insert(&self, table: &str, data: &[Record]) -> Result<()>;
    
    /// Query data
    async fn query(&self, query: Query) -> Result<QueryResult>;
    
    /// Vector search
    async fn vector_search(&self, query: VectorQuery) -> Result<Vec<SearchResult>>;
    
    /// Begin transaction
    async fn begin_transaction(&self) -> Result<Transaction>;
    
    /// Optimize database
    async fn optimize(&self) -> Result<()>;
    
    /// Get database statistics
    async fn stats(&self) -> Result<DatabaseStats>;
}

/// Table schema definition
#[derive(Debug, Clone)]
pub struct TableSchema {
    pub columns: Vec<Column>,
    pub indexes: Vec<Index>,
    pub constraints: Vec<Constraint>,
}

/// Column definition
#[derive(Debug, Clone)]
pub struct Column {
    pub name: String,
    pub data_type: DataType,
    pub nullable: bool,
    pub default: Option<Value>,
}

/// Data type enumeration
#[derive(Debug, Clone)]
pub enum DataType {
    // Scalar types
    Boolean,
    Int32,
    Int64,
    Float32,
    Float64,
    String,
    Binary,
    Timestamp,
    
    // Vector types
    Vector(usize), // dimension
    
    // Complex types
    Json,
    Array(Box<DataType>),
    Struct(Vec<(String, DataType)>),
}

/// Index definition
#[derive(Debug, Clone)]
pub struct Index {
    pub name: String,
    pub columns: Vec<String>,
    pub index_type: IndexType,
}

#[derive(Debug, Clone)]
pub enum IndexType {
    BTree,
    Hash,
    IVF_PQ { nlist: usize, nprobe: usize },
    HNSW { max_elements: usize, m: usize },
}

/// Database constraint
#[derive(Debug, Clone)]
pub enum Constraint {
    PrimaryKey(Vec<String>),
    ForeignKey { columns: Vec<String>, references: String },
    Unique(Vec<String>),
    Check(String),
}

/// Query structure
#[derive(Debug, Clone)]
pub struct Query {
    pub table: String,
    pub select: Vec<String>,
    pub filter: Option<Filter>,
    pub order_by: Vec<OrderBy>,
    pub limit: Option<usize>,
    pub offset: Option<usize>,
}

/// Filter expression
#[derive(Debug, Clone)]
pub enum Filter {
    Eq(String, Value),
    Ne(String, Value),
    Gt(String, Value),
    Gte(String, Value),
    Lt(String, Value),
    Lte(String, Value),
    In(String, Vec<Value>),
    Like(String, String),
    And(Box<Filter>, Box<Filter>),
    Or(Box<Filter>, Box<Filter>),
    Not(Box<Filter>),
}

/// Order by clause
#[derive(Debug, Clone)]
pub struct OrderBy {
    pub column: String,
    pub ascending: bool,
}

/// Value type
#[derive(Debug, Clone)]
pub enum Value {
    Null,
    Bool(bool),
    Int32(i32),
    Int64(i64),
    Float32(f32),
    Float64(f64),
    String(String),
    Binary(Vec<u8>),
    Timestamp(i64),
    Vector(Vec<f32>),
    Json(serde_json::Value),
}

/// Record type
#[derive(Debug, Clone)]
pub struct Record {
    pub values: Vec<(String, Value)>,
}

/// Query result
#[derive(Debug)]
pub struct QueryResult {
    pub columns: Vec<String>,
    pub rows: Vec<Record>,
    pub row_count: usize,
}

/// Vector query
#[derive(Debug, Clone)]
pub struct VectorQuery {
    pub table: String,
    pub vector: Vec<f32>,
    pub k: usize,
    pub filter: Option<Filter>,
    pub metric: DistanceMetric,
}

#[derive(Debug, Clone, Copy)]
pub enum DistanceMetric {
    L2,
    Cosine,
    InnerProduct,
}

/// Search result
#[derive(Debug, Clone)]
pub struct SearchResult {
    pub record: Record,
    pub score: f32,
}

/// Transaction handle
pub struct Transaction {
    inner: Arc<RwLock<TransactionInner>>,
}

struct TransactionInner {
    backend: DatabaseBackend,
    // Backend-specific transaction handle
    handle: Box<dyn std::any::Any + Send + Sync>,
}

impl Transaction {
    /// Commit the transaction
    pub async fn commit(self) -> Result<()> {
        let inner = self.inner.write().await;
        // Backend-specific commit logic
        Ok(())
    }
    
    /// Rollback the transaction
    pub async fn rollback(self) -> Result<()> {
        let inner = self.inner.write().await;
        // Backend-specific rollback logic
        Ok(())
    }
}

/// Database statistics
#[derive(Debug, Clone)]
pub struct DatabaseStats {
    pub backend: DatabaseBackend,
    pub table_count: usize,
    pub total_rows: usize,
    pub total_size_bytes: usize,
    pub index_count: usize,
    pub cache_hit_rate: f64,
}

/// Create a Hanzo database instance
pub async fn connect(config: HanzoDbConfig) -> Result<Arc<dyn HanzoDatabase>> {
    match config.backend {
        #[cfg(feature = "backend-lancedb")]
        DatabaseBackend::LanceDB => {
            let db = backends::lancedb::LanceDbBackend::new(config).await?;
            Ok(Arc::new(db))
        }
        #[cfg(not(feature = "backend-lancedb"))]
        DatabaseBackend::LanceDB => {
            anyhow::bail!("LanceDB backend not enabled. Compile with --features backend-lancedb")
        }
        #[cfg(feature = "backend-duckdb")]
        DatabaseBackend::DuckDB => {
            let db = backends::duckdb::DuckDbBackend::new(config).await?;
            Ok(Arc::new(db))
        }
        #[cfg(not(feature = "backend-duckdb"))]
        DatabaseBackend::DuckDB => {
            anyhow::bail!("DuckDB backend not enabled. Compile with --features backend-duckdb")
        }
        #[cfg(feature = "backend-postgres")]
        DatabaseBackend::PostgreSQL => {
            let db = backends::postgres::PostgresBackend::new(config).await?;
            Ok(Arc::new(db))
        }
        #[cfg(not(feature = "backend-postgres"))]
        DatabaseBackend::PostgreSQL => {
            anyhow::bail!("PostgreSQL backend not enabled. Compile with --features backend-postgres")
        }
        #[cfg(feature = "backend-redis")]
        DatabaseBackend::Redis => {
            let db = backends::redis::RedisBackend::new(config).await?;
            Ok(Arc::new(db))
        }
        #[cfg(not(feature = "backend-redis"))]
        DatabaseBackend::Redis => {
            anyhow::bail!("Redis backend not enabled. Compile with --features backend-redis")
        }
        #[cfg(feature = "backend-sqlite")]
        DatabaseBackend::SQLite => {
            let db = backends::sqlite::SqliteBackend::new(config).await?;
            Ok(Arc::new(db))
        }
        #[cfg(not(feature = "backend-sqlite"))]
        DatabaseBackend::SQLite => {
            anyhow::bail!("SQLite backend not enabled. Compile with --features backend-sqlite")
        }
    }
}


#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_backend_selection() {
        assert_eq!(
            DatabaseBackend::for_workload(WorkloadType::VectorSearch),
            DatabaseBackend::LanceDB
        );
        assert_eq!(
            DatabaseBackend::for_workload(WorkloadType::Analytics),
            DatabaseBackend::DuckDB
        );
        assert_eq!(
            DatabaseBackend::for_workload(WorkloadType::Transactional),
            DatabaseBackend::PostgreSQL
        );
    }
}