rustfs/crates/audit/src/system.rs

//  Copyright 2024 RustFS Team
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.

use crate::{
    AuditEntry, AuditError, AuditRegistry, AuditResult, observability,
    pipeline::{AuditPipeline, AuditRuntimeFacade, AuditRuntimeView},
};
use rustfs_config::server_config::Config;
use rustfs_targets::{ReplayWorkerManager, Target};
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};
use tracing::{debug, error, info, warn};

const LOG_COMPONENT_AUDIT: &str = "audit";
const LOG_SUBSYSTEM_SYSTEM: &str = "system";
const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state";
const EVENT_AUDIT_CONFIG_RELOADED: &str = "audit_config_reloaded";

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AuditTargetMetricSnapshot {
    pub failed_messages: u64,
    pub queue_length: u64,
    pub target_id: String,
    pub total_messages: u64,
}

/// State of the audit system
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AuditSystemState {
    Stopped,
    Starting,
    Running,
    Paused,
    Stopping,
}

/// Main audit system that manages target lifecycle and audit log dispatch
#[derive(Clone)]
pub struct AuditSystem {
    registry: Arc<Mutex<AuditRegistry>>,
    state: Arc<RwLock<AuditSystemState>>,
    config: Arc<RwLock<Option<Config>>>,
    /// Cancellation senders for active audit stream tasks (target_id -> cancel tx)
    stream_cancellers: Arc<RwLock<ReplayWorkerManager>>,
}

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

impl AuditSystem {
    fn pipeline(&self) -> AuditPipeline {
        AuditPipeline::new(self.registry.clone())
    }

    fn runtime_view(&self) -> AuditRuntimeView {
        AuditRuntimeView::new(self.registry.clone())
    }

    fn runtime_facade(&self) -> AuditRuntimeFacade {
        AuditRuntimeFacade::new(self.registry.clone(), self.stream_cancellers.clone())
    }

    /// Creates a new audit system
    pub fn new() -> Self {
        Self {
            registry: Arc::new(Mutex::new(AuditRegistry::new())),
            state: Arc::new(RwLock::new(AuditSystemState::Stopped)),
            config: Arc::new(RwLock::new(None)),
            stream_cancellers: Arc::new(RwLock::new(ReplayWorkerManager::new())),
        }
    }

    async fn create_targets_from_config(&self, config: &Config) -> AuditResult<Vec<Box<dyn Target<AuditEntry> + Send + Sync>>> {
        let registry = self.registry.lock().await;
        registry.create_audit_targets_from_config(config).await
    }

    async fn clear_runtime_targets(&self) -> AuditResult<()> {
        {
            let mut registry = self.registry.lock().await;
            let mut replay_workers = self.stream_cancellers.write().await;
            self.runtime_facade()
                .shutdown_runtime(&mut registry, &mut replay_workers)
                .await?;
        }

        let mut state = self.state.write().await;
        *state = AuditSystemState::Stopped;
        Ok(())
    }

    async fn commit_runtime_targets(
        &self,
        targets: Vec<Box<dyn Target<AuditEntry> + Send + Sync>>,
        final_state: AuditSystemState,
    ) -> AuditResult<()> {
        if targets.is_empty() {
            debug_audit_state("stopped", Some("no_enabled_targets"), None, 0);
            self.clear_runtime_targets().await?;
            return Ok(());
        }

        info!(
            event = EVENT_AUDIT_SYSTEM_STATE,
            component = LOG_COMPONENT_AUDIT,
            subsystem = LOG_SUBSYSTEM_SYSTEM,
            state = "targets_created",
            target_count = targets.len(),
            "audit system state"
        );

        let activation = self.runtime_facade().activate_targets_with_replay(targets).await;
        self.runtime_facade().replace_targets(activation).await?;

        let mut state = self.state.write().await;
        *state = final_state;
        Ok(())
    }

    /// Starts the audit system with the given configuration
    ///
    /// # Arguments
    /// * `config` - The configuration to use for starting the audit system
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn start(&self, config: Config) -> AuditResult<()> {
        let state = self.state.write().await;

        match *state {
            AuditSystemState::Running => {
                return Err(AuditError::AlreadyInitialized);
            }
            AuditSystemState::Starting => {
                warn_audit_state("starting", Some("already_starting"));
                return Ok(());
            }
            _ => {}
        }

        drop(state);

        info!(
            event = EVENT_AUDIT_SYSTEM_STATE,
            component = LOG_COMPONENT_AUDIT,
            subsystem = LOG_SUBSYSTEM_SYSTEM,
            state = "starting",
            "audit system state"
        );

        // Record system start
        observability::record_system_start();

        // Store configuration
        {
            let mut config_guard = self.config.write().await;
            *config_guard = Some(config.clone());
        }

        match self.create_targets_from_config(&config).await {
            Ok(targets) => {
                {
                    let mut state = self.state.write().await;
                    *state = AuditSystemState::Starting;
                }

                self.commit_runtime_targets(targets, AuditSystemState::Running).await?;
                info_audit_state("running", None, None);
                Ok(())
            }
            Err(e) => {
                error!(
                    event = EVENT_AUDIT_SYSTEM_STATE,
                    component = LOG_COMPONENT_AUDIT,
                    subsystem = LOG_SUBSYSTEM_SYSTEM,
                    state = "stopped",
                    reason = "target_creation_failed",
                    error = %e,
                    "Failed to create audit targets"
                );
                let mut state = self.state.write().await;
                *state = AuditSystemState::Stopped;
                Err(e)
            }
        }
    }

    /// Pauses the audit system
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn pause(&self) -> AuditResult<()> {
        let mut state = self.state.write().await;

        match *state {
            AuditSystemState::Running => {
                *state = AuditSystemState::Paused;
                info_audit_state("paused", None, None);
                Ok(())
            }
            AuditSystemState::Paused => {
                warn_audit_state("paused", Some("already_paused"));
                Ok(())
            }
            _ => Err(AuditError::Configuration("Cannot pause audit system in current state".to_string(), None)),
        }
    }

    /// Resumes the audit system
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn resume(&self) -> AuditResult<()> {
        let mut state = self.state.write().await;

        match *state {
            AuditSystemState::Paused => {
                *state = AuditSystemState::Running;
                info_audit_state("running", Some("resumed"), None);
                Ok(())
            }
            AuditSystemState::Running => {
                warn_audit_state("running", Some("already_running"));
                Ok(())
            }
            _ => Err(AuditError::Configuration("Cannot resume audit system in current state".to_string(), None)),
        }
    }

    /// Stops the audit system and closes all targets
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn close(&self) -> AuditResult<()> {
        let mut state = self.state.write().await;

        match *state {
            AuditSystemState::Stopped => {
                warn_audit_state("stopped", Some("already_stopped"));
                return Ok(());
            }
            AuditSystemState::Stopping => {
                warn_audit_state("stopping", Some("already_stopping"));
                return Ok(());
            }
            _ => {}
        }

        *state = AuditSystemState::Stopping;
        drop(state);

        info!(
            event = EVENT_AUDIT_SYSTEM_STATE,
            component = LOG_COMPONENT_AUDIT,
            subsystem = LOG_SUBSYSTEM_SYSTEM,
            state = "stopping",
            "audit system state"
        );

        // Stop all stream tasks first
        if let Err(e) = self.clear_runtime_targets().await {
            error!(
                event = EVENT_AUDIT_SYSTEM_STATE,
                component = LOG_COMPONENT_AUDIT,
                subsystem = LOG_SUBSYSTEM_SYSTEM,
                state = "stopping",
                reason = "target_shutdown_failed",
                error = %e,
                "Failed to close some audit targets"
            );
        }

        // Clear configuration
        let mut config_guard = self.config.write().await;
        *config_guard = None;

        info_audit_state("stopped", None, None);
        Ok(())
    }

    /// Gets the current state of the audit system
    pub async fn get_state(&self) -> AuditSystemState {
        self.state.read().await.clone()
    }

    /// Checks if the audit system is running
    ///
    /// # Returns
    /// * `bool` - True if running, false otherwise
    pub async fn is_running(&self) -> bool {
        matches!(*self.state.read().await, AuditSystemState::Running)
    }

    /// Dispatches an audit log entry to all active targets
    ///
    /// # Arguments
    /// * `entry` - The audit log entry to dispatch
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn dispatch(&self, entry: Arc<AuditEntry>) -> AuditResult<()> {
        let state = self.state.read().await;

        match *state {
            AuditSystemState::Running => {}
            AuditSystemState::Paused => {
                return Ok(());
            }
            _ => {
                return Err(AuditError::NotInitialized("Audit system is not running".to_string()));
            }
        }
        drop(state);
        self.pipeline().dispatch(entry).await
    }

    /// Dispatches a batch of audit log entries to all active targets
    ///
    /// # Arguments
    /// * `entries` - A vector of audit log entries to dispatch
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn dispatch_batch(&self, entries: Vec<Arc<AuditEntry>>) -> AuditResult<()> {
        let state = self.state.read().await;
        if *state != AuditSystemState::Running {
            return Err(AuditError::NotInitialized("Audit system is not running".to_string()));
        }
        drop(state);
        self.pipeline().dispatch_batch(entries).await
    }

    /// Enables a specific target
    ///
    /// # Arguments
    /// * `target_id` - The ID of the target to enable, TargetID to string
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn enable_target(&self, target_id: &str) -> AuditResult<()> {
        self.runtime_view().enable_target(target_id).await
    }

    /// Disables a specific target
    ///
    /// # Arguments
    /// * `target_id` - The ID of the target to disable, TargetID to string
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn disable_target(&self, target_id: &str) -> AuditResult<()> {
        self.runtime_view().disable_target(target_id).await
    }

    /// Removes a target from the system
    ///
    /// # Arguments
    /// * `target_id` - The ID of the target to remove, TargetID to string
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn remove_target(&self, target_id: &str) -> AuditResult<()> {
        self.runtime_view().remove_target(target_id).await
    }

    /// Updates or inserts a target
    ///
    /// # Arguments
    /// * `target_id` - The ID of the target to upsert, TargetID to string
    /// * `target` - The target instance to insert or update
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn upsert_target(&self, target_id: String, target: Box<dyn Target<AuditEntry> + Send + Sync>) -> AuditResult<()> {
        self.runtime_view().upsert_target(target_id, target).await
    }

    /// Lists all targets
    ///
    /// # Returns
    /// * `Vec<String>` - List of target IDs
    pub async fn list_targets(&self) -> Vec<String> {
        self.runtime_view().list_targets().await
    }

    /// Returns cloned target values for read-only runtime inspection.
    pub async fn get_target_values(&self) -> Vec<rustfs_targets::SharedTarget<AuditEntry>> {
        self.runtime_view().get_target_values().await
    }

    /// Returns per-target delivery metrics for Prometheus collection.
    pub async fn snapshot_target_metrics(&self) -> Vec<AuditTargetMetricSnapshot> {
        self.pipeline().snapshot_target_metrics().await
    }

    pub async fn snapshot_target_health(&self) -> Vec<rustfs_targets::RuntimeTargetHealthSnapshot> {
        self.pipeline().snapshot_target_health().await
    }

    pub async fn runtime_status_snapshot(&self) -> rustfs_targets::RuntimeStatusSnapshot {
        // Lock order must match every other path that holds both locks
        // (`clear_runtime_targets`, `AuditRuntimeFacade::replace_targets`):
        // acquire `registry` first, then `stream_cancellers`. Reversing the
        // order here would create an ABBA deadlock with those paths.
        let registry = self.registry.lock().await;
        let replay_workers = self.stream_cancellers.read().await;
        registry.runtime_manager().status_snapshot(&replay_workers)
    }

    /// Gets information about a specific target
    ///
    /// # Arguments
    /// * `target_id` - The ID of the target to retrieve, TargetID to string
    ///
    /// # Returns
    /// * `Option<String>` - Target ID if found
    pub async fn get_target(&self, target_id: &str) -> Option<String> {
        self.runtime_view().get_target(target_id).await
    }

    /// Reloads configuration and updates targets
    ///
    /// # Arguments
    /// * `new_config` - The new configuration to load
    ///
    /// # Returns
    /// * `AuditResult<()>` - Result indicating success or failure
    pub async fn reload_config(&self, new_config: Config) -> AuditResult<()> {
        info!(
            event = EVENT_AUDIT_CONFIG_RELOADED,
            component = LOG_COMPONENT_AUDIT,
            subsystem = LOG_SUBSYSTEM_SYSTEM,
            state = "reloading",
            "audit config reload"
        );

        observability::record_config_reload();

        // Store new configuration
        {
            let mut config_guard = self.config.write().await;
            *config_guard = Some(new_config.clone());
        }

        let final_state = match self.get_state().await {
            AuditSystemState::Paused => AuditSystemState::Paused,
            _ => AuditSystemState::Running,
        };

        match self.create_targets_from_config(&new_config).await {
            Ok(targets) => {
                self.commit_runtime_targets(targets, final_state).await?;
                info!(
                    event = EVENT_AUDIT_CONFIG_RELOADED,
                    component = LOG_COMPONENT_AUDIT,
                    subsystem = LOG_SUBSYSTEM_SYSTEM,
                    state = "reloaded",
                    "audit config reload"
                );
                Ok(())
            }
            Err(e) => {
                error!(
                    event = EVENT_AUDIT_CONFIG_RELOADED,
                    component = LOG_COMPONENT_AUDIT,
                    subsystem = LOG_SUBSYSTEM_SYSTEM,
                    state = "reload_failed",
                    error = %e,
                    "Failed to reload audit configuration"
                );
                Err(e)
            }
        }
    }

    /// Gets current audit system metrics
    ///
    /// # Returns
    /// * `AuditMetricsReport` - Current metrics report
    pub async fn get_metrics(&self) -> observability::AuditMetricsReport {
        observability::get_metrics_report().await
    }

    /// Validates system performance against requirements
    ///
    /// # Returns
    /// * `PerformanceValidation` - Performance validation results
    pub async fn validate_performance(&self) -> observability::PerformanceValidation {
        observability::validate_performance().await
    }

    /// Resets all metrics to initial state
    pub async fn reset_metrics(&self) {
        observability::reset_metrics().await;
    }
}

fn info_audit_state(state: &str, reason: Option<&str>, target_count: Option<usize>) {
    info!(
        event = EVENT_AUDIT_SYSTEM_STATE,
        component = LOG_COMPONENT_AUDIT,
        subsystem = LOG_SUBSYSTEM_SYSTEM,
        state,
        reason = reason.unwrap_or_default(),
        target_count = target_count.unwrap_or_default(),
        "audit system state"
    );
}

fn debug_audit_state(state: &str, reason: Option<&str>, error: Option<&str>, target_count: usize) {
    debug!(
        event = EVENT_AUDIT_SYSTEM_STATE,
        component = LOG_COMPONENT_AUDIT,
        subsystem = LOG_SUBSYSTEM_SYSTEM,
        state,
        reason = reason.unwrap_or_default(),
        error = error.unwrap_or_default(),
        target_count,
        "audit system state"
    );
}

fn warn_audit_state(state: &str, reason: Option<&str>) {
    warn!(
        event = EVENT_AUDIT_SYSTEM_STATE,
        component = LOG_COMPONENT_AUDIT,
        subsystem = LOG_SUBSYSTEM_SYSTEM,
        state,
        reason = reason.unwrap_or_default(),
        "audit system state"
    );
}

#[cfg(test)]
mod tests {
    use super::{AuditSystem, AuditSystemState};
    use async_trait::async_trait;
    use rustfs_targets::ReplayWorkerManager;
    use rustfs_targets::arn::TargetID;
    use rustfs_targets::store::{Key, Store};
    use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta};
    use rustfs_targets::{StoreError, Target, TargetError};
    use std::collections::HashMap;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tokio::sync::mpsc;

    #[derive(Clone)]
    struct TestTarget {
        close_calls: Arc<AtomicUsize>,
        id: TargetID,
    }

    impl TestTarget {
        fn new(id: &str, name: &str) -> Self {
            Self {
                close_calls: Arc::new(AtomicUsize::new(0)),
                id: TargetID::new(id.to_string(), name.to_string()),
            }
        }
    }

    #[async_trait]
    impl<E> Target<E> for TestTarget
    where
        E: rustfs_targets::PluginEvent,
    {
        fn id(&self) -> TargetID {
            self.id.clone()
        }

        async fn is_active(&self) -> Result<bool, TargetError> {
            Ok(true)
        }

        async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
            Ok(())
        }

        async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
            Ok(())
        }

        async fn close(&self) -> Result<(), TargetError> {
            self.close_calls.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }

        fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
            None
        }

        fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
            Box::new(self.clone())
        }

        fn is_enabled(&self) -> bool {
            true
        }
    }

    #[tokio::test]
    async fn reload_with_empty_config_stops_existing_runtime() {
        let system = AuditSystem::new();
        let target = TestTarget::new("primary", "webhook");
        let close_calls = Arc::clone(&target.close_calls);

        {
            let mut registry = system.registry.lock().await;
            registry.add_target("primary:webhook".to_string(), Box::new(target));
        }
        {
            let mut state = system.state.write().await;
            *state = AuditSystemState::Running;
        }
        {
            let mut replay_workers = system.stream_cancellers.write().await;
            let (cancel_tx, _cancel_rx) = mpsc::channel(1);
            replay_workers.insert("primary:webhook".to_string(), cancel_tx);
            assert_eq!(replay_workers.len(), 1);
        }

        system
            .reload_config(rustfs_config::server_config::Config(HashMap::new()))
            .await
            .expect("reload with empty config should succeed");

        assert_eq!(system.get_state().await, AuditSystemState::Stopped);
        assert!(system.list_targets().await.is_empty());
        assert_eq!(system.runtime_status_snapshot().await, ReplayWorkerManager::new().snapshot(0));
        assert_eq!(close_calls.load(Ordering::SeqCst), 1);
        assert_eq!(*system.config.read().await, Some(rustfs_config::server_config::Config(HashMap::new())));
    }

    /// Regression guard for backlog#961: `runtime_status_snapshot` and
    /// `clear_runtime_targets` both hold `registry` and `stream_cancellers`.
    /// They previously acquired the two locks in opposite orders (ABBA),
    /// which could deadlock the whole audit control plane under concurrency.
    /// Hammer both paths from multiple worker threads and assert the workload
    /// completes within a timeout instead of hanging.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn concurrent_status_and_clear_do_not_deadlock() {
        use std::time::Duration;

        const ITERATIONS: usize = 2_000;
        const TASKS_PER_PATH: usize = 4;

        let system = AuditSystem::new();

        // Seed a target + replay worker so both critical sections touch real state.
        {
            let mut registry = system.registry.lock().await;
            registry.add_target("primary:webhook".to_string(), Box::new(TestTarget::new("primary", "webhook")));
        }
        {
            let mut replay_workers = system.stream_cancellers.write().await;
            let (cancel_tx, _cancel_rx) = mpsc::channel(1);
            replay_workers.insert("primary:webhook".to_string(), cancel_tx);
        }

        let mut handles = Vec::new();

        for _ in 0..TASKS_PER_PATH {
            let status_system = system.clone();
            handles.push(tokio::spawn(async move {
                for _ in 0..ITERATIONS {
                    // registry -> stream_cancellers (read)
                    let _ = status_system.runtime_status_snapshot().await;
                }
            }));

            let clear_system = system.clone();
            handles.push(tokio::spawn(async move {
                for _ in 0..ITERATIONS {
                    // registry -> stream_cancellers (write)
                    clear_system
                        .clear_runtime_targets()
                        .await
                        .expect("clear_runtime_targets should succeed");
                }
            }));
        }

        let workload = async {
            for handle in handles {
                handle.await.expect("worker task panicked");
            }
        };

        tokio::time::timeout(Duration::from_secs(30), workload)
            .await
            .expect("audit lock paths deadlocked (backlog#961 regression)");
    }
}