redis-enterprise 0.8.7

//! Cluster management for Redis Enterprise
//!
//! ## Overview
//! - Query cluster info, settings, topology, and license
//! - Manage nodes (join, remove, maintenance mode)
//! - Configure cluster policies and services
//! - Handle certificates and LDAP configuration
//!
//! ## Examples
//!
//! ### Getting Cluster Information
//! ```no_run
//! use redis_enterprise::EnterpriseClient;
//!
//! # async fn example(client: EnterpriseClient) -> Result<(), Box<dyn std::error::Error>> {
//! // Get basic cluster info
//! let info = client.cluster().info().await?;
//! println!("Cluster: {}", info.name);
//!
//! // Check license status
//! let license = client.cluster().license().await?;
//! println!("Licensed shards: {:?}", license.shards_limit);
//! # Ok(())
//! # }
//! ```
//!
//! ### Node Management
//! ```no_run
//! # use redis_enterprise::EnterpriseClient;
//! # async fn example(client: EnterpriseClient) -> Result<(), Box<dyn std::error::Error>> {
//! // Join a new node to the cluster
//! let result = client.cluster().join_node(
//!     "192.168.1.100",
//!     "admin",
//!     "password"
//! ).await?;
//! println!("Node joined: {:?}", result);
//!
//! // Remove a node
//! let action = client.cluster().remove_node(3).await?;
//! println!("Removal started: {:?}", action);
//! # Ok(())
//! # }
//! ```

use crate::client::RestClient;
use crate::error::Result;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use typed_builder::TypedBuilder;

/// Response from cluster action operations
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterActionResponse {
    /// The action UID for tracking async operations
    pub action_uid: String,
    /// Description of the action
    pub description: Option<String>,
}

/// Node information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterNode {
    pub id: u32,
    pub address: String,
    pub status: String,
    pub role: Option<String>,
    pub total_memory: Option<u64>,
    pub used_memory: Option<u64>,
    pub cpu_cores: Option<u32>,
}

/// Cluster information from the REST API
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterInfo {
    /// Cluster unique ID (read-only)
    pub uid: Option<u32>,

    /// Cluster's fully qualified domain name (read-only)
    pub name: String,

    /// Cluster creation date (read-only)
    pub created: Option<String>,

    /// Last changed time (read-only)
    pub last_changed_time: Option<String>,

    /// License expiration status
    pub license_expired: Option<bool>,

    /// List of node UIDs in the cluster
    pub nodes: Option<Vec<u32>>,

    /// List of database UIDs in the cluster
    pub databases: Option<Vec<u32>>,

    /// Cluster status
    pub status: Option<String>,

    /// Enables/disables node/cluster email alerts
    pub email_alerts: Option<bool>,

    /// Indicates if cluster operates in rack-aware mode
    pub rack_aware: Option<bool>,

    /// Storage engine for Auto Tiering ('speedb' or 'rocksdb')
    pub bigstore_driver: Option<String>,

    /// API HTTP listening port (range: 1024-65535)
    pub cnm_http_port: Option<u16>,

    /// API HTTPS listening port (range: 1024-65535)
    pub cnm_https_port: Option<u16>,

    // Stats
    /// Total memory available in the cluster
    pub total_memory: Option<u64>,

    /// Total memory used in the cluster
    pub used_memory: Option<u64>,

    /// Total number of shards in the cluster
    pub total_shards: Option<u32>,

    // Additional fields from API audit
    /// Alert settings configuration for cluster and nodes
    pub alert_settings: Option<Value>,

    /// Whether cluster changes are currently blocked (maintenance mode)
    pub block_cluster_changes: Option<bool>,

    /// Whether CCS (Cluster Configuration Store) internode encryption is enabled
    pub ccs_internode_encryption: Option<bool>,

    /// Internal port used by the cluster API
    pub cluster_api_internal_port: Option<u32>,

    /// SSH public key for cluster authentication
    pub cluster_ssh_public_key: Option<String>,

    /// Port used by Cluster Manager (CM)
    pub cm_port: Option<u32>,

    /// Version of the Cluster Manager server
    pub cm_server_version: Option<u32>,

    /// Session timeout for Cluster Manager in minutes
    pub cm_session_timeout_minutes: Option<u32>,

    /// Maximum threads per worker for CNM HTTP server
    pub cnm_http_max_threads_per_worker: Option<u32>,

    /// Number of workers for CNM HTTP server
    pub cnm_http_workers: Option<u32>,

    /// Cipher suites for control plane TLS connections
    pub control_cipher_suites: Option<String>,

    /// Cipher suites for control plane TLS 1.3 connections
    pub control_cipher_suites_tls_1_3: Option<String>,

    /// Whether CRDB coordinator should ignore incoming requests
    pub crdb_coordinator_ignore_requests: Option<bool>,

    /// Port used by CRDB (Conflict-free Replicated Database) coordinator
    pub crdb_coordinator_port: Option<u32>,

    /// Supported CRDT featureset version number
    pub crdt_supported_featureset_version: Option<u32>,

    /// List of supported CRDT protocol versions
    pub crdt_supported_protocol_versions: Option<Vec<String>>,

    /// Timestamp when the cluster was created
    pub created_time: Option<String>,

    /// Cipher list for data plane connections
    pub data_cipher_list: Option<String>,

    /// Cipher suites for data plane TLS 1.3 connections
    pub data_cipher_suites_tls_1_3: Option<Vec<Value>>,

    /// Path to debug information files
    pub debuginfo_path: Option<String>,

    /// Whether private keys should be encrypted
    pub encrypt_pkeys: Option<bool>,

    /// Time-to-live for Entra ID (Azure AD) cache in seconds
    pub entra_id_cache_ttl: Option<u32>,

    /// Admin port for Envoy proxy
    pub envoy_admin_port: Option<u32>,

    /// Whether Envoy external authorization is enabled
    pub envoy_external_authorization: Option<bool>,

    /// Maximum number of downstream connections for Envoy
    pub envoy_max_downstream_connections: Option<u32>,

    /// Port for Envoy management server
    pub envoy_mgmt_server_port: Option<u32>,

    /// Admin port for gossip Envoy proxy
    pub gossip_envoy_admin_port: Option<u32>,

    /// Whether to handle metrics endpoint redirects
    pub handle_metrics_redirects: Option<bool>,

    /// Whether to handle HTTP redirects
    pub handle_redirects: Option<bool>,

    /// Whether HTTP support is enabled (in addition to HTTPS)
    pub http_support: Option<bool>,

    /// Configuration for log rotation
    pub logrotate_settings: Option<Value>,

    /// Whether to mask database credentials in logs
    pub mask_bdb_credentials: Option<bool>,

    /// Type of metrics system in use
    pub metrics_system: Option<u32>,

    /// Minimum TLS version for control plane connections
    #[serde(rename = "min_control_TLS_version")]
    pub min_control_tls_version: Option<String>,

    /// Minimum TLS version for data plane connections
    #[serde(rename = "min_data_TLS_version")]
    pub min_data_tls_version: Option<String>,

    /// Minimum TLS version for sentinel connections
    #[serde(rename = "min_sentinel_TLS_version")]
    pub min_sentinel_tls_version: Option<String>,

    /// Maximum size allowed for module uploads in megabytes
    pub module_upload_max_size_mb: Option<u32>,

    /// List of authorized subject names for mutual TLS authentication
    pub mtls_authorized_subjects: Option<Vec<String>>,

    /// Certificate authentication mode for mutual TLS
    pub mtls_certificate_authentication: Option<bool>,

    /// Validation type for MTLS client certificate subjects
    pub mtls_client_cert_subject_validation_type: Option<String>,

    /// Optimization level for multi-command operations
    pub multi_commands_opt: Option<String>,

    /// Whether HTTP OPTIONS method is forbidden
    pub options_method_forbidden: Option<bool>,

    /// Requirements for password complexity
    pub password_complexity: Option<bool>,

    /// Duration in seconds before passwords expire
    pub password_expiration_duration: Option<u32>,

    /// Algorithm used for hashing passwords
    pub password_hashing_algorithm: Option<String>,

    /// Minimum required length for passwords
    pub password_min_length: Option<u32>,

    /// Certificate used by proxy servers
    pub proxy_certificate: Option<String>,

    /// List of ports reserved for system use
    pub reserved_ports: Option<Vec<u32>>,

    /// Whether robust CRDT syncer mode is enabled
    pub robust_crdt_syncer: Option<bool>,

    /// Whether to verify S3 certificates
    pub s3_certificate_verification: Option<bool>,

    /// Cipher suites for sentinel TLS connections
    pub sentinel_cipher_suites: Option<Vec<String>>,

    /// Cipher suites for sentinel TLS 1.3 connections
    pub sentinel_cipher_suites_tls_1_3: Option<String>,

    /// TLS mode for sentinel connections
    pub sentinel_tls_mode: Option<String>,

    /// Whether slave high availability is enabled
    pub slave_ha: Option<bool>,

    /// Cooldown period for database slave HA in seconds
    pub slave_ha_bdb_cooldown_period: Option<u32>,

    /// General cooldown period for slave HA in seconds
    pub slave_ha_cooldown_period: Option<u32>,

    /// Grace period for slave HA operations in seconds
    pub slave_ha_grace_period: Option<u32>,

    /// Whether slowlog sanitization is supported
    pub slowlog_in_sanitized_support: Option<bool>,

    /// TLS mode for SMTP connections
    pub smtp_tls_mode: Option<String>,

    /// Whether to use TLS for SMTP connections
    pub smtp_use_tls: Option<bool>,

    /// Certificate used by syncer processes
    pub syncer_certificate: Option<String>,

    /// Ports reserved for system processes
    pub system_reserved_ports: Option<Vec<u32>>,

    /// Whether a cluster upgrade is currently in progress
    pub upgrade_in_progress: Option<bool>,

    /// Current upgrade mode for the cluster
    pub upgrade_mode: Option<bool>,

    /// Use external IPv6
    pub use_external_ipv6: Option<bool>,

    /// Use IPv6
    pub use_ipv6: Option<bool>,

    /// Wait command support
    pub wait_command: Option<bool>,
}

/// Cluster-wide settings configuration (57 fields)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterSettings {
    /// Automatic recovery on shard failure
    pub auto_recovery: Option<bool>,

    /// Automatic migration of shards from overbooked nodes
    pub automatic_node_offload: Option<bool>,

    /// BigStore migration thresholds
    pub bigstore_migrate_node_threshold: Option<u32>,
    pub bigstore_migrate_node_threshold_p: Option<u32>,
    pub bigstore_provision_node_threshold: Option<u32>,
    pub bigstore_provision_node_threshold_p: Option<u32>,

    /// Default BigStore version
    pub default_bigstore_version: Option<u32>,

    /// Data internode encryption
    pub data_internode_encryption: Option<bool>,

    /// Database connections auditing
    pub db_conns_auditing: Option<bool>,

    /// Default concurrent restore actions
    pub default_concurrent_restore_actions: Option<u32>,

    /// Default fork evict RAM
    pub default_fork_evict_ram: Option<bool>,

    /// Default proxy policies
    pub default_non_sharded_proxy_policy: Option<String>,
    pub default_sharded_proxy_policy: Option<String>,

    /// OSS cluster defaults
    pub default_oss_cluster: Option<bool>,
    pub default_oss_sharding: Option<bool>,

    /// Default Redis version for new databases
    pub default_provisioned_redis_version: Option<String>,

    /// Recovery settings
    pub default_recovery_wait_time: Option<u32>,

    /// Shards placement strategy
    pub default_shards_placement: Option<String>,

    /// Tracking table settings
    pub default_tracking_table_max_keys_policy: Option<String>,

    /// Additional cluster-wide settings
    pub email_alerts: Option<bool>,
    pub endpoint_rebind_enabled: Option<bool>,
    pub failure_detection_sensitivity: Option<String>,
    pub gossip_envoy_admin_port: Option<u32>,
    pub gossip_envoy_port: Option<u32>,
    pub gossip_envoy_proxy_mode: Option<bool>,
    pub hot_spare: Option<bool>,
    pub max_saved_events_per_type: Option<u32>,
    pub max_simultaneous_backups: Option<u32>,
    pub parallel_shards_upgrade: Option<u32>,
    pub persistent_node_removal: Option<bool>,
    pub rack_aware: Option<bool>,
    pub redis_migrate_node_threshold: Option<String>,
    pub redis_migrate_node_threshold_p: Option<u32>,
    pub redis_provision_node_threshold: Option<String>,
    pub redis_provision_node_threshold_p: Option<u32>,
    pub redis_upgrade_policy: Option<String>,
    pub resp3_default: Option<bool>,
    pub show_internals: Option<bool>,
    pub slave_threads_when_master: Option<bool>,
    pub use_empty_shard_backups: Option<bool>,
}

/// Bootstrap request for creating a new cluster
#[derive(Debug, Serialize, TypedBuilder)]
pub struct BootstrapRequest {
    #[builder(setter(into))]
    pub action: String,
    pub cluster: ClusterBootstrapInfo,
    pub credentials: BootstrapCredentials,
}

/// Cluster information for bootstrap
#[derive(Debug, Serialize, TypedBuilder)]
pub struct ClusterBootstrapInfo {
    #[builder(setter(into))]
    pub name: String,
}

/// Credentials for bootstrap
#[derive(Debug, Serialize, TypedBuilder)]
pub struct BootstrapCredentials {
    #[builder(setter(into))]
    pub username: String,
    #[builder(setter(into))]
    pub password: String,
}

/// Cluster handler for executing cluster commands
pub struct ClusterHandler {
    client: RestClient,
}

impl ClusterHandler {
    pub fn new(client: RestClient) -> Self {
        ClusterHandler { client }
    }

    /// Get cluster information (CLUSTER.INFO)
    pub async fn info(&self) -> Result<ClusterInfo> {
        self.client.get("/v1/cluster").await
    }

    /// Bootstrap a new cluster (CLUSTER.BOOTSTRAP)
    pub async fn bootstrap(&self, request: BootstrapRequest) -> Result<Value> {
        // The bootstrap endpoint returns empty response on success
        // Note: Despite docs saying /v1/bootstrap, the actual endpoint is /v1/bootstrap/create_cluster
        self.client
            .post_bootstrap("/v1/bootstrap/create_cluster", &request)
            .await
    }

    /// Update cluster configuration (CLUSTER.UPDATE)
    pub async fn update(&self, updates: Value) -> Result<Value> {
        self.client.put("/v1/cluster", &updates).await
    }

    /// Get cluster stats (CLUSTER.STATS)
    pub async fn stats(&self) -> Result<Value> {
        self.client.get("/v1/cluster/stats").await
    }

    /// Get cluster nodes (CLUSTER.NODES)
    pub async fn nodes(&self) -> Result<Vec<NodeInfo>> {
        self.client.get("/v1/nodes").await
    }

    /// Get cluster license (CLUSTER.LICENSE)
    pub async fn license(&self) -> Result<LicenseInfo> {
        self.client.get("/v1/license").await
    }

    /// Join node to cluster (CLUSTER.JOIN)
    pub async fn join_node(
        &self,
        node_address: &str,
        username: &str,
        password: &str,
    ) -> Result<Value> {
        let body = serde_json::json!({
            "action": "join_cluster",
            "cluster": {
                "nodes": [node_address]
            },
            "credentials": {
                "username": username,
                "password": password
            }
        });
        self.client.post("/v1/bootstrap/join", &body).await
    }

    /// Remove node from cluster (CLUSTER.REMOVE_NODE)
    pub async fn remove_node(&self, node_uid: u32) -> Result<Value> {
        self.client
            .delete(&format!("/v1/nodes/{}", node_uid))
            .await?;
        Ok(serde_json::json!({"message": format!("Node {} removed", node_uid)}))
    }

    /// Reset cluster to factory defaults (CLUSTER.RESET) - DANGEROUS
    pub async fn reset(&self) -> Result<ClusterActionResponse> {
        self.client
            .post("/v1/cluster/actions/reset", &serde_json::json!({}))
            .await
    }

    // raw variant removed: use reset()

    /// Recover cluster from failure (CLUSTER.RECOVER)
    pub async fn recover(&self) -> Result<ClusterActionResponse> {
        self.client
            .post("/v1/cluster/actions/recover", &serde_json::json!({}))
            .await
    }

    // raw variant removed: use recover()

    /// Get cluster settings (CLUSTER.SETTINGS)
    pub async fn settings(&self) -> Result<Value> {
        self.client.get("/v1/cluster/settings").await
    }

    /// Get cluster topology (CLUSTER.TOPOLOGY)
    pub async fn topology(&self) -> Result<Value> {
        self.client.get("/v1/cluster/topology").await
    }

    /// List available cluster actions - GET /v1/cluster/actions
    pub async fn actions(&self) -> Result<Value> {
        self.client.get("/v1/cluster/actions").await
    }

    /// Get a specific cluster action details - GET /v1/cluster/actions/{action}
    pub async fn action_detail(&self, action: &str) -> Result<Value> {
        self.client
            .get(&format!("/v1/cluster/actions/{}", action))
            .await
    }

    /// Execute a specific cluster action - POST /v1/cluster/actions/{action}
    pub async fn action_execute(&self, action: &str, body: Value) -> Result<Value> {
        self.client
            .post(&format!("/v1/cluster/actions/{}", action), &body)
            .await
    }

    /// Delete a specific cluster action - DELETE /v1/cluster/actions/{action}
    pub async fn action_delete(&self, action: &str) -> Result<()> {
        self.client
            .delete(&format!("/v1/cluster/actions/{}", action))
            .await
    }

    /// Get auditing DB connections - GET /v1/cluster/auditing/db_conns
    pub async fn auditing_db_conns(&self) -> Result<Value> {
        self.client.get("/v1/cluster/auditing/db_conns").await
    }

    /// Update auditing DB connections - PUT /v1/cluster/auditing/db_conns
    pub async fn auditing_db_conns_update(&self, cfg: Value) -> Result<Value> {
        self.client.put("/v1/cluster/auditing/db_conns", &cfg).await
    }

    /// Delete auditing DB connections - DELETE /v1/cluster/auditing/db_conns
    pub async fn auditing_db_conns_delete(&self) -> Result<()> {
        self.client.delete("/v1/cluster/auditing/db_conns").await
    }

    /// List cluster certificates - GET /v1/cluster/certificates
    pub async fn certificates(&self) -> Result<Value> {
        self.client.get("/v1/cluster/certificates").await
    }

    /// Delete a certificate - DELETE /v1/cluster/certificates/{uid}
    pub async fn certificate_delete(&self, uid: u32) -> Result<()> {
        self.client
            .delete(&format!("/v1/cluster/certificates/{}", uid))
            .await
    }

    /// Rotate certificates - POST /v1/cluster/certificates/rotate
    pub async fn certificates_rotate(&self) -> Result<Value> {
        self.client
            .post("/v1/cluster/certificates/rotate", &serde_json::json!({}))
            .await
    }

    /// Update certificate bundle - PUT /v1/cluster/update_cert
    pub async fn update_cert(&self, body: Value) -> Result<Value> {
        self.client.put("/v1/cluster/update_cert", &body).await
    }

    /// Delete LDAP configuration - DELETE /v1/cluster/ldap
    pub async fn ldap_delete(&self) -> Result<()> {
        self.client.delete("/v1/cluster/ldap").await
    }

    /// Get cluster module capabilities - GET /v1/cluster/module-capabilities
    pub async fn module_capabilities(&self) -> Result<Value> {
        self.client.get("/v1/cluster/module-capabilities").await
    }

    /// Get cluster policy - GET /v1/cluster/policy
    pub async fn policy(&self) -> Result<Value> {
        self.client.get("/v1/cluster/policy").await
    }

    /// Update cluster policy - PUT /v1/cluster/policy
    pub async fn policy_update(&self, policy: Value) -> Result<Value> {
        self.client.put("/v1/cluster/policy", &policy).await
    }

    /// Restore default cluster policy - PUT /v1/cluster/policy/restore_default
    pub async fn policy_restore_default(&self) -> Result<Value> {
        self.client
            .put("/v1/cluster/policy/restore_default", &serde_json::json!({}))
            .await
    }

    /// Get services configuration - GET /v1/cluster/services_configuration
    pub async fn services_configuration(&self) -> Result<Value> {
        self.client.get("/v1/cluster/services_configuration").await
    }

    /// Update services configuration - PUT /v1/cluster/services_configuration
    pub async fn services_configuration_update(&self, cfg: Value) -> Result<Value> {
        self.client
            .put("/v1/cluster/services_configuration", &cfg)
            .await
    }

    /// Get witness disk info - GET /v1/cluster/witness_disk
    pub async fn witness_disk(&self) -> Result<Value> {
        self.client.get("/v1/cluster/witness_disk").await
    }

    /// Get specific cluster alert detail - GET /v1/cluster/alerts/{alert}
    pub async fn alert_detail(&self, alert: &str) -> Result<Value> {
        self.client
            .get(&format!("/v1/cluster/alerts/{}", alert))
            .await
    }
}

/// Node information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeInfo {
    pub uid: u32,
    pub address: String,
    pub status: String,
    pub role: Option<String>,
    pub shards: Option<Vec<u32>>,
    pub total_memory: Option<u64>,
    pub used_memory: Option<u64>,
}

/// License information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LicenseInfo {
    pub license_type: Option<String>,
    pub expired: Option<bool>,
    pub expiration_date: Option<String>,
    pub shards_limit: Option<u32>,
    pub features: Option<Vec<String>>,
}