evidentsource_client/

evident_source.rs

//! EvidentSource - the main entrypoint for connecting to an EvidentSource server.

use std::time::Duration;

use chrono::{DateTime, Utc};
use futures::stream::{self, StreamExt};
use futures::Stream;
use http::Uri;
use tonic::transport::{Channel, ClientTlsConfig};

use evidentsource_core::domain::{DatabaseError, DatabaseName};
use evidentsource_core::{DatabaseCatalog, DatabaseIdentity};

use crate::auth::Credentials;
use crate::connection::Connection;
use crate::conversions::timestamp_to_datetime;
use crate::EvidentSourceClient;

/// A simple database identity returned from create_database.
#[derive(Debug, Clone)]
pub struct DatabaseIdentityImpl {
    name: DatabaseName,
    created_at: DateTime<Utc>,
}

impl DatabaseIdentity for DatabaseIdentityImpl {
    fn name(&self) -> &DatabaseName {
        &self.name
    }

    fn created_at(&self) -> DateTime<Utc> {
        self.created_at
    }
}

/// The main entrypoint for connecting to an EvidentSource server.
///
/// `EvidentSource` manages the gRPC connection and provides methods for:
/// - Listing available databases (`DatabaseCatalog` trait)
/// - Creating and deleting databases
/// - Connecting to a specific database for operations
///
/// # Example
///
/// ```ignore
/// use evidentsource_client::EvidentSource;
/// use evidentsource_core::domain::DatabaseName;
///
/// // Connect to the server
/// let es = EvidentSource::connect_to_server("http://localhost:50051").await?;
///
/// // List all databases
/// let mut databases = es.list_databases();
/// while let Some(name) = databases.next().await {
///     println!("Database: {}", name);
/// }
///
/// // Connect to a specific database
/// let db_name = DatabaseName::new("my-db")?;
/// let conn = es.connect(&db_name).await?;
///
/// // Use the connection for operations
/// let latest = conn.latest_database().await?;
/// println!("Latest revision: {}", latest.revision());
/// ```
#[derive(Clone)]
pub struct EvidentSource {
    client: EvidentSourceClient,
}

impl EvidentSource {
    /// Create a builder for configuring and connecting to an EvidentSource server.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use evidentsource_client::EvidentSource;
    /// use std::time::Duration;
    ///
    /// let es = EvidentSource::builder("http://localhost:50051")
    ///     .connect_timeout(Duration::from_secs(10))
    ///     .connect()
    ///     .await?;
    /// ```
    pub fn builder(addr: &str) -> EvidentSourceBuilder {
        EvidentSourceBuilder::new(addr)
    }

    /// Connect to an EvidentSource server without authentication.
    ///
    /// This only works if the server has `allow_anonymous=true`.
    ///
    /// # Arguments
    ///
    /// * `addr` - The server address (e.g., `http://localhost:50051` or `https://api.example.com`)
    ///
    /// # Example
    ///
    /// ```ignore
    /// let es = EvidentSource::connect_to_server("http://localhost:50051").await?;
    /// ```
    pub async fn connect_to_server(addr: &str) -> Result<Self, crate::Error> {
        Self::connect_with_auth(addr, Credentials::None).await
    }

    /// Connect to an EvidentSource server with authentication credentials.
    ///
    /// # Arguments
    ///
    /// * `addr` - The server address (e.g., `http://localhost:50051` or `https://api.example.com`)
    /// * `credentials` - Authentication credentials (BearerToken, DevMode, or None)
    ///
    /// # Examples
    ///
    /// ```ignore
    /// use evidentsource_client::{EvidentSource, Credentials, DevModeCredentials};
    ///
    /// // With bearer token (production - requires TLS)
    /// let es = EvidentSource::connect_with_auth(
    ///     "https://api.example.com:50051",
    ///     Credentials::BearerToken(my_jwt_token),
    /// ).await?;
    ///
    /// // With DevMode credentials (local development)
    /// let es = EvidentSource::connect_with_auth(
    ///     "http://localhost:50051",
    ///     Credentials::DevMode(
    ///         DevModeCredentials::new("dev-user@example.com")
    ///             .with_email("dev@example.com")
    ///             .with_display_name("Developer")
    ///     ),
    /// ).await?;
    /// ```
    pub async fn connect_with_auth(
        addr: &str,
        credentials: Credentials,
    ) -> Result<Self, crate::Error> {
        let client = EvidentSourceClient::with_credentials(addr, credentials).await?;
        Ok(Self { client })
    }

    /// Create an EvidentSource instance from an existing client.
    pub fn from_client(client: EvidentSourceClient) -> Self {
        Self { client }
    }

    /// Connect to a specific database.
    ///
    /// This returns a `Connection` that maintains a live subscription to
    /// database updates and implements `DatabaseProvider` and `DatabaseConnection`.
    ///
    /// # Arguments
    ///
    /// * `database` - The name of the database to connect to
    ///
    /// # Example
    ///
    /// ```ignore
    /// let db_name = DatabaseName::new("my-db")?;
    /// let conn = es.connect(&db_name).await?;
    /// ```
    pub async fn connect(&self, database: &DatabaseName) -> Result<Connection, DatabaseError> {
        Connection::new(self.client.clone(), database.clone()).await
    }

    /// Get a reference to the underlying gRPC client.
    ///
    /// This provides access to low-level operations not exposed through
    /// the high-level API.
    pub fn client(&self) -> &EvidentSourceClient {
        &self.client
    }

    /// Get a mutable reference to the underlying gRPC client.
    pub fn client_mut(&mut self) -> &mut EvidentSourceClient {
        &mut self.client
    }
}

impl std::fmt::Debug for EvidentSource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EvidentSource").finish()
    }
}

impl DatabaseCatalog for EvidentSource {
    type Identity = DatabaseIdentityImpl;

    fn list_databases(&self) -> impl Stream<Item = DatabaseName> {
        let mut client = self.client.clone();

        stream::once(async move {
            let result = client.fetch_catalog().await;

            match result {
                Ok(response_stream) => response_stream
                    .filter_map(|result| async move {
                        match result {
                            Ok(reply) => DatabaseName::new(&reply.database_name).ok(),
                            Err(_) => None,
                        }
                    })
                    .boxed(),
                Err(_) => stream::empty().boxed(),
            }
        })
        .flatten()
    }

    fn create_database(
        &self,
        name: DatabaseName,
    ) -> impl std::future::Future<Output = Result<Self::Identity, DatabaseError>> {
        let mut client = self.client.clone();

        async move {
            let proto_db = client
                .create_database(name.to_string())
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &name.to_string())
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            let db_name = DatabaseName::new(&proto_db.name)?;
            let created_at = proto_db
                .created_at
                .ok_or_else(|| {
                    DatabaseError::ServerError("missing created_at timestamp".to_string())
                })
                .and_then(|ts| {
                    timestamp_to_datetime(ts).map_err(|e| {
                        DatabaseError::ServerError(format!("invalid timestamp: {}", e))
                    })
                })?;

            Ok(DatabaseIdentityImpl {
                name: db_name,
                created_at,
            })
        }
    }

    fn delete_database(
        &self,
        name: DatabaseName,
    ) -> impl std::future::Future<Output = Result<(), DatabaseError>> {
        let mut client = self.client.clone();

        async move {
            client
                .delete_database(name.to_string())
                .await
                .map_err(|e| match e {
                    crate::Error::GrpcStatus(ref status) => {
                        crate::status_mapping::to_database_error(status, &name.to_string())
                    }
                    _ => DatabaseError::ServerError(e.to_string()),
                })?;

            Ok(())
        }
    }
}

/// TLS configuration for the connection.
#[derive(Debug, Clone)]
pub enum TlsConfig {
    /// Use the system's native TLS roots.
    Native,
    /// Disable TLS verification (not recommended for production).
    Disabled,
}

/// Configuration for exponential backoff retry behavior.
#[derive(Debug, Clone)]
pub struct BackoffConfig {
    /// Initial delay before first retry.
    pub initial: Duration,
    /// Maximum delay between retries.
    pub max: Duration,
    /// Multiplier applied to delay after each retry.
    pub multiplier: f64,
}

impl Default for BackoffConfig {
    fn default() -> Self {
        Self {
            initial: Duration::from_millis(100),
            max: Duration::from_secs(30),
            multiplier: 2.0,
        }
    }
}

/// Builder for configuring and connecting to an EvidentSource server.
///
/// # Example
///
/// ```ignore
/// use evidentsource_client::{EvidentSource, Credentials, DevModeCredentials};
/// use std::time::Duration;
///
/// let es = EvidentSource::builder("http://localhost:50051")
///     .credentials(Credentials::DevMode(
///         DevModeCredentials::new("dev-user@example.com")
///     ))
///     .connect_timeout(Duration::from_secs(10))
///     .tls(TlsConfig::Native)
///     .connect()
///     .await?;
/// ```
#[derive(Debug, Clone)]
pub struct EvidentSourceBuilder {
    addr: String,
    credentials: Credentials,
    tls_config: Option<TlsConfig>,
    connect_timeout: Option<Duration>,
    backoff: Option<BackoffConfig>,
}

impl EvidentSourceBuilder {
    /// Create a new builder with the given server address.
    pub fn new(addr: &str) -> Self {
        Self {
            addr: addr.to_string(),
            credentials: Credentials::None,
            tls_config: None,
            connect_timeout: None,
            backoff: None,
        }
    }

    /// Set the authentication credentials.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let es = EvidentSource::builder("http://localhost:50051")
    ///     .credentials(Credentials::DevMode(DevModeCredentials::new("dev-user")))
    ///     .connect()
    ///     .await?;
    /// ```
    pub fn credentials(mut self, credentials: Credentials) -> Self {
        self.credentials = credentials;
        self
    }

    /// Set the TLS configuration.
    ///
    /// For HTTPS addresses, TLS is enabled by default with native roots.
    pub fn tls(mut self, config: TlsConfig) -> Self {
        self.tls_config = Some(config);
        self
    }

    /// Set the connection timeout.
    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
        self.connect_timeout = Some(timeout);
        self
    }

    /// Set the backoff configuration for retries.
    pub fn backoff(mut self, config: BackoffConfig) -> Self {
        self.backoff = Some(config);
        self
    }

    /// Connect to the server with the configured options.
    pub async fn connect(self) -> Result<EvidentSource, crate::Error> {
        let uri = Uri::try_from(&self.addr)?;
        let mut channel_builder = Channel::builder(uri.clone());

        // Apply connect timeout
        if let Some(timeout) = self.connect_timeout {
            channel_builder = channel_builder.connect_timeout(timeout);
        }

        // Apply TLS configuration
        let is_https = uri.scheme_str() == Some("https");
        match self.tls_config {
            Some(TlsConfig::Native) | None if is_https => {
                let tls_config = ClientTlsConfig::new()
                    .with_native_roots()
                    .domain_name(uri.host().unwrap_or("localhost"));
                channel_builder = channel_builder.tls_config(tls_config)?;
            }
            Some(TlsConfig::Disabled) => {
                // TLS disabled, don't configure
            }
            _ => {
                // HTTP connection, no TLS needed
            }
        }

        let channel = channel_builder.connect().await?;
        let interceptor = crate::auth::AuthInterceptor::new(self.credentials);
        let client =
            crate::com::evidentsource::evident_source_client::EvidentSourceClient::with_interceptor(
                channel,
                interceptor,
            );

        Ok(EvidentSource {
            client: EvidentSourceClient { client },
        })
    }

    /// Get the configured backoff settings.
    ///
    /// This can be used when creating connections to propagate retry settings.
    pub fn get_backoff(&self) -> Option<&BackoffConfig> {
        self.backoff.as_ref()
    }
}