batata-client 0.0.2

//! Batata Client - Rust client for Batata/Nacos service discovery and configuration management
//!
//! # Overview
//!
//! Batata Client provides a Rust SDK for interacting with Batata/Nacos servers, supporting:
//! - Configuration management (get, publish, remove, listen)
//! - Service discovery (register, deregister, query, subscribe)
//!
//! # Quick Start
//!
//! ```rust,no_run
//! use batata_client::{BatataClient, ClientConfig};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     // Create client
//!     let client = BatataClient::builder()
//!         .server_addr("localhost:8848")
//!         .namespace("public")
//!         .build()
//!         .await?;
//!
//!     // Get configuration
//!     let config_service = client.config_service();
//!     let content = config_service.get_config("my-config", "DEFAULT_GROUP").await?;
//!     println!("Config: {}", content);
//!
//!     // Register service
//!     let naming_service = client.naming_service();
//!     naming_service.register_instance_simple("my-service", "127.0.0.1", 8080).await?;
//!
//!     // Shutdown
//!     client.shutdown().await;
//!
//!     Ok(())
//! }
//! ```

pub mod api;
pub mod auth;
pub mod cache;
pub mod common;
pub mod config;
pub mod crypto;
pub mod error;
pub mod logging;
pub mod naming;
pub mod remote;

#[cfg(test)]
mod tests;

pub use api::config::{
    ConfigBatchListenRequest, ConfigChangeBatchListenResponse, ConfigChangeNotifyRequest,
    ConfigInfo, ConfigListenContext, ConfigPublishRequest, ConfigPublishResponse,
    ConfigQueryRequest, ConfigQueryResponse, ConfigRemoveRequest, ConfigRemoveResponse,
    ConfigSearchRequest, ConfigSearchResponse, ConfigSearchItem,
};
pub use api::naming::{
    BatchInstanceRequest, BatchInstanceResponse, Instance, InstanceRequest, InstanceResponse,
    QueryServiceResponse, Service, ServiceListRequest, ServiceListResponse, ServiceQueryRequest,
    SubscribeServiceRequest, SubscribeServiceResponse,
};
pub use api::remote::{RequestTrait, ResponseTrait};
pub use auth::{AccessToken, AuthManager, Credentials};
pub use cache::FileCache;
pub use common::constants::*;
pub use config::{ConfigChangeEvent, ConfigChangeType, ConfigListener, ConfigService};
pub use error::{BatataError, Result};
pub use naming::{
    LoadBalancer, NamingService, RandomBalancer, ServiceChangeEvent, ServiceListener,
    WeightedRoundRobinBalancer,
};
pub use remote::RpcClient;

use std::collections::HashMap;
use std::sync::Arc;

use parking_lot::RwLock;
use tracing::info;

/// Cache configuration
#[derive(Clone, Debug)]
pub struct CacheConfig {
    /// Cache directory for local failover cache
    pub cache_dir: Option<String>,
    /// Do not load cache at startup
    pub not_load_cache_at_start: bool,
    /// Update in-memory cache when empty (failover mode)
    pub update_cache_when_empty: bool,
    /// Enable file-based failover cache
    pub failover_enabled: bool,
}

impl Default for CacheConfig {
    fn default() -> Self {
        Self {
            cache_dir: None,
            not_load_cache_at_start: false,
            update_cache_when_empty: true,
            failover_enabled: true,
        }
    }
}

impl CacheConfig {
    /// Create a new cache config with directory
    pub fn new(cache_dir: impl Into<String>) -> Self {
        Self {
            cache_dir: Some(cache_dir.into()),
            ..Default::default()
        }
    }

    /// Set cache directory
    pub fn with_cache_dir(mut self, dir: impl Into<String>) -> Self {
        self.cache_dir = Some(dir.into());
        self
    }

    /// Set not load cache at start
    pub fn with_not_load_cache_at_start(mut self, enabled: bool) -> Self {
        self.not_load_cache_at_start = enabled;
        self
    }

    /// Set update cache when empty
    pub fn with_update_cache_when_empty(mut self, enabled: bool) -> Self {
        self.update_cache_when_empty = enabled;
        self
    }

    /// Set failover enabled
    pub fn with_failover_enabled(mut self, enabled: bool) -> Self {
        self.failover_enabled = enabled;
        self
    }
}

/// TLS configuration
#[derive(Clone, Debug, Default)]
pub struct TlsConfig {
    /// Enable TLS
    pub enabled: bool,
    /// Path to CA certificate file (PEM format)
    pub ca_cert_path: Option<String>,
    /// Path to client certificate file (PEM format)
    pub client_cert_path: Option<String>,
    /// Path to client key file (PEM format)
    pub client_key_path: Option<String>,
    /// Skip server certificate verification (not recommended for production)
    pub skip_verify: bool,
}

impl TlsConfig {
    /// Create a new TLS config with TLS enabled
    pub fn new() -> Self {
        Self {
            enabled: true,
            ..Default::default()
        }
    }

    /// Set CA certificate path
    pub fn with_ca_cert(mut self, path: impl Into<String>) -> Self {
        self.ca_cert_path = Some(path.into());
        self
    }

    /// Set client certificate and key paths
    pub fn with_client_cert(
        mut self,
        cert_path: impl Into<String>,
        key_path: impl Into<String>,
    ) -> Self {
        self.client_cert_path = Some(cert_path.into());
        self.client_key_path = Some(key_path.into());
        self
    }

    /// Skip server certificate verification
    pub fn with_skip_verify(mut self, skip: bool) -> Self {
        self.skip_verify = skip;
        self
    }
}

/// Client configuration
#[derive(Clone, Debug)]
pub struct ClientConfig {
    /// Server addresses (host:port)
    pub server_addrs: Vec<String>,
    /// Namespace (default: public)
    pub namespace: String,
    /// Application name
    pub app_name: String,
    /// Custom labels
    pub labels: HashMap<String, String>,
    /// Request timeout in milliseconds
    pub timeout_ms: u64,
    /// Retry times on failure
    pub retry_times: u32,
    /// Authentication credentials
    pub credentials: Credentials,
    /// TLS configuration
    pub tls: TlsConfig,
    /// Cache configuration
    pub cache: CacheConfig,
}

impl Default for ClientConfig {
    fn default() -> Self {
        Self {
            server_addrs: vec!["localhost:8848".to_string()],
            namespace: DEFAULT_NAMESPACE.to_string(),
            app_name: String::new(),
            labels: HashMap::new(),
            timeout_ms: DEFAULT_TIMEOUT_MS,
            retry_times: 3,
            credentials: Credentials::default(),
            tls: TlsConfig::default(),
            cache: CacheConfig::default(),
        }
    }
}

/// Builder for BatataClient
pub struct BatataClientBuilder {
    config: ClientConfig,
}

impl BatataClientBuilder {
    /// Create a new builder with default configuration
    pub fn new() -> Self {
        Self {
            config: ClientConfig::default(),
        }
    }

    /// Set server address (single server)
    pub fn server_addr(mut self, addr: &str) -> Self {
        self.config.server_addrs = vec![addr.to_string()];
        self
    }

    /// Set server addresses (multiple servers)
    pub fn server_addrs(mut self, addrs: Vec<String>) -> Self {
        self.config.server_addrs = addrs;
        self
    }

    /// Set namespace
    pub fn namespace(mut self, namespace: &str) -> Self {
        self.config.namespace = namespace.to_string();
        self
    }

    /// Set application name
    pub fn app_name(mut self, app_name: &str) -> Self {
        self.config.app_name = app_name.to_string();
        self
    }

    /// Add a custom label
    pub fn label(mut self, key: &str, value: &str) -> Self {
        self.config.labels.insert(key.to_string(), value.to_string());
        self
    }

    /// Set labels
    pub fn labels(mut self, labels: HashMap<String, String>) -> Self {
        self.config.labels = labels;
        self
    }

    /// Set timeout in milliseconds
    pub fn timeout_ms(mut self, timeout_ms: u64) -> Self {
        self.config.timeout_ms = timeout_ms;
        self
    }

    /// Set retry times
    pub fn retry_times(mut self, retry_times: u32) -> Self {
        self.config.retry_times = retry_times;
        self
    }

    /// Set username and password for authentication
    pub fn username_password(mut self, username: &str, password: &str) -> Self {
        self.config.credentials = Credentials::with_username_password(username, password);
        self
    }

    /// Set access key and secret key for authentication
    pub fn access_key(mut self, access_key: &str, secret_key: &str) -> Self {
        self.config.credentials = Credentials::with_access_key(access_key, secret_key);
        self
    }

    /// Set credentials
    pub fn credentials(mut self, credentials: Credentials) -> Self {
        self.config.credentials = credentials;
        self
    }

    /// Configure for Alibaba Cloud ACM
    pub fn acm(
        mut self,
        access_key: &str,
        secret_key: &str,
        endpoint: &str,
        region_id: &str,
    ) -> Self {
        self.config.credentials =
            Credentials::with_acm(access_key, secret_key, endpoint, region_id);
        self
    }

    /// Set ACM endpoint
    pub fn acm_endpoint(mut self, endpoint: &str) -> Self {
        self.config.credentials.set_endpoint(endpoint);
        self
    }

    /// Set ACM region ID
    pub fn acm_region_id(mut self, region_id: &str) -> Self {
        self.config.credentials.set_region_id(region_id);
        self
    }

    /// Enable TLS with default settings
    pub fn tls(mut self, enabled: bool) -> Self {
        self.config.tls.enabled = enabled;
        self
    }

    /// Set TLS configuration
    pub fn tls_config(mut self, tls: TlsConfig) -> Self {
        self.config.tls = tls;
        self
    }

    /// Set cache directory for local failover cache
    pub fn cache_dir(mut self, dir: &str) -> Self {
        self.config.cache.cache_dir = Some(dir.to_string());
        self
    }

    /// Set cache configuration
    pub fn cache_config(mut self, cache: CacheConfig) -> Self {
        self.config.cache = cache;
        self
    }

    /// Set not load cache at startup
    pub fn not_load_cache_at_start(mut self, enabled: bool) -> Self {
        self.config.cache.not_load_cache_at_start = enabled;
        self
    }

    /// Set update cache when empty (failover mode)
    pub fn update_cache_when_empty(mut self, enabled: bool) -> Self {
        self.config.cache.update_cache_when_empty = enabled;
        self
    }

    /// Enable/disable failover cache
    pub fn failover_enabled(mut self, enabled: bool) -> Self {
        self.config.cache.failover_enabled = enabled;
        self
    }

    /// Build the client
    pub async fn build(self) -> Result<BatataClient> {
        BatataClient::new(self.config).await
    }
}

impl Default for BatataClientBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Main Batata client
pub struct BatataClient {
    #[allow(dead_code)]
    config: ClientConfig,
    rpc_client: Arc<RpcClient>,
    config_service: Arc<ConfigService>,
    naming_service: Arc<NamingService>,
    started: Arc<RwLock<bool>>,
}

impl BatataClient {
    /// Create a new client builder
    pub fn builder() -> BatataClientBuilder {
        BatataClientBuilder::new()
    }

    /// Create a new client with configuration
    pub async fn new(config: ClientConfig) -> Result<Self> {
        let rpc_client = RpcClient::new(config.server_addrs.clone())?
            .with_namespace(&config.namespace)
            .with_app_name(&config.app_name)
            .with_labels(config.labels.clone())
            .with_timeout(config.timeout_ms)
            .with_retry(config.retry_times);

        // Start RPC client
        rpc_client.start().await?;

        let rpc_client = Arc::new(rpc_client);

        // Create config service with cache config
        let config_service = Arc::new(ConfigService::new(
            rpc_client.clone(),
            &config.namespace,
            config.cache.clone(),
        ));

        // Create naming service with cache config
        let naming_service = Arc::new(NamingService::new(
            rpc_client.clone(),
            &config.namespace,
            config.cache.clone(),
        ));

        let client = Self {
            config,
            rpc_client,
            config_service,
            naming_service,
            started: Arc::new(RwLock::new(true)),
        };

        info!("BatataClient created and connected");

        Ok(client)
    }

    /// Get configuration service
    pub fn config_service(&self) -> Arc<ConfigService> {
        self.config_service.clone()
    }

    /// Get naming service
    pub fn naming_service(&self) -> Arc<NamingService> {
        self.naming_service.clone()
    }

    /// Get RPC client
    pub fn rpc_client(&self) -> Arc<RpcClient> {
        self.rpc_client.clone()
    }

    /// Check if client is connected
    pub fn is_connected(&self) -> bool {
        self.rpc_client.is_connected()
    }

    /// Get connection ID
    pub fn connection_id(&self) -> Option<String> {
        self.rpc_client.connection_id()
    }

    /// Shutdown the client
    pub async fn shutdown(&self) {
        if !*self.started.read() {
            return;
        }

        *self.started.write() = false;

        // Stop services
        self.config_service.stop().await;
        self.naming_service.stop().await;
        self.rpc_client.stop().await;

        info!("BatataClient shutdown");
    }

    /// Start config service (for listening)
    pub async fn start_config_service(&self) -> Result<()> {
        self.config_service.start().await
    }

    /// Start naming service (for heartbeat)
    pub async fn start_naming_service(&self) -> Result<()> {
        self.naming_service.start().await
    }
}

impl Drop for BatataClient {
    fn drop(&mut self) {
        // Note: async drop is not possible, so cleanup should be done via shutdown()
    }
}

// Re-export common types for convenience
pub mod prelude {
    pub use crate::{
        BatataClient, BatataClientBuilder, BatataError, CacheConfig, ClientConfig,
        ConfigChangeEvent, ConfigListener, ConfigService, Instance, NamingService, Result,
        Service, ServiceChangeEvent, ServiceListener,
    };
}