pixeluvw_supabase 0.1.0

#[cfg(feature = "auth")]
use crate::auth::AuthService;
use crate::builder::QueryBuilder;
use crate::client_builder::SupabaseClientBuilder;
use crate::config::ClientConfig;
use crate::error::{Result, SupaError};
#[cfg(feature = "functions")]
use crate::functions::FunctionsClient;
#[cfg(feature = "realtime")]
use crate::realtime::RealtimeClient;
use crate::schema::Schema;
#[cfg(feature = "storage")]
use crate::storage::StorageClient;

use reqwest::Client;
use serde_json::Value;
use std::sync::Arc;
use url::Url;

//  SupabaseClient
// ============================================================================

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

/// Supabase user profile.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct User {
    /// Unique user identifier (UUID).
    pub id: String,
    /// User's email address.
    pub email: Option<String>,
    /// Application metadata (managed by server).
    pub app_metadata: serde_json::Value,
    /// User metadata (can be updated by user).
    pub user_metadata: serde_json::Value,
    /// When the user was created.
    pub created_at: DateTime<Utc>,
}

/// Authentication session containing tokens and user info.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Session {
    /// JWT access token for authenticating requests.
    pub access_token: String,
    /// Token type (usually "bearer").
    pub token_type: String,
    /// Seconds until the token expires.
    pub expires_in: i64,
    /// Token for refreshing the session.
    pub refresh_token: Option<String>,
    /// User profile if available.
    pub user: Option<User>,
    /// Unix timestamp when the session expires (client-calculated).
    pub expires_at: Option<i64>,
}

impl Session {
    /// Check if the session has expired.
    pub fn is_expired(&self) -> bool {
        if let Some(exp) = self.expires_at {
            let now = Utc::now().timestamp();
            // detailed check: consider expired if within 60 seconds of expiration to be safe
            now >= (exp - 60)
        } else {
            false
        }
    }
}

// ============================================================================

// ============================================================================
//  Client Builder
// ============================================================================
// (Empty now)

// ============================================================================
//  Middleware System
// ============================================================================

/// Middleware context passed through the request lifecycle.
#[derive(Debug, Clone)]
pub struct RequestContext {
    pub table: String,
    pub method: String,
    pub url: String,
}

/// Middleware trait for intercepting requests and responses.
///
/// Implement this trait to add logging, metrics, authentication injection,
/// request modification, or any other cross-cutting concerns.
///
/// # Example
/// ```ignore
/// struct LoggingMiddleware;
///
/// impl Middleware for LoggingMiddleware {
///     fn on_request(&self, ctx: &RequestContext) {
///         println!("Request: {} {}", ctx.method, ctx.url);
///     }
///     
///     fn on_response(&self, ctx: &RequestContext, status: u16, duration_ms: u64) {
///         println!("Response: {} {} in {}ms", ctx.method, status, duration_ms);
///     }
/// }
/// ```
pub trait Middleware: Send + Sync {
    /// Called before a request is sent.
    fn on_request(&self, ctx: &RequestContext) {
        let _ = ctx; // Default no-op
    }

    /// Called after a response is received.
    fn on_response(&self, ctx: &RequestContext, status: u16, duration_ms: u64) {
        let _ = (ctx, status, duration_ms); // Default no-op
    }

    /// Called when an error occurs.
    fn on_error(&self, ctx: &RequestContext, error: &str) {
        let _ = (ctx, error); // Default no-op
    }
}

pub(crate) struct SupabaseInner {
    pub(crate) url: Url,
    pub(crate) key: String,
    pub(crate) http: Client,
    pub(crate) config: ClientConfig,
    pub(crate) middlewares: Vec<Arc<dyn Middleware>>,
    pub(crate) session: std::sync::RwLock<Option<Session>>,
    /// Whether the client was configured with a service_role key.
    pub(crate) is_service_role: bool,
    #[cfg(feature = "auth")]
    pub(crate) session_store: std::sync::RwLock<Option<Arc<dyn crate::auth_store::SessionStore>>>,
    pub(crate) schema: std::sync::RwLock<Option<Schema>>,
}

#[derive(Clone)]
pub struct SupabaseClient {
    pub(crate) inner: Arc<SupabaseInner>,
}

impl std::fmt::Debug for SupabaseClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let session_preview = self
            .inner
            .session
            .read()
            .ok()
            .and_then(|lock| lock.as_ref().map(|_| "Some(Session { ... })"));
        f.debug_struct("SupabaseClient")
            .field("url", &self.inner.url)
            .field("session", &session_preview)
            .field("middlewares", &self.inner.middlewares.len())
            .finish()
    }
}

impl SupabaseClient {
    /// Create a new client builder.
    pub fn builder() -> SupabaseClientBuilder {
        SupabaseClientBuilder::new()
    }

    /// Create a new client with default configuration.
    ///
    /// # Arguments
    ///
    /// * `url` - Your Supabase project URL (e.g., `https://xxx.supabase.co`)
    /// * `key` - Your Supabase anon/service key
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use pixeluvw_supabase::SupabaseClient;
    ///
    /// let client = SupabaseClient::new(
    ///     "https://your-project.supabase.co",
    ///     "your-anon-key"
    /// )?;
    /// # Ok::<(), pixeluvw_supabase::SupaError>(())
    /// ```
    ///
    /// # Security Warning
    ///
    /// **Never hardcode credentials in production!** Use [`from_env`](Self::from_env) instead.
    pub fn new(url: &str, key: &str) -> Result<Self> {
        Self::with_config(url, key, ClientConfig::default())
    }

    /// Create a new client from environment variables.
    ///
    /// This is the **recommended** way to create a client in production.
    ///
    /// # Environment Variables
    ///
    /// Required environment variables:
    /// - `SUPABASE_URL` - Your Supabase project URL
    /// - `SUPABASE_KEY` - Your Supabase anon or service key
    ///
    /// # Loading from .env files
    ///
    /// This method does **not** automatically load `.env` files.
    /// Call `dotenv::dotenv().ok()` before calling this method if needed:
    ///
    /// ```rust,no_run
    /// use pixeluvw_supabase::SupabaseClient;
    ///
    /// dotenv::dotenv().ok(); // Load .env file first
    /// let client = SupabaseClient::from_env()?;
    /// # Ok::<(), pixeluvw_supabase::SupaError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if `SUPABASE_URL` or `SUPABASE_KEY` environment variables are not set.
    pub fn from_env() -> Result<Self> {
        let url = std::env::var("SUPABASE_URL").map_err(|_| SupaError::ClientError {
            message: "SUPABASE_URL environment variable not set".to_string(),
        })?;
        let key = std::env::var("SUPABASE_KEY").map_err(|_| SupaError::ClientError {
            message: "SUPABASE_KEY environment variable not set".to_string(),
        })?;
        Self::new(&url, &key)
    }

    /// Create a new client from environment variables with custom configuration.
    ///
    /// Combines the convenience of [`from_env`](Self::from_env) with custom configuration options.
    ///
    /// # Loading from .env files
    ///
    /// This method does **not** automatically load `.env` files.
    /// Call `dotenv::dotenv().ok()` before calling this method if needed.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use pixeluvw_supabase::{SupabaseClient, ClientConfig};
    ///
    /// dotenv::dotenv().ok(); // Load .env file first
    ///
    /// let config = ClientConfig {
    ///     timeout_secs: 60,
    ///     max_retries: 5,
    ///     ..Default::default()
    /// };
    ///
    /// let client = SupabaseClient::from_env_with_config(config)?;
    /// # Ok::<(), pixeluvw_supabase::SupaError>(())
    /// ```
    pub fn from_env_with_config(config: ClientConfig) -> Result<Self> {
        let url = std::env::var("SUPABASE_URL").map_err(|_| SupaError::ClientError {
            message: "SUPABASE_URL environment variable not set".to_string(),
        })?;
        let key = std::env::var("SUPABASE_KEY").map_err(|_| SupaError::ClientError {
            message: "SUPABASE_KEY environment variable not set".to_string(),
        })?;
        Self::with_config(&url, &key, config)
    }

    /// Create a new client with custom configuration.
    ///
    /// # Arguments
    ///
    /// * `url` - Your Supabase project URL
    /// * `key` - Your Supabase anon/service key
    /// * `config` - Custom configuration for timeouts, retries, etc.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use pixeluvw_supabase::{SupabaseClient, ClientConfig};
    ///
    /// let config = ClientConfig {
    ///     timeout_secs: 60,
    ///     max_retries: 5,
    ///     retry_base_delay_ms: 200,
    /// };
    ///
    /// let client = SupabaseClient::with_config(
    ///     "https://your-project.supabase.co",
    ///     "your-anon-key",
    ///     config
    /// )?;
    /// # Ok::<(), pixeluvw_supabase::SupaError>(())
    /// ```
    pub fn with_config(url: &str, key: &str, config: ClientConfig) -> Result<Self> {
        let http = Client::builder()
            .timeout(std::time::Duration::from_secs(config.timeout_secs))
            .build()
            .map_err(|e| SupaError::ClientError {
                message: format!("Failed to build HTTP client: {}", e),
            })?;

        let inner = SupabaseInner {
            url: Url::parse(url)?,
            key: key.to_string(),
            http,
            config,
            middlewares: Vec::new(),
            session: std::sync::RwLock::new(None),
            is_service_role: false,
            #[cfg(feature = "auth")]
            session_store: std::sync::RwLock::new(None),
            schema: std::sync::RwLock::new(None),
        };
        Ok(Self {
            inner: Arc::new(inner),
        })
    }

    /// Manually set the auth token.
    ///
    /// This creates a minimal Session with just the access token.
    /// Auto-refresh will NOT work with this unless a full session is provided via `set_session`.
    pub fn set_auth_token(&self, token: impl Into<String>) {
        let token = token.into();
        let session = Session {
            access_token: token,
            token_type: "bearer".into(),
            expires_in: 3600, // Dummy
            refresh_token: None,
            user: None,
            expires_at: None, // Never expires for check purposes or treated as infinite
        };
        self.set_session(session);
    }

    /// Set a full session (enables auto-refresh if refresh_token is present).
    ///
    /// This also persists the session if a store is configured.
    pub fn set_session(&self, session: Session) {
        if let Ok(mut lock) = self.inner.session.write() {
            *lock = Some(session.clone());
        }

        #[cfg(feature = "auth")]
        {
            if let Ok(store) = self.inner.session_store.read() {
                if let Some(s) = store.as_ref() {
                    let _ = s.save(&session); // Ignore persist errors for now, or log them
                }
            }
        }
    }

    pub fn reset_auth_token(&self) {
        if let Ok(mut lock) = self.inner.session.write() {
            *lock = None;
        }

        #[cfg(feature = "auth")]
        {
            if let Ok(store) = self.inner.session_store.read() {
                if let Some(s) = store.as_ref() {
                    let _ = s.delete();
                }
            }
        }
    }

    /// Get the current session copy, if any.
    ///
    /// Returns `None` if no session is set or if the lock is poisoned.
    pub fn get_session(&self) -> Option<Session> {
        self.inner.session.read().ok().and_then(|lock| lock.clone())
    }

    pub fn from(&self, table: &str) -> QueryBuilder {
        QueryBuilder::new(self.clone(), table)
    }

    /// Add a middleware to the client.
    ///
    /// Middlewares are called in order for requests, and reverse order for responses.
    /// Since the client uses Arc internally, this creates a new client with the
    /// additional middleware.
    ///
    /// # Example
    /// ```ignore
    /// let client = SupabaseClient::new(url, key)?
    ///     .with_middleware(Arc::new(LoggingMiddleware));
    /// ```
    pub fn with_middleware(self, middleware: Arc<dyn Middleware>) -> Self {
        // We need to create a new inner with the middleware
        let mut middlewares = self.inner.middlewares.clone();
        middlewares.push(middleware);

        // Clone current session (use None if lock is poisoned)
        let current_session = self.inner.session.read().ok().and_then(|lock| lock.clone());

        let current_store = self
            .inner
            .session_store
            .read()
            .ok()
            .and_then(|lock| lock.clone());

        #[allow(unused_mut)]
        let mut current_schema = None;
        if let Ok(lock) = self.inner.schema.read() {
            current_schema = lock.clone();
        }

        let new_inner = SupabaseInner {
            url: self.inner.url.clone(),
            key: self.inner.key.clone(),
            http: self.inner.http.clone(),
            config: self.inner.config.clone(),
            middlewares,
            session: std::sync::RwLock::new(current_session),
            is_service_role: self.inner.is_service_role,
            #[cfg(feature = "auth")]
            session_store: std::sync::RwLock::new(current_store),
            schema: std::sync::RwLock::new(current_schema),
        };

        Self {
            inner: Arc::new(new_inner),
        }
    }

    /// Helper to handle Supabase responses.
    pub(crate) async fn handle_response<T: serde::de::DeserializeOwned>(
        &self,
        response: reqwest::Response,
    ) -> Result<T> {
        if !response.status().is_success() {
            let status = response.status().as_u16();
            let error_text = response.text().await.unwrap_or_default();
            // Try to parse as JSON error if possible, but for generic handling, return ApiError
            return Err(SupaError::ApiError {
                code: status,
                message: error_text,
                details: None,
            });
        }
        let data: T = response.json().await?;
        Ok(data)
    }

    #[cfg(feature = "auth")]
    pub fn auth(&self) -> AuthService {
        AuthService::new(self.clone())
    }

    #[cfg(feature = "storage")]
    pub fn storage(&self) -> StorageClient {
        StorageClient::new(self.clone())
    }

    #[cfg(feature = "functions")]
    pub fn functions(&self) -> FunctionsClient {
        FunctionsClient::new(self.clone())
    }

    #[cfg(feature = "realtime")]
    pub fn realtime(&self) -> RealtimeClient {
        RealtimeClient::new(self.clone())
    }

    pub async fn execute(&self, query: QueryBuilder) -> Result<Value> {
        let url = self
            .inner
            .url
            .join(&format!("rest/v1/{}", query.get_table()))?;
        let method = query.get_method().clone();
        let config = &self.inner.config;

        // Create middleware context
        let ctx = RequestContext {
            table: query.get_table().to_string(),
            method: method.to_string(),
            url: url.to_string(),
        };

        // Call on_request for all middlewares
        for mw in &self.inner.middlewares {
            mw.on_request(&ctx);
        }

        let start_time = std::time::Instant::now();
        let mut last_error = None;

        for attempt in 0..=config.max_retries {
            let mut req = self
                .inner
                .http
                .request(method.clone(), url.clone())
                .header("apikey", &self.inner.key);

            // Get token (refreshing if needed)
            if let Ok(token) = self.get_access_token().await {
                req = req.header("Authorization", format!("Bearer {}", token));
            } else {
                req = req.header("Authorization", self.auth_header());
            }

            for (k, v) in query.get_headers() {
                req = req.header(k, v);
            }
            req = req.query(query.get_params());

            if let Some(body) = query.get_body() {
                req = req.json(body);
            }

            match req.send().await {
                Ok(resp) => {
                    let status = resp.status();
                    let duration_ms = start_time.elapsed().as_millis() as u64;

                    // Retry on transient errors (429, 502, 503, 504)
                    if Self::is_retryable_status(status.as_u16()) && attempt < config.max_retries {
                        let delay = config.retry_base_delay_ms * 2u64.pow(attempt);
                        tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
                        last_error = Some(SupaError::ApiError {
                            code: status.as_u16(),
                            message: format!("Retry {}: status {}", attempt + 1, status),
                            details: None,
                        });
                        continue;
                    }

                    // Call on_response for all middlewares (reverse order)
                    for mw in self.inner.middlewares.iter().rev() {
                        mw.on_response(&ctx, status.as_u16(), duration_ms);
                    }

                    if !status.is_success() {
                        let error_text = resp.text().await.unwrap_or_default();
                        let err_msg = format!("Error {}: {}", status, error_text);

                        // Call on_error for all middlewares
                        for mw in &self.inner.middlewares {
                            mw.on_error(&ctx, &err_msg);
                        }

                        return Err(SupaError::ApiError {
                            code: status.as_u16(),
                            message: err_msg,
                            details: Some(error_text),
                        });
                    }

                    let body = resp.json::<Value>().await?;
                    return Ok(body);
                }
                Err(e) => {
                    // Retry on network errors
                    if attempt < config.max_retries {
                        let delay = config.retry_base_delay_ms * 2u64.pow(attempt);
                        tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
                        last_error = Some(SupaError::RequestError(e));
                        continue;
                    }

                    // Call on_error for all middlewares
                    let err_msg = format!("Network error: {}", e);
                    for mw in &self.inner.middlewares {
                        mw.on_error(&ctx, &err_msg);
                    }

                    return Err(SupaError::RequestError(e));
                }
            }
        }

        let err = last_error.unwrap_or_else(|| SupaError::ClientError {
            message: "Max retries exceeded".to_string(),
        });

        // Call on_error for max retries exceeded
        for mw in &self.inner.middlewares {
            mw.on_error(&ctx, &format!("{:?}", err));
        }

        Err(err)
    }

    /// Check if a status code is retryable (transient error).
    fn is_retryable_status(status: u16) -> bool {
        matches!(status, 429 | 502 | 503 | 504)
    }

    /// Internal method to refresh the session if needed.
    pub async fn refresh_session_if_needed(&self) -> Result<()> {
        let needs_refresh;
        let refresh_token;

        {
            let lock = match self.inner.session.read() {
                Ok(l) => l,
                Err(_) => return Ok(()), // Lock poisoned, skip refresh
            };
            if let Some(session) = &*lock {
                if session.is_expired() && session.refresh_token.is_some() {
                    needs_refresh = true;
                    refresh_token = session.refresh_token.clone().unwrap();
                } else {
                    needs_refresh = false;
                    refresh_token = String::new();
                }
            } else {
                needs_refresh = false;
                refresh_token = String::new();
            }
        }

        if needs_refresh {
            let url = self
                .inner
                .url
                .join("auth/v1/token?grant_type=refresh_token")?;
            let params = serde_json::json!({
                "refresh_token": refresh_token,
            });

            // We use the HTTP client directly to avoid recursion
            let resp = self
                .inner
                .http
                .post(url)
                .header("apikey", &self.inner.key)
                .json(&params)
                .send()
                .await?;

            if resp.status().is_success() {
                let mut new_session: Session = resp.json().await?;
                // Calculate expires_at
                let now = Utc::now().timestamp();
                new_session.expires_at = Some(now + new_session.expires_in);

                // Update session
                self.set_session(new_session);
            }
        }

        Ok(())
    }

    /// Get the valid access token, refreshing if necessary.
    pub async fn get_access_token(&self) -> Result<String> {
        self.refresh_session_if_needed().await?;

        let lock = match self.inner.session.read() {
            Ok(l) => l,
            Err(_) => return Ok(self.inner.key.clone()), // Lock poisoned, fallback to anon key
        };
        match &*lock {
            Some(session) => Ok(session.access_token.clone()),
            None => Ok(self.inner.key.clone()), // Fallback to anon key for Public access
        }
    }

    pub async fn rpc(&self, function_name: &str, params: Value) -> Result<Value> {
        let url = self
            .inner
            .url
            .join(&format!("rest/v1/rpc/{}", function_name))?;

        let token = self.get_access_token().await?;

        let resp = self
            .inner
            .http
            .post(url)
            .header("apikey", &self.inner.key)
            .header("Authorization", format!("Bearer {}", token))
            .json(&params)
            .send()
            .await?;

        if !resp.status().is_success() {
            let status = resp.status();
            let error_text = resp.text().await.unwrap_or_default();
            return Err(SupaError::ApiError {
                code: status.as_u16(),
                message: format!("RPC Error: {}", error_text),
                details: Some(error_text),
            });
        }

        let body = resp.json::<Value>().await?;
        Ok(body)
    }

    // Deprecated sync auth_header, mostly for legacy internal use or simple strings
    // But for async flow, prefer `get_access_token`
    pub fn auth_header(&self) -> String {
        let lock = match self.inner.session.read() {
            Ok(l) => l,
            Err(_) => return format!("Bearer {}", self.inner.key), // Lock poisoned, fallback
        };
        match &*lock {
            Some(s) => format!("Bearer {}", s.access_token),
            None => format!("Bearer {}", self.inner.key),
        }
    }

    /// Fetch the database schema from the PostgREST OpenAPI endpoint.
    ///
    /// This returns the raw schema definition. To persist it in the client for
    /// introspection, use [`initialize`](Self::initialize).
    pub async fn fetch_schema(&self) -> Result<Schema> {
        // PostgREST OpenAPI spec is typically at /rest/v1/
        let url = self.inner.url.join("rest/v1/")?;

        let resp = self
            .inner
            .http
            .get(url)
            .header("apikey", &self.inner.key)
            .header("Authorization", self.auth_header())
            .send()
            .await
            .map_err(|e| SupaError::RequestError(e))?;

        if !resp.status().is_success() {
            return Err(SupaError::ApiError {
                code: resp.status().as_u16(),
                message: format!("Failed to fetch schema: {}", resp.status()),
                details: None,
            });
        }

        let schema: Schema = resp.json().await.map_err(|e| SupaError::RequestError(e))?;
        Ok(schema)
    }

    /// Initialize the client by fetching and caching the database schema.
    ///
    /// This allows you to inspect tables and columns using [`get_schema`](Self::get_schema).
    pub async fn initialize(&self) -> Result<()> {
        let schema = self.fetch_schema().await?;
        if let Ok(mut lock) = self.inner.schema.write() {
            *lock = Some(schema);
        }
        Ok(())
    }

    /// Get the cached database schema.
    ///
    /// Returns `None` if [`initialize`](Self::initialize) has not been called.
    pub fn get_schema(&self) -> Option<Schema> {
        let lock = self.inner.schema.read().ok()?;
        lock.clone()
    }
}