// Copyright 2025 The Drasi Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! MS SQL Server executor for stored procedure invocation.

use anyhow::{anyhow, Result};
use drasi_lib::identity::Credentials;
use log::{debug, info};
use serde_json::Value;
use std::sync::Arc;
use std::time::Duration;
use tiberius::{
    AuthMethod, Client as TiberiusClient, Config as TiberiusConfig, EncryptionLevel, Query,
};
use tokio::net::TcpStream;
use tokio::sync::RwLock;
use tokio::time::timeout;
use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};

use crate::config::MsSqlStoredProcReactionConfig;

/// MS SQL Server stored procedure executor
pub struct MsSqlExecutor {
    client: Arc<RwLock<TiberiusClient<Compat<TcpStream>>>>,
    command_timeout: Duration,
    retry_attempts: u32,
}

impl MsSqlExecutor {
    /// Create a new MS SQL executor
    ///
    /// The `identity_provider` parameter allows injecting a credential provider
    /// from the runtime context. If provided, it takes precedence over the
    /// config's identity_provider. Falls back to config's user/password if neither is set.
    pub async fn new(
        config: &MsSqlStoredProcReactionConfig,
        identity_provider: Option<std::sync::Arc<dyn drasi_lib::identity::IdentityProvider>>,
    ) -> Result<Self> {
        let port = config.get_port();

        // Resolve credentials: injected provider > config provider > user/password
        let effective_provider = identity_provider.as_ref().map(|p| p.as_ref());
        let config_provider = config.identity_provider.as_deref();
        let provider = effective_provider.or(config_provider);

        let (username, password) = if let Some(provider) = provider {
            debug!("Using identity provider for authentication");
            let context = drasi_lib::identity::CredentialContext::new()
                .with_property("hostname", &config.hostname)
                .with_property("port", port.to_string());
            let credentials = provider.get_credentials(&context).await?;
            if credentials.is_certificate() {
                anyhow::bail!(
                    "Certificate-based authentication is not supported for MS SQL Server. \
                     The tiberius driver does not expose client certificate configuration. \
                     Use token or password authentication instead."
                );
            }
            credentials
                .try_into_auth_pair()
                .map_err(|_| anyhow::anyhow!("Unexpected credential type for MS SQL Server"))?
        } else {
            debug!("Using username/password for authentication");
            (config.user.clone(), config.password.clone())
        };

        info!(
            "Connecting to MS SQL Server: {}:{}/{}",
            config.hostname, port, config.database
        );

        // Build MS SQL connection config
        let mut tiberius_config = TiberiusConfig::new();
        tiberius_config.host(&config.hostname);
        tiberius_config.port(port);
        tiberius_config.authentication(AuthMethod::sql_server(&username, &password));
        tiberius_config.database(&config.database);
        tiberius_config.trust_cert(); // For self-signed certificates

        // Configure encryption
        if config.ssl {
            tiberius_config.encryption(EncryptionLevel::Required);
        } else {
            tiberius_config.encryption(EncryptionLevel::NotSupported);
        }

        // Connect to the database
        let tcp = TcpStream::connect(tiberius_config.get_addr())
            .await
            .map_err(|e| anyhow!("Failed to connect to MS SQL Server: {e}"))?;

        tcp.set_nodelay(true)
            .map_err(|e| anyhow!("Failed to set TCP_NODELAY: {e}"))?;

        let client = TiberiusClient::connect(tiberius_config, tcp.compat_write())
            .await
            .map_err(|e| anyhow!("Failed to establish MS SQL connection: {e}"))?;

        info!(
            "Connected to MS SQL Server: {}:{}/{}",
            config.hostname, port, config.database
        );

        Ok(Self {
            client: Arc::new(RwLock::new(client)),
            command_timeout: Duration::from_millis(config.command_timeout_ms),
            retry_attempts: config.retry_attempts,
        })
    }

    /// Test the database connection
    pub async fn test_connection(&self) -> Result<()> {
        let mut client = self.client.write().await;

        timeout(self.command_timeout, client.simple_query("SELECT 1"))
            .await
            .map_err(|_| anyhow!("Connection test timed out"))?
            .map_err(|e| anyhow!("Connection test failed: {e}"))?;

        info!("Database connection test successful");
        Ok(())
    }

    /// Execute a rendered command with the given ordered parameters.
    ///
    /// The `command` is the SQL batch produced by rendering a command template
    /// (typically `EXEC <proc> @P1, @P2, …`). Parameters are bound positionally
    /// through the driver in the order they appear in `parameters`, matching the
    /// `@P1`, `@P2`, … placeholders emitted by the `{{param}}` template helper.
    /// Values are never interpolated into the SQL text, so row data cannot alter
    /// the statement.
    pub async fn execute_command(&self, command: &str, parameters: Vec<Value>) -> Result<()> {
        let command = command.to_string();
        let params = parameters;
        let client = self.client.clone();
        let cmd_timeout = self.command_timeout;

        self.execute_with_retry(|| async {
            let mut client = client.write().await;

            debug!("Executing: {} with {} parameters", command, params.len());

            // Build parameterized query
            let mut query = Query::new(command.clone());

            // Bind parameters - tiberius requires explicit type binding.
            for param in params.iter() {
                match param {
                    Value::Null => {
                        query.bind(None::<&str>);
                    }
                    Value::Bool(b) => {
                        query.bind(*b);
                    }
                    Value::Number(n) => {
                        // Bind integers as BIGINT and reals as FLOAT
                        // consistently, so a parameter's SQL type does not
                        // depend on the magnitude of a particular row's value.
                        if let Some(i) = n.as_i64() {
                            query.bind(i);
                        } else if let Some(u) = n.as_u64() {
                            // A u64 that does not fit in i64 (> i64::MAX) cannot
                            // be represented by SQL BIGINT. Bind its exact decimal
                            // form as a string rather than falling through to f64,
                            // which would silently lose integer precision.
                            query.bind(u.to_string());
                        } else if let Some(f) = n.as_f64() {
                            query.bind(f);
                        } else {
                            query.bind(n.to_string());
                        }
                    }
                    Value::String(s) => {
                        query.bind(s.as_str());
                    }
                    Value::Array(_) | Value::Object(_) => {
                        // Complex values are bound as their JSON string form.
                        query.bind(param.to_string());
                    }
                }
            }

            // Execute the command
            timeout(cmd_timeout, query.execute(&mut *client))
                .await
                .map_err(|_| anyhow!("Command execution timed out after {cmd_timeout:?}"))?
                .map_err(|e| anyhow!("Failed to execute command: {e}"))?;

            debug!("Command executed successfully");
            Ok(())
        })
        .await
    }

    /// Execute with retry logic
    async fn execute_with_retry<F, Fut>(&self, operation: F) -> Result<()>
    where
        F: Fn() -> Fut,
        Fut: std::future::Future<Output = Result<()>>,
    {
        let mut last_error = None;

        for attempt in 0..=self.retry_attempts {
            if attempt > 0 {
                let backoff = Duration::from_millis(100 * 2u64.pow(attempt - 1));
                debug!("Retrying after {backoff:?} (attempt {attempt})");
                tokio::time::sleep(backoff).await;
            }

            match operation().await {
                Ok(()) => return Ok(()),
                Err(e) => {
                    last_error = Some(e);
                    if attempt < self.retry_attempts {
                        debug!("Attempt {} failed, retrying...", attempt + 1);
                    }
                }
            }
        }

        Err(last_error.unwrap_or_else(|| anyhow!("Operation failed with no error")))
    }
}
