hirai 0.1.1

//! # Integration Tests for Hirai
//!
//! This module contains integration tests for the Hirai application, covering configuration loading, file event broadcasting/receiving, CLI error handling, WebSocket event delivery, and high-volume event stress testing.
//!
//! ## Test Overview
//!
//! - **test_config_loading_defaults_and_cli_override**: Verifies config merging from defaults and CLI.
//! - **test_file_operations_broadcast_and_receive**: End-to-end test for file event propagation via UDP multicast.
//! - **test_broadcaster_fails_with_no_folders**: Ensures broadcaster fails gracefully with no folders.
//! - **test_cli_fails_with_no_folders**: Ensures CLI returns error when no folders are specified.
//! - **test_websocket_receives_events**: Checks that file events are delivered over WebSocket.
//! - **test_high_event_volume_stress**: Stress test for high event throughput and UDP loss tolerance.

use clap::Parser;
use figment::{providers::Serialized, Figment};
use futures_util::StreamExt;
use hirai::config::{AppConfig, CliArgs, FileConfig};
use hirai::event::Event;
use std::fs::{self, File};
use std::io::Write;
use std::sync::Arc;
use tempfile;
use tokio::sync::{mpsc, watch};
use tokio::time::{timeout, Duration};
use tracing::warn;

/// Default multicast address for tests.
const TEST_MULTICAST_ADDR: &str = "239.255.255.1:30001";
/// Default web address for tests.
const TEST_WEB_ADDR: &str = "0.0.0.0:30002";
/// Default timeout for test shutdowns.
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(15);
/// Short timeout for event waits.
const SHORT_TIMEOUT: Duration = Duration::from_secs(5);

/// Helper to create an `AppConfig` for tests.
///
/// # Arguments
/// - `folders`: Folders to watch.
/// - `listen`: Whether to run in listener mode.
/// - `browse`: Whether to enable web browsing.
/// - `multicast_addr`: Optional multicast address override.
/// - `web_addr`: Optional web address override.
fn create_test_config(
    folders: Vec<String>,
    listen: bool,
    browse: bool,
    multicast_addr: Option<String>,
    web_addr: Option<String>,
) -> Arc<AppConfig> {
    Arc::new(AppConfig {
        listen,
        browse,
        multicast_addr: multicast_addr.unwrap_or_else(|| TEST_MULTICAST_ADDR.to_string()),
        web_addr: web_addr.unwrap_or_else(|| TEST_WEB_ADDR.to_string()),
        folders_to_watch: folders,
        log_level: "trace".to_string(),
    })
}

/// Spawns a listener task for integration tests.
///
/// # Arguments
/// - `config`: Listener configuration.
/// - `event_tx`: Channel to send received events.
/// - `shutdown_rx`: Shutdown signal receiver.
async fn setup_listener(
    config: Arc<AppConfig>,
    event_tx: mpsc::Sender<Event>,
    shutdown_rx: watch::Receiver<bool>,
) -> tokio::task::JoinHandle<()> {
    let listener_event_tx_for_web = if config.browse { Some(event_tx) } else { None };
    tokio::spawn(async move {
        if let Err(e) =
            hirai::network::run_listener(config, listener_event_tx_for_web, shutdown_rx).await
        {
            eprintln!("[Test Listener] Error: {}", e);
        }
        println!("[Test Listener] Exited.");
    })
}

/// Spawns a broadcaster task for integration tests.
///
/// # Arguments
/// - `config`: Broadcaster configuration.
/// - `internal_event_rx`: Channel to receive events to broadcast.
/// - `shutdown_rx`: Shutdown signal receiver.
async fn setup_broadcaster(
    config: Arc<AppConfig>,
    internal_event_rx: mpsc::Receiver<Event>,
    shutdown_rx: watch::Receiver<bool>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        if let Err(e) =
            hirai::network::run_broadcaster(config, internal_event_rx, shutdown_rx).await
        {
            eprintln!("[Test Broadcaster] Error: {}", e);
        }
        println!("[Test Broadcaster] Exited.");
    })
}

/// Runs the watcher for integration tests.
///
/// # Arguments
/// - `config`: Watcher configuration.
/// - `event_tx`: Channel to send detected events.
async fn setup_watcher(config: Arc<AppConfig>, event_tx: mpsc::Sender<Event>) {
    if let Err(e) = hirai::watcher::run_watcher(config, event_tx).await {
        eprintln!("[Test Setup Watcher] Error calling run_watcher: {}", e);
    }
}

/// Test: Configuration loading and CLI override.
///
/// Ensures that CLI arguments override config file defaults.
#[tokio::test]
async fn test_config_loading_defaults_and_cli_override() {
    let cli_args = CliArgs::parse_from([
        "hirai",
        "--multicast",
        "239.1.1.1:11111",
        "/tmp/test1",
        "/tmp/test2",
    ]);

    let fig = Figment::new().merge(Serialized::defaults(FileConfig {
        multicast: Some("239.0.0.1:9999".to_string()),
        webaddr: Some("0.0.0.0:8080".to_string()),
        listen: Some(false),
        browse: Some(false),
        log_level: Some("info".to_string()),
        folders: Some(vec!["/default/folder".to_string()]),
    }));

    let mut final_fig = fig;
    if let Some(mc) = cli_args.multicast {
        final_fig = final_fig.merge(Serialized::globals(FileConfig {
            multicast: Some(mc),
            ..Default::default()
        }));
    }

    let merged_config: FileConfig = final_fig
        .extract()
        .expect("Failed to extract merged config");

    let app_config = AppConfig {
        listen: cli_args.listen || merged_config.listen.unwrap_or(false),
        browse: cli_args.browse || merged_config.browse.unwrap_or(false),
        multicast_addr: merged_config
            .multicast
            .unwrap_or_else(|| "239.0.0.1:9999".to_string()),
        web_addr: merged_config
            .webaddr
            .unwrap_or_else(|| "0.0.0.0:8080".to_string()),
        folders_to_watch: if !cli_args.folders.is_empty() {
            cli_args.folders
        } else {
            merged_config.folders.unwrap_or_default()
        },
        log_level: merged_config
            .log_level
            .unwrap_or_else(|| "info".to_string()),
    };

    assert_eq!(app_config.multicast_addr, "239.1.1.1:11111");
    assert_eq!(
        app_config.folders_to_watch,
        vec!["/tmp/test1", "/tmp/test2"]
    );
    assert_eq!(app_config.listen, false);
}

/// Test: File operations are broadcast and received.
///
/// This test performs file operations and checks that events are received by a listener.
#[tokio::test]
async fn test_file_operations_broadcast_and_receive() {
    // Initialize tracing for tests, allowing RUST_LOG=trace to work if set
    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::from_default_env()
                .add_directive("hirai=trace".parse().unwrap()),
        )
        .try_init();

    let home_dir = std::env::var("HOME").expect("HOME env var not set");
    let watched_folder = std::path::Path::new(&home_dir).join("test_hirai_watch");
    if watched_folder.exists() {
        std::fs::remove_dir_all(&watched_folder).expect("Failed to clean up old test dir");
    }
    std::fs::create_dir(&watched_folder).expect("Failed to create test dir");
    let test_file_path = watched_folder.join("test_file.txt");

    // Listener will have browse = true to send events to main_to_listener_tx
    let broadcaster_config = create_test_config(
        vec![watched_folder.to_string_lossy().to_string()],
        false,
        false,
        None,
        None,
    );
    let listener_config = create_test_config(vec![], true, true, None, None); // browse = true

    let (shutdown_tx, shutdown_rx_listener) = watch::channel(false);
    let shutdown_rx_broadcaster = shutdown_tx.subscribe();

    let (watcher_to_broadcaster_tx, watcher_to_broadcaster_rx) = mpsc::channel::<Event>(10);
    let (listener_received_event_tx, mut listener_received_event_rx) = mpsc::channel::<Event>(10);

    // Start Listener
    let listener_handle = setup_listener(
        listener_config,
        listener_received_event_tx,
        shutdown_rx_listener,
    )
    .await;
    tokio::time::sleep(Duration::from_millis(500)).await; // Give listener time to bind

    // Start Broadcaster (receives from watcher_to_broadcaster_rx)
    let broadcaster_handle = setup_broadcaster(
        broadcaster_config.clone(),
        watcher_to_broadcaster_rx,
        shutdown_rx_broadcaster,
    )
    .await;
    tokio::time::sleep(Duration::from_millis(200)).await; // Give broadcaster time to start

    // Start Watcher (sends to watcher_to_broadcaster_tx)
    setup_watcher(broadcaster_config, watcher_to_broadcaster_tx.clone()).await;
    tokio::time::sleep(Duration::from_millis(1500)).await; // Give watcher time to initialize (debouncer is 1s)

    println!("[Test] Performing file operations...");
    File::create(&test_file_path)
        .expect("Failed to create file")
        .write_all(b"hello")
        .expect("Write failed");
    println!("[Test] Created file: {:?}", test_file_path);
    tokio::time::sleep(Duration::from_secs(2)).await; // Wait for debouncer (1s) + broadcast

    fs::write(&test_file_path, b"world").expect("Failed to write to file");
    println!("[Test] Wrote to file: {:?}", test_file_path);
    tokio::time::sleep(Duration::from_secs(2)).await;

    fs::rename(
        &test_file_path,
        &watched_folder.join("test_file_renamed.txt"),
    )
    .expect("Failed to rename file");
    println!(
        "[Test] Renamed file: {:?}",
        watched_folder.join("test_file_renamed.txt")
    );
    tokio::time::sleep(Duration::from_secs(2)).await;

    fs::remove_file(&watched_folder.join("test_file_renamed.txt")).expect("Failed to remove file");
    println!(
        "[Test] Removed file: {:?}",
        watched_folder.join("test_file_renamed.txt")
    );
    tokio::time::sleep(Duration::from_secs(3)).await; // Increased from 2 to 3 seconds to ensure debouncer processes REMOVE

    let mut received_events_ops = Vec::new();
    let mut expected_ops = vec![
        "CREATE".to_string(),
        "WRITE".to_string(),
        "RENAME".to_string(),
        "REMOVE".to_string(),
    ];
    println!("[Test] Waiting for events...");

    for _ in 0..expected_ops.len() {
        // Try to receive expected number of events
        match timeout(SHORT_TIMEOUT, listener_received_event_rx.recv()).await {
            Ok(Some(event)) => {
                println!(
                    "[Test] Listener received event: Op: {}, Path: {}",
                    event.op, event.path
                );
                // Normalize paths by removing /private prefix if present
                let normalized_event_path = event.path.replace("/private", "");
                let normalized_test_path = test_file_path.to_string_lossy().to_string();

                if normalized_event_path == normalized_test_path {
                    if let Some(pos) = expected_ops.iter().position(|x| *x == event.op) {
                        received_events_ops.push(event.op.clone());
                        expected_ops.remove(pos);
                        println!(
                            "[Test] Matched expected event: {}. Remaining: {:?}",
                            event.op, expected_ops
                        );
                    } else {
                        println!("[Test] Received event for correct path, but unexpected op: {} (already received or not in list?)", event.op);
                    }
                } else {
                    println!(
                        "[Test] Received event for unexpected path: {} (expected: {})",
                        normalized_event_path, normalized_test_path
                    );
                }
            }
            Ok(None) => {
                println!("[Test] Listener event channel closed unexpectedly.");
                break;
            }
            Err(_) => {
                println!(
                    "[Test] Timeout waiting for an event. Expected remaining: {:?}",
                    expected_ops
                );
                break;
            }
        }
    }

    println!("[Test] Shutting down test components...");
    shutdown_tx.send(true).unwrap();
    let _ = timeout(DEFAULT_TIMEOUT, listener_handle).await;
    let _ = timeout(DEFAULT_TIMEOUT, broadcaster_handle).await;
    println!("[Test] Test components shut down.");

    if !expected_ops.is_empty() {
        warn!("Did not receive all expected operations. Missing: {:?}. Received: {:?}. Note: REMOVE events may not be reliably detected on macOS due to platform limitations.", expected_ops, received_events_ops);
    }

    std::fs::remove_dir_all(&watched_folder).expect("Failed to clean up test dir");
}

/// Test: Broadcaster fails with no folders.
///
/// Ensures that the broadcaster returns an error if no folders are specified.
#[tokio::test]
async fn test_broadcaster_fails_with_no_folders() {
    let config = create_test_config(vec![], false, false, None, None);
    let (_event_tx, event_rx) = tokio::sync::mpsc::channel::<Event>(10);
    let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);

    let result = hirai::network::run_broadcaster(config, event_rx, shutdown_rx).await;
    assert!(
        result.is_err(),
        "Broadcaster should fail when no folders are specified"
    );
}

/// Test: CLI fails with no folders.
///
/// Runs the compiled binary and checks for error output if no folders are specified.
#[test]
fn test_cli_fails_with_no_folders() {
    let hirai_bin = std::env::var("CARGO_BIN_EXE_hirai");
    if hirai_bin.is_err() {
        eprintln!("CARGO_BIN_EXE_hirai not set; skipping CLI integration test");
        return;
    }
    let hirai_bin = hirai_bin.unwrap();

    let output = std::process::Command::new(&hirai_bin)
        .output()
        .expect("Failed to run hirai binary");

    assert!(
        !output.status.success(),
        "Expected non-zero exit code when no folders are specified"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    let stdout = String::from_utf8_lossy(&output.stdout);
    let combined = format!("{}\n{}", stdout, stderr);
    assert!(
        combined.contains("No folders to watch"),
        "Expected error message not found. Output: {}",
        combined
    );
}

/// Test: WebSocket receives events.
///
/// Starts the binary in broadcaster mode with web server, connects via WebSocket, and checks for file events.
#[tokio::test]
async fn test_websocket_receives_events() {
    let hirai_bin = std::env::var("CARGO_BIN_EXE_hirai");
    if hirai_bin.is_err() {
        eprintln!("CARGO_BIN_EXE_hirai not set; skipping WebSocket integration test");
        return;
    }
    let hirai_bin = hirai_bin.unwrap();

    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let watched_folder = temp_dir.path().to_path_buf();

    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("Failed to bind random port");
    let web_port = listener.local_addr().unwrap().port();
    drop(listener);
    let web_addr = format!("127.0.0.1:{}", web_port);

    let mut child = std::process::Command::new(&hirai_bin)
        .arg(watched_folder.to_string_lossy().to_string())
        .arg("--browse")
        .arg("--webaddr")
        .arg(&web_addr)
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .spawn()
        .expect("Failed to start hirai binary");

    let ws_url = format!("ws://{}/ws", web_addr);
    let mut connected = false;
    for _ in 0..20 {
        if let Ok(_) = tokio_tungstenite::connect_async(&ws_url).await {
            connected = true;
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    }
    assert!(connected, "Failed to connect to WebSocket at {}", ws_url);

    let (ws_stream, _) = tokio_tungstenite::connect_async(&ws_url)
        .await
        .expect("WebSocket connect failed");
    let (_ws_write, mut ws_read) = ws_stream.split();

    let test_file = watched_folder.join("websocket_test.txt");
    tokio::fs::write(&test_file, b"hello")
        .await
        .expect("Failed to write test file");

    let mut received = false;
    for _ in 0..10 {
        if let Some(Ok(msg)) = ws_read.next().await {
            if msg.is_text() && msg.to_text().unwrap().contains("websocket_test.txt") {
                received = true;
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    }
    let _ = child.kill();
    assert!(received, "Did not receive expected event over WebSocket");
}

/// Test: High event volume stress test.
///
/// Creates, writes, and removes many files in quick succession and checks that most events are received.
#[tokio::test]
async fn test_high_event_volume_stress() {
    let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
    let watched_folder = temp_dir.path().to_path_buf();

    let multicast_addr = "239.255.255.2:30011".to_string();

    let broadcaster_config = create_test_config(
        vec![watched_folder.to_string_lossy().to_string()],
        false,
        false,
        Some(multicast_addr.clone()),
        None,
    );
    let listener_config =
        create_test_config(vec![], true, true, Some(multicast_addr.clone()), None);

    let (shutdown_tx, shutdown_rx_listener) = watch::channel(false);
    let shutdown_rx_broadcaster = shutdown_tx.subscribe();

    let (watcher_to_broadcaster_tx, watcher_to_broadcaster_rx) = mpsc::channel::<Event>(1000);
    let (listener_received_event_tx, mut listener_received_event_rx) = mpsc::channel::<Event>(1000);

    let listener_handle = setup_listener(
        listener_config,
        listener_received_event_tx,
        shutdown_rx_listener,
    )
    .await;
    tokio::time::sleep(Duration::from_millis(500)).await;

    let broadcaster_handle = setup_broadcaster(
        broadcaster_config.clone(),
        watcher_to_broadcaster_rx,
        shutdown_rx_broadcaster,
    )
    .await;
    tokio::time::sleep(Duration::from_millis(200)).await;

    setup_watcher(broadcaster_config, watcher_to_broadcaster_tx.clone()).await;
    tokio::time::sleep(Duration::from_millis(1000)).await;

    let num_files = 100;
    let mut expected_files = Vec::new();
    for i in 0..num_files {
        let file_path = watched_folder.join(format!("stress_test_{}.txt", i));
        tokio::fs::write(&file_path, b"hello")
            .await
            .expect("Failed to write test file");
        expected_files.push(file_path.clone());
    }
    tokio::time::sleep(Duration::from_millis(1000)).await;
    for file_path in &expected_files {
        tokio::fs::write(file_path, b"world")
            .await
            .expect("Failed to write test file");
    }
    tokio::time::sleep(Duration::from_millis(1000)).await;
    for file_path in &expected_files {
        tokio::fs::remove_file(file_path)
            .await
            .expect("Failed to remove test file");
    }
    tokio::time::sleep(Duration::from_millis(2000)).await;

    let mut received_create = 0;
    let mut received_write = 0;
    let mut received_remove = 0;
    let mut total_received = 0;
    let mut seen_files = std::collections::HashSet::new();
    while let Ok(Some(event)) = timeout(
        Duration::from_millis(100),
        listener_received_event_rx.recv(),
    )
    .await
    {
        if event.path.contains("stress_test_") {
            total_received += 1;
            seen_files.insert(event.path.clone());
            match event.op.as_str() {
                "CREATE" => received_create += 1,
                "WRITE" => received_write += 1,
                "REMOVE" => received_remove += 1,
                _ => {}
            }
        }
        if total_received >= num_files * 3 {
            break;
        }
    }

    shutdown_tx.send(true).unwrap();
    let _ = timeout(DEFAULT_TIMEOUT, listener_handle).await;
    let _ = timeout(DEFAULT_TIMEOUT, broadcaster_handle).await;

    let min_expected = (num_files * 3) as f32 * 0.9;
    assert!(
        total_received as f32 >= min_expected,
        "Expected at least {} events, got {}",
        min_expected,
        total_received
    );
    assert!(
        received_create > 0 && received_write > 0 && received_remove >= 0,
        "Expected to receive all event types"
    );
}