rustfs-audit 1.0.0

//  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.

//! Performance and observability tests for audit system

use rustfs_audit::*;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::time::timeout;

#[tokio::test]
async fn test_audit_system_startup_performance() {
    // Test that audit system starts within reasonable time
    let system = AuditSystem::new();
    let start = Instant::now();

    // Create minimal config for testing
    let config = rustfs_config::server_config::Config(std::collections::HashMap::new());

    // System should start quickly even with empty config
    let _result = timeout(Duration::from_secs(5), system.start(config)).await;
    let elapsed = start.elapsed();

    println!("Audit system startup took: {elapsed:?}");

    // Should complete within 5 seconds
    assert!(elapsed < Duration::from_secs(5), "Startup took too long: {elapsed:?}");

    // Clean up
    let _ = system.close().await;
}

#[tokio::test]
async fn test_concurrent_target_creation() {
    // Test that multiple targets can be created concurrently
    let registry = AuditRegistry::new();

    // Create config with multiple webhook instances
    let mut config = rustfs_config::server_config::Config(std::collections::HashMap::new());
    let mut webhook_section = std::collections::HashMap::new();

    // Create multiple instances for concurrent creation test
    for i in 1..=5 {
        let mut kvs = rustfs_config::server_config::KVS::new();
        kvs.insert("enable".to_string(), "on".to_string());
        kvs.insert("endpoint".to_string(), format!("http://localhost:302{i}/webhook"));
        webhook_section.insert(format!("instance_{i}"), kvs);
    }

    config.0.insert("audit_webhook".to_string(), webhook_section);

    let start = Instant::now();

    // This will fail due to server storage not being initialized, but we can measure timing
    let result = registry.create_audit_targets_from_config(&config).await;
    let elapsed = start.elapsed();

    println!("Concurrent target creation took: {elapsed:?}");

    // Should complete quickly even with multiple targets
    assert!(elapsed < Duration::from_secs(10), "Target creation took too long: {elapsed:?}");

    // Verify it fails with expected error (server not initialized)
    match result {
        Err(AuditError::StorageNotAvailable(_)) => {
            // Expected in test environment
        }
        Err(e) => {
            println!("Unexpected error during concurrent creation: {e}");
        }
        Ok(_) => {
            println!("Unexpected success in test environment");
        }
    }
}

#[tokio::test]
async fn test_audit_log_dispatch_performance() {
    let system = AuditSystem::new();

    // Create minimal config
    let config = rustfs_config::server_config::Config(HashMap::new());
    let start_result = system.start(config).await;
    if start_result.is_err() {
        println!("AuditSystem failed to start: {start_result:?}");
        return; // Alternatively: assert!(false, "AuditSystem failed to start");
    }

    use jiff::Timestamp;
    use rustfs_targets::EventName;
    use serde_json::json;
    use std::collections::HashMap;
    let id = 1;

    let mut req_header = hashbrown::HashMap::new();
    req_header.insert("authorization".to_string(), format!("Bearer test-token-{id}"));
    req_header.insert("content-type".to_string(), "application/octet-stream".to_string());

    let mut resp_header = hashbrown::HashMap::new();
    resp_header.insert("x-response".to_string(), "ok".to_string());

    let mut tags = hashbrown::HashMap::new();
    tags.insert(format!("tag-{id}"), json!("sample"));

    let mut req_query = hashbrown::HashMap::new();
    req_query.insert("id".to_string(), id.to_string());

    let api_details = ApiDetails {
        name: Some("PutObject".to_string()),
        bucket: Some("test-bucket".to_string()),
        object: Some(format!("test-object-{id}")),
        status: Some("success".to_string()),
        status_code: Some(200),
        input_bytes: Some(1024),
        output_bytes: Some(0),
        header_bytes: Some(128),
        time_to_first_byte: Some("1ms".to_string()),
        time_to_first_byte_in_ns: Some("1000000".to_string()),
        time_to_response: Some("2ms".to_string()),
        time_to_response_in_ns: Some("2000000".to_string()),
        ..Default::default()
    };
    // Create sample audit log entry
    let audit_entry = AuditEntry {
        version: "1".to_string(),
        deployment_id: Some(format!("test-deployment-{id}")),
        site_name: Some("test-site".to_string()),
        time: Timestamp::now(),
        event: EventName::ObjectCreatedPut,
        entry_type: Some("object".to_string()),
        trigger: "api".to_string(),
        api: api_details,
        remote_host: Some("127.0.0.1".to_string()),
        request_id: Some(format!("test-request-{id}")),
        user_agent: Some("test-agent".to_string()),
        req_path: Some(format!("/test-bucket/test-object-{id}")),
        req_host: Some("test-host".to_string()),
        req_node: Some("node-1".to_string()),
        req_claims: None,
        req_query: Some(req_query),
        req_header: Some(req_header),
        resp_header: Some(resp_header),
        tags: Some(tags),
        access_key: Some(format!("AKIA{id}")),
        parent_user: Some(format!("parent-{id}")),
        error: None,
    };

    let start = Instant::now();

    // Dispatch audit log against an unstarted system state. Empty config keeps
    // the audit system stopped, so dispatch should fail fast without targets.
    let result = system.dispatch(Arc::new(audit_entry)).await;
    let elapsed = start.elapsed();

    println!("Audit log dispatch took: {elapsed:?}");

    // Should be very fast (sub-millisecond for no targets)
    assert!(elapsed < Duration::from_millis(100), "Dispatch took too long: {elapsed:?}");

    assert!(
        matches!(result, Err(AuditError::NotInitialized(_))),
        "Dispatch on a stopped system should return NotInitialized, got: {result:?}"
    );

    // Clean up
    let _ = system.close().await;
}

#[tokio::test]
async fn test_system_state_transitions() {
    let system = AuditSystem::new();

    // Initial state should be stopped
    assert_eq!(system.get_state().await, rustfs_audit::system::AuditSystemState::Stopped);

    // Start system
    let config = rustfs_config::server_config::Config(std::collections::HashMap::new());
    let start_result = system.start(config).await;

    // Empty config keeps the audit system stopped even when start() succeeds.
    let state = system.get_state().await;
    match start_result {
        Ok(_) => {
            assert_eq!(state, rustfs_audit::system::AuditSystemState::Stopped);
        }
        Err(_) => {
            // Expected in test environment due to server storage not being initialized
            assert_eq!(state, rustfs_audit::system::AuditSystemState::Stopped);
        }
    }

    // Clean up
    let _ = system.close().await;
    assert_eq!(system.get_state().await, rustfs_audit::system::AuditSystemState::Stopped);
}

#[test]
fn test_event_name_mask_performance() {
    use rustfs_targets::EventName;

    // Test that event name mask calculation is efficient
    let events = vec![
        EventName::ObjectCreatedPut,
        EventName::ObjectAccessedGet,
        EventName::ObjectRemovedDelete,
        EventName::ObjectCreatedAll,
        EventName::Everything,
    ];

    let start = Instant::now();

    // Calculate masks for many events
    for _ in 0..1000 {
        for event in &events {
            let _mask = event.mask();
        }
    }

    let elapsed = start.elapsed();
    println!("Event mask calculation (5000 ops) took: {elapsed:?}");

    // Should be very fast
    assert!(elapsed < Duration::from_millis(100), "Mask calculation too slow: {elapsed:?}");
}

#[test]
fn test_event_name_expansion_performance() {
    use rustfs_targets::EventName;

    // Test that event name expansion is efficient
    let compound_events = vec![
        EventName::ObjectCreatedAll,
        EventName::ObjectAccessedAll,
        EventName::ObjectRemovedAll,
        EventName::Everything,
    ];

    let start = Instant::now();

    // Expand events many times
    for _ in 0..1000 {
        for event in &compound_events {
            let _expanded = event.expand();
        }
    }

    let elapsed = start.elapsed();
    println!("Event expansion (4000 ops) took: {elapsed:?}");

    // Should be very fast
    assert!(elapsed < Duration::from_millis(100), "Expansion too slow: {elapsed:?}");
}

#[tokio::test]
async fn test_registry_operations_performance() {
    let registry = AuditRegistry::new();

    let start = Instant::now();

    // Test basic registry operations
    for _ in 0..1000 {
        let targets = registry.list_targets();
        let _target = registry.get_target("nonexistent");
        assert!(targets.is_empty());
    }

    let elapsed = start.elapsed();
    println!("Registry operations (2000 ops) took: {elapsed:?}");

    // Should be very fast for empty registry
    assert!(elapsed < Duration::from_millis(100), "Registry ops too slow: {elapsed:?}");
}

// Performance requirements validation
#[test]
fn test_performance_requirements() {
    // According to requirements: ≥ 3k EPS/node; P99 < 30ms (default)

    // These are synthetic tests since we can't actually achieve 3k EPS
    // without real server storage and network targets, but we can validate
    // that our core algorithms are efficient enough

    let start = Instant::now();

    // Simulate processing 3000 events worth of operations
    for i in 0..3000 {
        // Simulate event name parsing and processing
        let _event_id = format!("s3:ObjectCreated:Put_{i}");
        let _timestamp = jiff::Timestamp::now().to_string();

        // Simulate basic audit entry creation overhead
        let _entry_size = 512; // bytes
        let _processing_time = std::time::Duration::from_nanos(100); // simulated
    }

    let elapsed = start.elapsed();
    let eps = 3000.0 / elapsed.as_secs_f64();

    println!("Simulated 3000 events in {elapsed:?} ({eps:.0} EPS)");

    // Our core processing should easily handle 3k EPS worth of CPU overhead
    // The actual EPS limit will be determined by network I/O to targets
    assert!(eps > 10000.0, "Core processing too slow for 3k EPS target: {eps} EPS");

    // P99 latency requirement: < 30ms
    // For core processing, we should be much faster than this
    let avg_latency = elapsed / 3000;
    println!("Average processing latency: {avg_latency:?}");

    assert!(avg_latency < Duration::from_millis(1), "Processing latency too high: {avg_latency:?}");
}