athena_rs 1.1.0

//! Supabase client with health tracking and retry logic.
//!
//! This module provides a health-aware wrapper around the Supabase SDK client
//! that tracks connection failures and implements circuit breaker patterns.

use crate::drivers::scylla::health::{self, HostHealthTracker, HostOffline, SystemClock};
use serde_json::Value;
use std::env::var;
use std::error::Error;
use std::time::Duration;
use std::time::Instant;
use supabase_rs::SupabaseClient;
use supabase_rs::query::QueryBuilder;

use tracing::{debug, error, info, warn};

/// Configuration for connecting to a Supabase instance.
#[derive(Clone, Debug)]
pub struct SupabaseConnectionInfo {
    /// The Supabase project URL (e.g., "https://xxx.supabase.co")
    pub url: String,
    /// The Supabase API key (anon or service role key)
    pub key: String,
    /// Logical host identifier for health tracking (extracted from URL)
    pub host: String,
    /// Optional display name for logging
    pub name: Option<String>,
}

impl SupabaseConnectionInfo {
    /// Creates a new connection configuration.
    pub fn new(url: String, key: String) -> Self {
        let host: String = extract_host_from_url(&url);
        Self {
            url,
            key,
            host,
            name: None,
        }
    }

    /// Creates a new connection configuration with a display name.
    pub fn with_name(url: String, key: String, name: String) -> Self {
        let host: String = extract_host_from_url(&url);
        Self {
            url,
            key,
            host,
            name: Some(name),
        }
    }

    /// Load from environment variables.
    pub fn from_env(url_key: &str, key_key: &str) -> Result<Self, String> {
        let url: String = var(url_key)
            .map_err(|e| format!("Missing environment variable '{}': {}", url_key, e))?;
        let key: String = var(key_key)
            .map_err(|e| format!("Missing environment variable '{}': {}", key_key, e))?;
        Ok(Self::new(url, key))
    }
}

/// Extract host identifier from a Supabase URL for health tracking.
#[doc(hidden)]
pub fn extract_host_from_url(url: &str) -> String {
    url.trim_start_matches("https://")
        .trim_start_matches("http://")
        .split('/')
        .next()
        .unwrap_or(url)
        .to_string()
}

/// Health-aware Supabase client wrapper.
///
/// This client wraps the Supabase SDK and adds:
/// - Connection health tracking
/// - Automatic circuit breaking on repeated failures
/// - Retry logic with exponential backoff
/// - Detailed logging and error handling
pub struct HealthAwareSupabaseClient {
    info: SupabaseConnectionInfo,
    client: SupabaseClient,
    tracker: &'static HostHealthTracker<SystemClock>,
}

impl HealthAwareSupabaseClient {
    /// Creates a new health-aware client.
    ///
    /// # Errors
    ///
    /// Returns an error if the Supabase client cannot be initialized.
    pub fn new(info: SupabaseConnectionInfo) -> Result<Self, Box<dyn Error>> {
        let client: SupabaseClient = SupabaseClient::new(info.url.clone(), info.key.clone())
            .map_err(|e| format!("Failed to create SupabaseClient: {:?}", e))?;

        Ok(Self {
            info,
            client,
            tracker: health::global_tracker(),
        })
    }

    /// Creates a client from environment variables.
    pub fn from_env(url_key: &str, key_key: &str) -> Result<Self, Box<dyn Error>> {
        let info: SupabaseConnectionInfo = SupabaseConnectionInfo::from_env(url_key, key_key)?;
        Self::new(info)
    }

    /// Check if the host is currently offline due to health tracking.
    pub fn is_offline(&self) -> Option<Instant> {
        self.tracker.offline_until(&self.info.host)
    }

    /// Force the host offline for testing or manual circuit breaking.
    pub fn force_offline(&self, duration: Duration) -> Instant {
        warn!(
            host = %self.info.host,
            name = ?self.info.name,
            duration_secs = duration.as_secs(),
            "Forcing Supabase host offline"
        );
        self.tracker.force_offline(&self.info.host, duration)
    }

    /// Reset health tracking state for this host.
    pub fn reset_health(&self) {
        info!(
            host = %self.info.host,
            name = ?self.info.name,
            "Resetting Supabase host health tracking"
        );
        self.tracker.reset_host(&self.info.host);
    }

    /// Record a successful operation.
    fn record_success(&self) {
        self.tracker.record_success(&self.info.host);
    }

    /// Record a failed operation and check if circuit should break.
    fn record_failure(&self) -> Option<Instant> {
        self.tracker.record_failure(&self.info.host)
    }

    /// Check if the host is available before attempting an operation.
    fn check_availability(&self) -> Result<(), Box<dyn Error>> {
        if let Some(deadline) = self.is_offline() {
            return Err(Box::new(HostOffline::new(self.info.host.clone(), deadline)));
        }
        Ok(())
    }

    /// Execute a select query on a table.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use athena_rs::drivers::supabase::client::{HealthAwareSupabaseClient, SupabaseConnectionInfo};
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let info = SupabaseConnectionInfo::from_env("SUPABASE_URL", "SUPABASE_KEY")?;
    /// let client = HealthAwareSupabaseClient::new(info)?;
    ///
    /// let results = client.select("users")
    ///     .columns(vec!["id", "email", "created_at"])
    ///     .execute()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn select(&self, table_name: &str) -> HealthAwareQueryBuilder<'_> {
        HealthAwareQueryBuilder {
            client: self,
            builder: self.client.select(table_name),
            table_name: table_name.to_string(),
        }
    }

    /// Execute a raw query with health tracking.
    pub async fn execute(
        &self,
        table_name: &str,
        query_string: &str,
    ) -> Result<Vec<Value>, Box<dyn Error>> {
        self.check_availability()?;

        debug!(
            host = %self.info.host,
            table = %table_name,
            query = %query_string,
            "Executing Supabase query"
        );

        let rows: Vec<Value> = match self.client.execute(table_name, query_string).await {
            Ok(result) => {
                self.record_success();
                info!(
                    host = %self.info.host,
                    table = %table_name,
                    rows = result.len(),
                    "Supabase query succeeded"
                );
                result
            }
            Err(err) => {
                error!(
                    host = %self.info.host,
                    table = %table_name,
                    error = %err,
                    "Supabase query failed"
                );

                if let Some(deadline) = self.record_failure() {
                    return Err(Box::new(HostOffline::new(self.info.host.clone(), deadline))
                        as Box<dyn Error>);
                }

                return Err(
                    Box::new(std::io::Error::new(std::io::ErrorKind::Other, err)) as Box<dyn Error>,
                );
            }
        };

        Ok(rows)
    }

    /// Insert data into a table.
    pub async fn insert(&self, table_name: &str, data: Value) -> Result<String, Box<dyn Error>> {
        self.check_availability()?;

        debug!(
            host = %self.info.host,
            table = %table_name,
            "Inserting into Supabase table"
        );

        match self.client.insert(table_name, data).await {
            Ok(id) => {
                self.record_success();
                info!(
                    host = %self.info.host,
                    table = %table_name,
                    id = %id,
                    "Supabase insert succeeded"
                );
                Ok(id)
            }
            Err(err) => {
                error!(
                    host = %self.info.host,
                    table = %table_name,
                    error = %err,
                    "Supabase insert failed"
                );

                if let Some(deadline) = self.record_failure() {
                    return Err(Box::new(HostOffline::new(self.info.host.clone(), deadline))
                        as Box<dyn Error>);
                }

                Err(Box::new(std::io::Error::new(std::io::ErrorKind::Other, err)) as Box<dyn Error>)
            }
        }
    }

    /// Update data in a table.
    pub async fn update(
        &self,
        table_name: &str,
        row_id: &str,
        data: Value,
    ) -> Result<(), Box<dyn Error>> {
        self.check_availability()?;

        debug!(
            host = %self.info.host,
            table = %table_name,
            row_id = %row_id,
            "Updating Supabase table"
        );

        match self.client.update(table_name, row_id, data).await {
            Ok(_) => {
                self.record_success();
                info!(
                    host = %self.info.host,
                    table = %table_name,
                    row_id = %row_id,
                    "Supabase update succeeded"
                );
            }
            Err(err) => {
                error!(
                    host = %self.info.host,
                    table = %table_name,
                    row_id = %row_id,
                    error = %err,
                    "Supabase update failed"
                );

                if let Some(deadline) = self.record_failure() {
                    return Err(Box::new(HostOffline::new(self.info.host.clone(), deadline))
                        as Box<dyn Error>);
                }

                return Err(
                    Box::new(std::io::Error::new(std::io::ErrorKind::Other, err)) as Box<dyn Error>,
                );
            }
        }

        Ok(())
    }

    /// Delete data from a table.
    pub async fn delete(&self, table_name: &str, row_id: &str) -> Result<(), Box<dyn Error>> {
        self.check_availability()?;

        debug!(
            host = %self.info.host,
            table = %table_name,
            row_id = %row_id,
            "Deleting from Supabase table"
        );

        match self.client.delete(table_name, row_id).await {
            Ok(_) => {
                self.record_success();
                info!(
                    host = %self.info.host,
                    table = %table_name,
                    row_id = %row_id,
                    "Supabase delete succeeded"
                );
                Ok(())
            }
            Err(err) => {
                error!(
                    host = %self.info.host,
                    table = %table_name,
                    row_id = %row_id,
                    error = %err,
                    "Supabase delete failed"
                );

                if let Some(deadline) = self.record_failure() {
                    return Err(Box::new(HostOffline::new(self.info.host.clone(), deadline))
                        as Box<dyn Error>);
                }

                Err(Box::new(std::io::Error::new(std::io::ErrorKind::Other, err)) as Box<dyn Error>)
            }
        }
    }
}

/// Health-aware query builder wrapper.
///
/// Wraps the Supabase SDK QueryBuilder and adds health tracking.
pub struct HealthAwareQueryBuilder<'a> {
    client: &'a HealthAwareSupabaseClient,
    builder: QueryBuilder,
    table_name: String,
}

impl<'a> HealthAwareQueryBuilder<'a> {
    /// Select specific columns.
    pub fn columns(mut self, columns: Vec<&str>) -> Self {
        self.builder = self.builder.columns(columns);
        self
    }

    /// Add an equality filter.
    pub fn eq(mut self, column: &str, value: &str) -> Self {
        self.builder = self.builder.eq(column, value);
        self
    }

    /// Add a not-equal filter.
    pub fn neq(mut self, column: &str, value: &str) -> Self {
        self.builder = self.builder.neq(column, value);
        self
    }

    /// Add a greater-than filter.
    pub fn gt(mut self, column: &str, value: &str) -> Self {
        self.builder = self.builder.gt(column, value);
        self
    }

    /// Add a less-than filter.
    pub fn lt(mut self, column: &str, value: &str) -> Self {
        self.builder = self.builder.lt(column, value);
        self
    }

    /// Add ordering.
    pub fn order(mut self, column: &str, ascending: bool) -> Self {
        self.builder = self.builder.order(column, ascending);
        self
    }

    /// Set result limit.
    pub fn limit(mut self, count: usize) -> Self {
        self.builder = self.builder.limit(count);
        self
    }

    /// Set result offset.
    pub fn offset(mut self, count: usize) -> Self {
        self.builder = self.builder.offset(count);
        self
    }

    /// Execute the query with health tracking.
    pub async fn execute(self) -> Result<Vec<Value>, Box<dyn Error>> {
        self.client.check_availability()?;

        debug!(
            host = %self.client.info.host,
            table = %self.table_name,
            "Executing health-aware Supabase query"
        );

        let rows: Vec<Value> = match self.builder.execute().await {
            Ok(result) => {
                self.client.record_success();
                info!(
                    host = %self.client.info.host,
                    table = %self.table_name,
                    rows = result.len(),
                    "Health-aware query succeeded"
                );
                result
            }
            Err(err) => {
                error!(
                    host = %self.client.info.host,
                    table = %self.table_name,
                    error = %err,
                    "Health-aware query failed"
                );

                if let Some(deadline) = self.client.record_failure() {
                    return Err(
                        Box::new(HostOffline::new(self.client.info.host.clone(), deadline))
                            as Box<dyn Error>,
                    );
                }

                return Err(
                    Box::new(std::io::Error::new(std::io::ErrorKind::Other, err)) as Box<dyn Error>,
                );
            }
        };

        Ok(rows)
    }
}