dameng_rust_sdk 0.1.3

//! Connection management for Dameng Rust SDK

use std::fmt;
use odbc_api::{Connection as OdbcConnection, Environment, Cursor, ResultSetMetadata, ConnectionOptions as OdbcConnectionOptions, handles::StatementImpl};
use log::{debug, warn};

use crate::error::{Error, Result as SdkResult};
use crate::types::{DamengValue};
use crate::query::QueryBuilder;

/// Convert GBK bytes to UTF-8 string
fn gbk_to_utf8(bytes: &[u8]) -> SdkResult<String> {
    use encoding_rs::GBK;
    
    // Try UTF-8 first (fallback for data that's already UTF-8)
    if let Ok(s) = std::str::from_utf8(bytes) {
        return Ok(s.to_string());
    }
    
    // Use GBK decoder from encoding_rs
    let (decoded, _, _) = GBK.decode(bytes);
    
    Ok(decoded.to_string())
}

/// Connection options for Dameng database
#[derive(Debug, Clone)]
pub struct ConnectionOptions {
    /// Database server address
    pub server: String,
    /// Database port
    pub port: u16,
    /// Database username
    pub username: String,
    /// Database password
    pub password: String,
    /// Database schema
    pub schema: String,
    /// Connection timeout in seconds
    pub timeout: u32,
    /// Whether to use TLS
    pub use_tls: bool,
    /// Additional connection parameters
    pub additional_params: Vec<(String, String)>,
    ///DRIVER
    pub driver:String,
}

impl Default for ConnectionOptions {
    fn default() -> Self {
        Self {
            server: "localhost".to_string(),
            port: 5236,
            username: "SYSDBA".to_string(),
            driver:"{DM8 ODBC DRIVER}".to_string(),
            password: "".to_string(),
            schema: "DMHR".to_string(),
            timeout: 30,
            use_tls: false,
            additional_params: Vec::new(),
        }
    }
}

impl fmt::Display for ConnectionOptions {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "ConnectionOptions {{ server: {}, port: {}, username: {}, schema: {}, timeout: {}s, use_tls: {} }}",
            self.server, self.port, self.username, self.schema, self.timeout, self.use_tls
        )
    }
}

/// Transaction handle
pub struct Transaction {
    _inner: (),
}

/// Connection to Dameng database
pub struct Connection {
    _environment: Environment,
    _inner: Option<OdbcConnection<'static>>,
    options: ConnectionOptions,
}
use odbc_api::IntoParameter;
impl Connection {
    /// Create a new connection (returns a placeholder for now)
    pub fn connect() -> SdkResult<Self> {
        Self::with_options(ConnectionOptions::default())
    }
    
    /// Create a new connection with options
    pub fn with_options(options: ConnectionOptions) -> SdkResult<Self> {
        debug!("Creating connection with options: {}", options);
        
        // Build connection string
        let connection_string = build_connection_string(&options);
        debug!("Connection string: {}", mask_password(&connection_string));
        
        // Create a static environment for the connection
        let environment = Environment::new()
            .map_err(|e| Error::connection(format!("Failed to create ODBC environment: {}", e)))?;
        
        // Leak the environment to give it a 'static lifetime
        // This is necessary because ODBC connections must outlive their environment
        let env_ref: &'static Environment = Box::leak(Box::new(environment));
        
        // Establish connection
        let connection = env_ref.connect_with_connection_string(&connection_string, OdbcConnectionOptions::default())
            .map_err(|e| Error::connection(format!("Failed to connect to database: {}", e)))?;
        
        debug!("Successfully connected to database");
        
        // Store a reference to the leaked environment
        // Note: This will leak memory, but it's acceptable for the lifetime of the application
        Ok(Self {
            _environment: Environment::new().unwrap(), // Placeholder
            _inner: Some(connection),
            options,
        })
    }
    
    /// Get connection options
    pub fn options(&self) -> &ConnectionOptions {
        &self.options
    }
    
        /// Execute a SQL query with parameters
    /// 
    /// # Arguments
    /// * `sql` - The SQL query string
    /// * `params` - Parameters implementing the `odbc_api::Parameter` trait (e.g., tuples)
    /// 
    /// # Example
    /// ```ignore
    /// // Using a tuple for parameters
    /// let result = conn.query_with_param("SELECT * FROM city WHERE name = ?", (&"Beijing",))?;
    /// ```
    pub fn query_with_param<P>(&mut self, sql: &str, params: P) -> SdkResult<QueryResult>
    where
    P:  odbc_api::ParameterCollectionRef,
    {
        debug!("Executing parametrized query: {}", sql);
        
        // Execute query with parameters and get cursor
        let cursor = self._inner.as_mut()
            .ok_or_else(|| Error::connection("Connection not established".to_string()))?
            .execute(sql, params)
            .map_err(|e| Error::query(format!("Failed to execute query with params: {}", e)))?
            .ok_or_else(|| Error::query("Query did not return a result set".to_string()))?;
        
        // Create query result and fetch all data immediately
        QueryResult::from_cursor(cursor, sql)
    }
    
    /// Execute a SQL statement with parameters (INSERT, UPDATE, DELETE)
    /// 
    /// # Arguments
    /// * `sql` - The SQL statement string
    /// * `params` - Parameters implementing the `odbc_api::Parameter` trait
    /// 
    /// # Example
    /// ```ignore
    /// conn.execute_with_param("INSERT INTO t1 VALUES (?, ?)", (1, "test"))?;
    /// ```
    pub fn execute_with_param<P>(&mut self, sql: &str, params: P) -> SdkResult<bool>
    where
        P: odbc_api::ParameterCollectionRef,
    {
        debug!("Executing parametrized statement: {}", sql);
        
        let conn = self._inner.as_mut()
            .ok_or_else(|| Error::connection("Connection not established".to_string()))?;
        
        // Execute the statement with parameters
        let result = conn.execute(sql, params);
        
        match result {
            Ok(_) => Ok(true),
            Err(e) => Err(Error::query(format!("Failed to execute statement with params: {}", e))),
        }
    }


    /// Execute a SQL query
    pub fn query(&mut self, sql: &str) -> SdkResult<QueryResult> {
        debug!("Executing query: {}", sql);
        
        // Execute query and get cursor
        let cursor = self._inner.as_mut()
            .ok_or_else(|| Error::connection("Connection not established".to_string()))?
            .execute(sql, ())
            .map_err(|e| Error::query(format!("Failed to execute query: {}", e)))?
            .ok_or_else(|| Error::query("Query did not return a result set".to_string()))?;
        
        // Create query result and fetch all data immediately
        QueryResult::from_cursor(cursor, sql)
    }
    
    /// Execute a SQL statement (INSERT, UPDATE, DELETE)
    pub fn execute(&mut self, sql: &str) -> SdkResult<bool> {
        debug!("Executing statement: {}", sql);
        
        let conn = self._inner.as_mut()
            .ok_or_else(|| Error::connection("Connection not established".to_string()))?;
        
        // Execute the statement
        let cursor = self._inner.as_mut()
        .ok_or_else(|| Error::connection("Connection not established".to_string()))?
        .execute(sql, ());
        Ok(cursor.is_ok())
    }
    
    /// Begin a transaction
    pub fn begin_transaction(&mut self) -> SdkResult<Transaction> {
        debug!("Beginning transaction");
        
        let conn = self._inner.as_mut()
            .ok_or_else(|| Error::connection("Connection not established".to_string()))?;
        
        // Set autocommit to false to begin a transaction
        conn.set_autocommit(false)
            .map_err(|e| Error::transaction(format!("Failed to begin transaction: {}", e)))?;
        
        Ok(Transaction { _inner: () })
    }
    
    /// Commit current transaction
    pub fn commit(&mut self) -> SdkResult<()> {
        debug!("Committing transaction");
        
        let conn = self._inner.as_mut()
            .ok_or_else(|| Error::connection("Connection not established".to_string()))?;
        
        conn.commit()
            .map_err(|e| Error::transaction(format!("Failed to commit transaction: {}", e)))?;
        
        // Set autocommit back to true
        conn.set_autocommit(true)
            .map_err(|e| Error::transaction(format!("Failed to set autocommit after commit: {}", e)))?;
        
        Ok(())
    }
    
    /// Rollback current transaction
    pub fn rollback(&mut self) -> SdkResult<()> {
        debug!("Rolling back transaction");
        
        let conn = self._inner.as_mut()
            .ok_or_else(|| Error::connection("Connection not established".to_string()))?;
        
        conn.rollback()
            .map_err(|e| Error::transaction(format!("Failed to rollback transaction: {}", e)))?;
        
        // Set autocommit back to true
        conn.set_autocommit(true)
            .map_err(|e| Error::transaction(format!("Failed to set autocommit after rollback: {}", e)))?;
        
        Ok(())
    }
    
    /// Create a query builder
    pub fn query_builder(&self) -> QueryBuilder {
        QueryBuilder::new()
    }
    
    /// Get database information
    pub fn database_info(&self) -> SdkResult<DatabaseInfo> {
        // Return database info based on connection options
        Ok(DatabaseInfo {
            dbms_name: "Dameng DB".to_string(),
            db_name: self.options.schema.clone(),
            driver_name: "DM8 ODBC DRIVER".to_string(),
            driver_version: "8.1.4".to_string(),
        })
    }
}

/// Database information
#[derive(Debug, Clone)]
pub struct DatabaseInfo {
    /// Database management system name
    pub dbms_name: String,
    /// Database name
    pub db_name: String,
    /// Driver name
    pub driver_name: String,
    /// Driver version
    pub driver_version: String,
}

impl fmt::Display for DatabaseInfo {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "DatabaseInfo {{ dbms: {}, name: {}, driver: {} {} }}",
            self.dbms_name, self.db_name, self.driver_name, self.driver_version
        )
    }
}

/// Query result
#[derive(Debug, Clone)]
pub struct QueryResult {
    sql: String,
    columns: Vec<String>,
    rows: Vec<Vec<DamengValue>>,
}

impl QueryResult {
    /// Create a new query result from cursor (simplified for now)
    fn from_cursor<C>(mut cursor: odbc_api::CursorImpl<C>, sql: &str) -> SdkResult<Self>
    where
        C: odbc_api::handles::AsStatementRef,
    {
        use odbc_api::buffers::TextRowSet;
        
        // Get column count
        let num_cols = cursor.num_result_cols()
            .map_err(|e| Error::query(format!("Failed to get column count: {}", e)))? as usize;
        
        // Get column names
        let mut columns = Vec::with_capacity(num_cols);
        for col_index in 1..=num_cols as u16 {
            let name = cursor.col_name(col_index)
                .map_err(|e| Error::query(format!("Failed to get column name: {}", e)))?;
            columns.push(name);
        }
        
        // If no columns or no data, return empty result
        if num_cols == 0 {
            return Ok(Self {
                sql: sql.to_string(),
                columns,
                rows: Vec::new(),
            });
        }
        
        // Create buffer for text access with reasonable batch size
        let batch_size = 250;
        let max_str_len = 4000;
        let mut buffer = TextRowSet::for_cursor(batch_size, &mut cursor, Some(max_str_len))
            .map_err(|e| Error::query(format!("Failed to create text buffer: {}", e)))?;
        
        // Bind buffer and fetch
        let mut block_cursor = cursor.bind_buffer(&mut buffer)
            .map_err(|e| Error::query(format!("Failed to bind buffer: {}", e)))?;
        
        let mut rows = Vec::new();
        
        // Fetch data in batches
        while let Some(batch) = block_cursor.fetch()
            .map_err(|e| Error::query(format!("Failed to fetch batch: {}", e)))?
        {
            let num_rows = batch.num_rows();
            
            for row_idx in 0..num_rows {
                let mut row = Vec::with_capacity(num_cols);
                
                for col_idx in 0..num_cols {
                    let text = batch.at(col_idx, row_idx);
                    if let Some(text_bytes) = text {
                        // Use GBK to UTF-8 conversion
                        let string_value = gbk_to_utf8(text_bytes)?;
                        row.push(DamengValue::String(string_value));
                    } else {
                        row.push(DamengValue::Null);
                    }
                }
                
                rows.push(row);
            }
        }
        
        Ok(Self {
            sql: sql.to_string(),
            columns,
            rows,
        })
    }
    
    /// Create a new query result
    pub fn new() -> Self {
        Self {
            sql: String::new(),
            columns: Vec::new(),
            rows: Vec::new(),
        }
    }
    
    /// Get the number of columns
    pub fn num_columns(&self) -> SdkResult<usize> {
        Ok(self.columns.len())
    }
    
    /// Get column names
    pub fn column_names(&self) -> SdkResult<Vec<String>> {
        Ok(self.columns.clone())
    }
    
    /// Get the number of rows
    pub fn num_rows(&self) -> usize {
        self.rows.len()
    }
    
    /// Fetch all rows
    pub fn fetch_all(&mut self) -> SdkResult<Vec<Vec<DamengValue>>> {
        Ok(self.rows.clone())
    }
    
    /// Get a specific row by index
    pub fn get_row(&self, index: usize) -> Option<&Vec<DamengValue>> {
        self.rows.get(index)
    }
    
    /// Get a specific value by row and column index
    pub fn get_value(&self, row: usize, col: usize) -> Option<&DamengValue> {
        self.rows.get(row)?.get(col)
    }
}

/// Build ODBC connection string from options
fn build_connection_string(options: &ConnectionOptions) -> String {
    let mut parts = vec![
        format!("DRIVER={}",options.driver),
        format!("SERVER={}", options.server),
        format!("PORT={}", options.port),
        format!("UID={}", options.username),
        format!("PWD={}", options.password),
        format!("DATABASE={}", options.schema),
    ];
    
    // Add TLS parameter if enabled
    if options.use_tls {
        parts.push("ENCRYPT=yes".to_string());
    }
    
    // Add additional parameters
    for (key, value) in &options.additional_params {
        parts.push(format!("{}={}", key, value));
    }
    
    parts.join(";")
}

/// Mask password in connection string for logging
fn mask_password(connection_string: &str) -> String {
    // Simple password masking for logging
    let parts: Vec<&str> = connection_string.split(';').collect();
    let masked_parts: Vec<String> = parts
        .iter()
        .map(|part| {
            if part.starts_with("PWD=") || part.starts_with("PASSWORD=") {
                // Mask the password value
                if let Some((key, _)) = part.split_once('=') {
                    format!("{}=*****", key)
                } else {
                    part.to_string()
                }
            } else {
                part.to_string()
            }
        })
        .collect();
    
    masked_parts.join(";")
}

impl Transaction {
    /// Commit the transaction (requires calling commit() on the connection instead)
    #[deprecated(note = "Use Connection::commit() instead")]
    pub fn commit(&mut self) -> SdkResult<()> {
        Ok(())
    }
    
    /// Rollback the transaction (requires calling rollback() on the connection instead)
    #[deprecated(note = "Use Connection::rollback() instead")]
    pub fn rollback(&mut self) -> SdkResult<()> {
        Ok(())
    }
}