pxh 0.9.22

use std::{
    env,
    io::{Read, Write},
    path::Path,
    process::{Command, Stdio},
    time::{SystemTime, UNIX_EPOCH},
};

use pxh::test_utils::pxh_path;
use tempfile::TempDir;

type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;

// Helper to create a Command with coverage environment variables
fn pxh_command() -> Command {
    let mut cmd = Command::new(pxh_path());

    // Propagate coverage environment variables if they exist
    if let Ok(profile_file) = env::var("LLVM_PROFILE_FILE") {
        cmd.env("LLVM_PROFILE_FILE", profile_file);
    }
    if let Ok(llvm_cov) = env::var("CARGO_LLVM_COV") {
        cmd.env("CARGO_LLVM_COV", llvm_cov);
    }

    cmd
}

// Helper to create test commands using pxh insert
// If days_ago is provided, creates command with that age, otherwise uses current time
fn insert_test_command(db_path: &Path, command: &str, days_ago: Option<u32>) -> Result<()> {
    let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
    let timestamp = match days_ago {
        Some(days) => now - (days as u64 * 86400),
        None => now,
    };

    let output = pxh_command()
        .args([
            "--db",
            db_path.to_str().unwrap(),
            "insert",
            "--shellname",
            "bash",
            "--hostname",
            "test-host",
            "--username",
            "test-user",
            "--session-id",
            "1",
            "--start-unix-timestamp",
            &timestamp.to_string(),
            command,
        ])
        .output()?;

    if !output.status.success() {
        return Err(format!(
            "Failed to insert command: {}",
            String::from_utf8_lossy(&output.stderr)
        )
        .into());
    }

    let seal_output = pxh_command()
        .args([
            "--db",
            db_path.to_str().unwrap(),
            "seal",
            "--session-id",
            "1",
            "--exit-status",
            "0",
            "--end-unix-timestamp",
            &(timestamp + 100).to_string(),
        ])
        .output()?;

    if !seal_output.status.success() {
        return Err(format!(
            "Failed to seal command: {}",
            String::from_utf8_lossy(&seal_output.stderr)
        )
        .into());
    }

    Ok(())
}

// Helper to count commands in a database
fn count_commands(db_path: &Path) -> Result<i64> {
    use rusqlite::Connection;
    let conn = Connection::open(db_path)?;
    let count: i64 =
        conn.prepare("SELECT COUNT(*) FROM command_history")?.query_row([], |row| row.get(0))?;
    Ok(count)
}

// Helper to spawn connected sync processes for testing stdin/stdout mode
// Creates two pxh processes with their stdin/stdout cross-connected:
// - Server process: reads from stdin, writes to stdout
// - Client process: reads from server's stdout, writes to server's stdin
// This allows bidirectional communication between client and server for sync testing
fn spawn_sync_processes(
    client_args: Vec<String>,
    server_args: Vec<String>,
) -> Result<(std::process::Child, std::process::Child)> {
    let mut server = pxh_command()
        .args(server_args)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;

    let server_stdin = server.stdin.take().expect("Failed to get server stdin");
    let server_stdout = server.stdout.take().expect("Failed to get server stdout");

    // Brief delay to ensure server process is initialized before client connects
    std::thread::sleep(std::time::Duration::from_millis(50));

    let client = pxh_command()
        .args(client_args)
        .stdin(server_stdout)
        .stdout(server_stdin)
        .stderr(Stdio::piped())
        .spawn()?;

    Ok((client, server))
}

// Helper to create a database with standard test commands
fn create_test_db_with_commands(db_path: &Path, commands: &[&str]) -> Result<()> {
    for (i, command) in commands.iter().enumerate() {
        insert_test_command(db_path, command, None)?;
        if i < commands.len() - 1 {
            std::thread::sleep(std::time::Duration::from_millis(100));
        }
    }
    Ok(())
}

// Helper to create a pair of databases for sync testing
fn create_test_db_pair(
    temp_dir: &Path,
    client_commands: &[&str],
    server_commands: &[&str],
) -> Result<(std::path::PathBuf, std::path::PathBuf)> {
    let client_db = temp_dir.join("client.db");
    let server_db = temp_dir.join("server.db");

    create_test_db_with_commands(&client_db, client_commands)?;
    create_test_db_with_commands(&server_db, server_commands)?;

    Ok((client_db, server_db))
}

// =============================================================================
// Directory-based sync tests
// =============================================================================

#[test]
fn test_directory_sync() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let sync_dir = temp_dir.path().join("sync_dir");
    std::fs::create_dir(&sync_dir)?;

    let db1 = sync_dir.join("db1.db");
    let db2 = sync_dir.join("db2.db");
    let output_db = temp_dir.path().join("output.db");

    // Create commands in db1
    insert_test_command(&db1, "echo from_db1_1", None)?;
    insert_test_command(&db1, "echo from_db1_2", None)?;

    // Create commands in db2
    insert_test_command(&db2, "echo from_db2_1", None)?;
    insert_test_command(&db2, "echo from_db2_2", None)?;

    // Sync databases from sync_dir
    let output = pxh_command()
        .args(["--db", output_db.to_str().unwrap(), "sync", sync_dir.to_str().unwrap()])
        .output()?;

    assert!(output.status.success());

    // Verify merged database has all commands
    assert_eq!(count_commands(&output_db)?, 4);

    Ok(())
}

#[test]
fn test_directory_sync_rejects_since() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let source_db = temp_dir.path().join("source.db");
    let dest_db = temp_dir.path().join("dest.db");

    insert_test_command(&source_db, "echo test", None)?;

    // --since should be rejected in directory sync mode
    let output = pxh_command()
        .args([
            "--db",
            dest_db.to_str().unwrap(),
            "sync",
            "--since",
            "3",
            temp_dir.path().to_str().unwrap(),
        ])
        .output()?;

    assert!(!output.status.success(), "--since should be rejected in directory sync mode");
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(stderr.contains("--since"), "error should mention --since: {stderr}");

    Ok(())
}

// =============================================================================
// Remote sync tests (stdin/stdout mode)
// =============================================================================

#[test]
fn test_bidirectional_sync_via_stdin_stdout() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let (client_db, server_db) = create_test_db_pair(
        temp_dir.path(),
        &["echo client1", "echo client2", "echo unique_client"],
        &["echo server1", "echo server2", "echo unique_server"],
    )?;

    // Spawn connected processes
    let server_args = vec![
        "--db".to_string(),
        server_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--server".to_string(),
    ];

    let client_args = vec![
        "--db".to_string(),
        client_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--stdin-stdout".to_string(),
    ];

    let (client, server) = spawn_sync_processes(client_args, server_args)?;

    let client_output = client.wait_with_output()?;
    let server_output = server.wait_with_output()?;

    assert!(
        client_output.status.success(),
        "Client failed: {}",
        String::from_utf8_lossy(&client_output.stderr)
    );
    assert!(
        server_output.status.success(),
        "Server failed: {}",
        String::from_utf8_lossy(&server_output.stderr)
    );

    // Check command counts
    let client_count = count_commands(&client_db)?;
    let server_count = count_commands(&server_db)?;

    // Both databases should have the same number of commands after bidirectional sync
    assert_eq!(client_count, server_count);

    // We expect 6 unique commands total:
    // client1, client2, unique_client, server1, server2, unique_server
    assert_eq!(client_count, 6);

    Ok(())
}

#[test]
fn test_send_only_sync() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let (client_db, server_db) =
        create_test_db_pair(temp_dir.path(), &["echo from_client1", "echo from_client2"], &[])?;

    let server_args = vec![
        "--db".to_string(),
        server_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--server".to_string(),
    ];

    let client_args = vec![
        "--db".to_string(),
        client_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--stdin-stdout".to_string(),
        "--send-only".to_string(),
    ];

    let (client, server) = spawn_sync_processes(client_args, server_args)?;

    let client_output = client.wait_with_output()?;
    let server_output = server.wait_with_output()?;

    assert!(
        client_output.status.success(),
        "Client failed: {}",
        String::from_utf8_lossy(&client_output.stderr)
    );
    assert!(
        server_output.status.success(),
        "Server failed: {}",
        String::from_utf8_lossy(&server_output.stderr)
    );

    // Server should have client's commands
    assert_eq!(count_commands(&server_db)?, 2);
    // Client should still have only its original commands
    assert_eq!(count_commands(&client_db)?, 2);

    Ok(())
}

#[test]
fn test_receive_only_sync() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let (client_db, server_db) =
        create_test_db_pair(temp_dir.path(), &[], &["echo from_server1", "echo from_server2"])?;

    let server_args = vec![
        "--db".to_string(),
        server_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--server".to_string(),
    ];

    let client_args = vec![
        "--db".to_string(),
        client_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--stdin-stdout".to_string(),
        "--receive-only".to_string(),
    ];

    let (client, server) = spawn_sync_processes(client_args, server_args)?;

    let client_output = client.wait_with_output()?;
    let server_output = server.wait_with_output()?;

    assert!(
        client_output.status.success(),
        "Client failed: {}",
        String::from_utf8_lossy(&client_output.stderr)
    );
    assert!(
        server_output.status.success(),
        "Server failed: {}",
        String::from_utf8_lossy(&server_output.stderr)
    );

    // Client should have server's commands
    assert_eq!(count_commands(&client_db)?, 2);
    // Server should still have only its original commands
    assert_eq!(count_commands(&server_db)?, 2);

    Ok(())
}

#[test]
fn test_sync_with_since_option() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let client_db = temp_dir.path().join("client.db");
    let server_db = temp_dir.path().join("server.db");

    // Create old and new commands in both databases
    insert_test_command(&client_db, "echo old_client", Some(10))?;
    insert_test_command(&client_db, "echo recent_client", Some(1))?;

    insert_test_command(&server_db, "echo old_server", Some(10))?;
    insert_test_command(&server_db, "echo medium_server", Some(5))?;
    insert_test_command(&server_db, "echo recent_server", Some(1))?;

    let server_args = vec![
        "--db".to_string(),
        server_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--server".to_string(),
        "--since".to_string(),
        "7".to_string(),
    ];

    let client_args = vec![
        "--db".to_string(),
        client_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--stdin-stdout".to_string(),
        "--since".to_string(),
        "7".to_string(),
    ];

    let (client, server) = spawn_sync_processes(client_args, server_args)?;

    let client_output = client.wait_with_output()?;
    let server_output = server.wait_with_output()?;

    assert!(
        client_output.status.success(),
        "Client failed: {}",
        String::from_utf8_lossy(&client_output.stderr)
    );
    assert!(
        server_output.status.success(),
        "Server failed: {}",
        String::from_utf8_lossy(&server_output.stderr)
    );

    // Client should have its old command plus recent commands from both
    assert_eq!(count_commands(&client_db)?, 4); // old_client, recent_client, medium_server, recent_server

    // Server should have all its commands plus recent client command
    assert_eq!(count_commands(&server_db)?, 4); // old_server, medium_server, recent_server, recent_client

    Ok(())
}

#[test]
fn test_sync_error_handling() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let (client_db, server_db) =
        create_test_db_pair(temp_dir.path(), &["echo test"], &["echo test"])?;

    // Test with conflicting options
    let server_args = vec![
        "--db".to_string(),
        server_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--server".to_string(),
    ];

    let client_args = vec![
        "--db".to_string(),
        client_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--stdin-stdout".to_string(),
        "--send-only".to_string(),
        "--receive-only".to_string(),
    ];

    let result = spawn_sync_processes(client_args, server_args);

    match result {
        Ok((client, _server)) => {
            let client_output = client.wait_with_output()?;
            // Client should fail due to conflicting options
            assert!(!client_output.status.success());
        }
        Err(_) => {
            // Expected failure during spawn
        }
    }

    Ok(())
}

// =============================================================================
// SSH sync tests
// =============================================================================

#[test]
fn test_ssh_sync_command_parsing() -> Result<()> {
    // Test that SSH sync commands are properly formed
    let output = pxh_command().args(["sync", "--remote", "user@host", "--help"]).output()?;

    // Should show help without error
    assert!(String::from_utf8(output.stdout)?.contains("Remote host to sync with"));

    Ok(())
}

// =============================================================================
// Scrub directory mode tests
// =============================================================================

#[test]
fn test_scrub_dir_mode_scan() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let db1 = temp_dir.path().join("db1.db");
    let db2 = temp_dir.path().join("db2.db");

    // Insert a command that looks like it contains a secret
    // Using a pattern that matches the AWS API Key pattern: AKIA[0-9A-Z]{16}
    insert_test_command(&db1, "aws configure set aws_access_key_id AKIAIOSFODNN7EXAMPLE", None)?;
    insert_test_command(&db1, "echo normal command", None)?;

    insert_test_command(&db2, "export AWS_SECRET_ACCESS_KEY=AKIAIOSFODNN7EXAMPLE2", None)?;
    insert_test_command(&db2, "ls -la", None)?;

    // Run scrub --scan --dir with dry-run first
    let output = pxh_command()
        .args([
            "--db",
            temp_dir.path().join("unused.db").to_str().unwrap(), // Need a db arg but won't be used
            "scrub",
            "--scan",
            "--dir",
            temp_dir.path().to_str().unwrap(),
            "--dry-run",
        ])
        .output()?;

    assert!(output.status.success(), "scrub --dir failed: {:?}", output);
    let stdout = String::from_utf8_lossy(&output.stdout);
    assert!(stdout.contains("database files"), "Should report found database files");

    // Verify commands still exist (dry-run)
    assert_eq!(count_commands(&db1)?, 2);
    assert_eq!(count_commands(&db2)?, 2);

    Ok(())
}

#[test]
fn test_scrub_dir_mode_with_contraband_pattern() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let db1 = temp_dir.path().join("db1.db");
    let db2 = temp_dir.path().join("db2.db");

    // Insert commands containing "TOPSECRET"
    insert_test_command(&db1, "echo TOPSECRET_VALUE=abc123", None)?;
    insert_test_command(&db1, "echo normal command", None)?;

    insert_test_command(&db2, "export MY_TOPSECRET=password", None)?;
    insert_test_command(&db2, "ls -la", None)?;

    // Verify initial counts
    assert_eq!(count_commands(&db1)?, 2);
    assert_eq!(count_commands(&db2)?, 2);

    // Run scrub with explicit contraband pattern
    let output = pxh_command()
        .args([
            "--db",
            temp_dir.path().join("unused.db").to_str().unwrap(),
            "scrub",
            "--dir",
            temp_dir.path().to_str().unwrap(),
            "-y",        // Skip confirmation
            "TOPSECRET", // The contraband pattern to match
        ])
        .output()?;

    assert!(
        output.status.success(),
        "scrub --dir failed: stdout={} stderr={}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );

    // Verify the TOPSECRET commands were removed
    assert_eq!(count_commands(&db1)?, 1, "db1 should have 1 command after scrub");
    assert_eq!(count_commands(&db2)?, 1, "db2 should have 1 command after scrub");

    // Verify the normal commands remain
    let conn1 = rusqlite::Connection::open(&db1)?;
    let has_normal1: bool = conn1.query_row(
        "SELECT EXISTS(SELECT 1 FROM command_history WHERE full_command LIKE '%normal command%')",
        [],
        |r| r.get(0),
    )?;
    assert!(has_normal1, "Normal command should still exist in db1");

    let conn2 = rusqlite::Connection::open(&db2)?;
    let has_normal2: bool = conn2.query_row(
        "SELECT EXISTS(SELECT 1 FROM command_history WHERE full_command LIKE '%ls -la%')",
        [],
        |r| r.get(0),
    )?;
    assert!(has_normal2, "Normal command should still exist in db2");

    Ok(())
}

// =============================================================================
// Sync secret filtering tests
// =============================================================================

#[test]
fn test_sync_filters_secrets_by_default() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let client_db = temp_dir.path().join("client.db");
    let server_db = temp_dir.path().join("server.db");

    // Client has normal commands
    insert_test_command(&client_db, "echo normal", None)?;

    // Server has a command that looks like it contains a secret
    // (AWS key pattern: AKIA + 16 alphanumeric chars)
    insert_test_command(
        &server_db,
        "aws configure set aws_access_key_id AKIAIOSFODNN7EXAMPLE",
        None,
    )?;
    insert_test_command(&server_db, "echo safe_command", None)?;

    // Run bidirectional sync (client receives from server with secret filtering)
    let server_args = vec![
        "--db".to_string(),
        server_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--server".to_string(),
    ];

    let client_args = vec![
        "--db".to_string(),
        client_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--stdin-stdout".to_string(),
    ];

    let (client, server) = spawn_sync_processes(client_args, server_args)?;

    let client_output = client.wait_with_output()?;
    let server_output = server.wait_with_output()?;

    assert!(
        client_output.status.success(),
        "Client failed: {}",
        String::from_utf8_lossy(&client_output.stderr)
    );
    assert!(
        server_output.status.success(),
        "Server failed: {}",
        String::from_utf8_lossy(&server_output.stderr)
    );

    // Verify the client got the filtered message in stderr
    let client_stderr = String::from_utf8_lossy(&client_output.stderr);

    // Client should have exactly 2 commands (its original + safe_command)
    // if the AWS pattern matched and was filtered
    let client_count = count_commands(&client_db)?;

    // Check if filtering occurred - either by count or by stderr message
    let filtering_occurred = client_count == 2 || client_stderr.contains("filtered");
    assert!(
        filtering_occurred,
        "Secret filtering should have occurred. Client has {} commands, stderr: {}",
        client_count, client_stderr
    );

    // Verify the AWS key command is NOT in the client database
    let conn = rusqlite::Connection::open(&client_db)?;
    let has_secret: bool = conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM command_history WHERE full_command LIKE '%AKIAIOSFODNN7%')",
        [],
        |r| r.get(0),
    )?;
    assert!(!has_secret, "AWS key command should NOT be in client database after sync");

    Ok(())
}

#[test]
fn test_sync_no_secret_filter_flag() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let client_db = temp_dir.path().join("client.db");
    let server_db = temp_dir.path().join("server.db");

    // Client starts empty
    insert_test_command(&client_db, "echo init", None)?;

    // Server has a command with something that might be a secret
    insert_test_command(&server_db, "export TOKEN=ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", None)?;
    insert_test_command(&server_db, "echo normal", None)?;

    // Run sync with --no-secret-filter
    let server_args = vec![
        "--db".to_string(),
        server_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--server".to_string(),
    ];

    let client_args = vec![
        "--db".to_string(),
        client_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--stdin-stdout".to_string(),
        "--no-secret-filter".to_string(),
    ];

    let (client, server) = spawn_sync_processes(client_args, server_args)?;

    let client_output = client.wait_with_output()?;
    let server_output = server.wait_with_output()?;

    assert!(
        client_output.status.success(),
        "Client failed: {}",
        String::from_utf8_lossy(&client_output.stderr)
    );
    assert!(
        server_output.status.success(),
        "Server failed: {}",
        String::from_utf8_lossy(&server_output.stderr)
    );

    // With --no-secret-filter, client should get ALL server commands
    let client_count = count_commands(&client_db)?;
    assert_eq!(client_count, 3, "Client should have all 3 commands (no filtering)");

    Ok(())
}

// =============================================================================
// Remote scrub tests (via v2 protocol)
// =============================================================================

#[test]
fn test_remote_scrub_via_v2_protocol() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let server_db = temp_dir.path().join("server.db");

    // Insert commands on server, including one with a "secret"
    insert_test_command(&server_db, "export AWS_KEY=AKIAIOSFODNN7EXAMPLE", None)?;
    insert_test_command(&server_db, "echo normal", None)?;

    let initial_count = count_commands(&server_db)?;
    assert_eq!(initial_count, 2);

    // Spawn server in server mode
    let mut server = pxh_command()
        .args(["--db", server_db.to_str().unwrap(), "sync", "--server"])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;

    let mut stdin = server.stdin.take().unwrap();
    let mut stdout = server.stdout.take().unwrap();

    // Send scrub-v2 protocol with scan option
    stdin.write_all(b"scrub-v2\n")?;
    stdin.write_all(b"{\"scrub_scan\":true}\n")?;
    stdin.flush()?;
    drop(stdin);

    // Read response
    let mut response = String::new();
    stdout.read_to_string(&mut response)?;

    let status = server.wait()?;
    assert!(status.success(), "Server scrub should succeed");

    // Check if scrub was reported
    // (may or may not have found secrets depending on pattern matching)
    assert!(
        response.contains("entries")
            || response.contains("scrub")
            || response.contains("No entries"),
        "Response should mention scrub result: {}",
        response
    );

    Ok(())
}

#[test]
fn test_scrub_help_shows_new_options() -> Result<()> {
    let output = pxh_command().args(["scrub", "--help"]).output()?;
    let stdout = String::from_utf8_lossy(&output.stdout);

    assert!(stdout.contains("--dir"), "Help should mention --dir option");
    assert!(stdout.contains("--remote"), "Help should mention --remote option");

    Ok(())
}

#[test]
fn test_sync_help_shows_no_secret_filter() -> Result<()> {
    let output = pxh_command().args(["sync", "--help"]).output()?;
    let stdout = String::from_utf8_lossy(&output.stdout);

    assert!(stdout.contains("--no-secret-filter"), "Help should mention --no-secret-filter option");

    Ok(())
}

#[test]
fn test_remote_scrub_with_explicit_pattern() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let server_db = temp_dir.path().join("server.db");

    // Insert commands on server
    insert_test_command(&server_db, "export MYSECRETKEY=value123", None)?;
    insert_test_command(&server_db, "echo normal", None)?;

    let initial_count = count_commands(&server_db)?;
    assert_eq!(initial_count, 2);

    // Spawn server in server mode
    let mut server = pxh_command()
        .args(["--db", server_db.to_str().unwrap(), "sync", "--server"])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;

    let mut stdin = server.stdin.take().unwrap();
    let mut stdout = server.stdout.take().unwrap();

    // Send scrub-v2 protocol with explicit pattern
    stdin.write_all(b"scrub-v2\n")?;
    stdin.write_all(b"{\"scrub\":\"MYSECRETKEY\"}\n")?;
    stdin.flush()?;
    drop(stdin);

    // Read response
    let mut response = String::new();
    stdout.read_to_string(&mut response)?;

    let status = server.wait()?;
    assert!(status.success(), "Server scrub should succeed");

    // Check response
    assert!(
        response.contains("Scrubbed 1 entries") || response.contains("1 entries"),
        "Response should indicate 1 entry scrubbed: {}",
        response
    );

    // Verify the command was actually removed
    let final_count = count_commands(&server_db)?;
    assert_eq!(final_count, 1, "Server should have 1 command after scrub");

    Ok(())
}

#[test]
fn test_scan_dir_mode() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let db1 = temp_dir.path().join("db1.db");
    let db2 = temp_dir.path().join("db2.db");

    // Insert a command with AWS key pattern
    insert_test_command(&db1, "aws configure set aws_access_key_id AKIAIOSFODNN7EXAMPLE", None)?;
    insert_test_command(&db1, "echo normal command", None)?;

    insert_test_command(&db2, "echo safe command", None)?;
    insert_test_command(&db2, "ls -la", None)?;

    // Run scan --dir
    let output = pxh_command()
        .args([
            "--db",
            temp_dir.path().join("unused.db").to_str().unwrap(),
            "scan",
            "--dir",
            temp_dir.path().to_str().unwrap(),
        ])
        .output()?;

    assert!(
        output.status.success(),
        "scan --dir failed: {}",
        String::from_utf8_lossy(&output.stderr)
    );

    let stdout = String::from_utf8_lossy(&output.stdout);

    // Should report scanning database files
    assert!(stdout.contains("database files"), "Should report scanning database files");

    // Should find at least one potential secret (the AWS key)
    assert!(
        stdout.contains("AKIA") || stdout.contains("AWS") || stdout.contains("potential secret"),
        "Should find AWS key pattern in output: {}",
        stdout
    );

    Ok(())
}

#[test]
fn test_scan_help_shows_dir_option() -> Result<()> {
    let output = pxh_command().args(["scan", "--help"]).output()?;
    let stdout = String::from_utf8_lossy(&output.stdout);

    assert!(stdout.contains("--dir"), "Help should mention --dir option");

    Ok(())
}

#[test]
fn test_scrub_dir_requires_scan_or_pattern() -> Result<()> {
    let temp_dir = TempDir::new()?;

    // Create an empty database in the directory
    let db = temp_dir.path().join("test.db");
    insert_test_command(&db, "echo test", None)?;

    // Try to run scrub --dir without --scan or contraband pattern
    let output = pxh_command()
        .args([
            "--db",
            temp_dir.path().join("unused.db").to_str().unwrap(),
            "scrub",
            "--dir",
            temp_dir.path().to_str().unwrap(),
        ])
        .output()?;

    // Should fail with an error about requiring --scan or pattern
    assert!(!output.status.success(), "scrub --dir without --scan or pattern should fail");
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("--scan") || stderr.contains("contraband") || stderr.contains("pattern"),
        "Error should mention requiring --scan or pattern: {}",
        stderr
    );

    Ok(())
}

#[test]
fn test_scrub_dir_rejects_empty_contraband() -> Result<()> {
    let temp_dir = TempDir::new()?;

    let output = pxh_command()
        .args([
            "--db",
            temp_dir.path().join("unused.db").to_str().unwrap(),
            "scrub",
            "--dir",
            temp_dir.path().to_str().unwrap(),
            "",
        ])
        .output()?;

    assert!(!output.status.success(), "scrub --dir with empty contraband should fail");
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(stderr.contains("non-empty"), "Error should mention non-empty: {stderr}");

    Ok(())
}

#[test]
fn test_scrub_remote_requires_scan_or_pattern() -> Result<()> {
    let temp_dir = TempDir::new()?;

    // Try to run scrub --remote without --scan or contraband pattern
    let output = pxh_command()
        .args([
            "--db",
            temp_dir.path().join("unused.db").to_str().unwrap(),
            "scrub",
            "--remote",
            "fake-host",
        ])
        .output()?;

    // Should fail with an error about requiring --scan or pattern
    assert!(!output.status.success(), "scrub --remote without --scan or pattern should fail");
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("--scan") || stderr.contains("contraband") || stderr.contains("pattern"),
        "Error should mention requiring --scan or pattern: {}",
        stderr
    );

    Ok(())
}

#[test]
fn test_scrub_remote_rejects_empty_contraband() -> Result<()> {
    let temp_dir = TempDir::new()?;

    // Try to run scrub --remote with an empty contraband string
    let output = pxh_command()
        .args([
            "--db",
            temp_dir.path().join("unused.db").to_str().unwrap(),
            "scrub",
            "--remote",
            "fake-host",
            "",
        ])
        .output()?;

    assert!(!output.status.success(), "scrub --remote with empty contraband should fail");
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(stderr.contains("non-empty"), "Error should mention non-empty: {stderr}");

    Ok(())
}

#[test]
fn test_v2_protocol_malformed_json_fails() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let db_path = temp_dir.path().join("server.db");

    // Create a database with some data
    insert_test_command(&db_path, "echo hello", None)?;

    // Start server mode
    let mut server = pxh_command()
        .args(["--db", db_path.to_str().unwrap(), "sync", "--server"])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;

    let stdin = server.stdin.as_mut().unwrap();

    // Send v2 mode with malformed JSON
    writeln!(stdin, "receive-v2")?;
    writeln!(stdin, "{{invalid json")?;
    stdin.flush()?;

    // Server should fail with parse error
    let output = server.wait_with_output()?;
    assert!(!output.status.success(), "Server should fail on malformed v2 JSON");
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("Failed to parse v2 protocol options"),
        "Error should mention parsing failure: {}",
        stderr
    );

    Ok(())
}

#[test]
fn test_directory_sync_filters_secrets() -> Result<()> {
    let temp_dir = TempDir::new()?;

    // Create source database with a secret
    let source_db = temp_dir.path().join("source.db");
    insert_test_command(&source_db, "export AWS_KEY=AKIAIOSFODNN7EXAMPLE", None)?;
    insert_test_command(&source_db, "echo normal command", None)?;

    // Create target database
    let target_db = temp_dir.path().join("target.db");
    insert_test_command(&target_db, "echo existing", None)?;

    // Sync from directory (should filter secrets by default)
    let output = pxh_command()
        .args(["--db", target_db.to_str().unwrap(), "sync", temp_dir.path().to_str().unwrap()])
        .output()?;

    assert!(output.status.success(), "Sync should succeed");
    let stdout = String::from_utf8_lossy(&output.stdout);

    // The output should mention filtering
    assert!(
        stdout.contains("filtered") || count_commands(&target_db)? == 2,
        "Secret should be filtered during directory sync. Output: {}",
        stdout
    );

    // Verify the target has the normal command but not the secret
    let target_count = count_commands(&target_db)?;
    assert!(
        target_count == 2, // existing + normal command (not the secret)
        "Expected 2 commands in target (existing + normal), got {}",
        target_count
    );

    Ok(())
}

#[test]
fn test_directory_sync_preserves_machine_id() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let sync_dir = temp_dir.path().join("sync_dir");
    std::fs::create_dir(&sync_dir)?;

    let source_db = sync_dir.join("source.db");
    let output_db = temp_dir.path().join("output.db");

    // Create a command in source_db and set machine_id
    insert_test_command(&source_db, "echo with_machine_id", None)?;
    {
        let conn = rusqlite::Connection::open(&source_db)?;
        conn.execute("UPDATE command_history SET machine_id = 99", [])?;
    }

    // Sync
    let output = pxh_command()
        .args([
            "--db",
            output_db.to_str().unwrap(),
            "sync",
            "--no-secret-filter",
            sync_dir.to_str().unwrap(),
        ])
        .output()?;
    assert!(output.status.success(), "sync failed: {}", String::from_utf8_lossy(&output.stderr));

    // Verify machine_id was preserved
    let conn = rusqlite::Connection::open(&output_db)?;
    let machine_id: Option<i64> = conn.query_row(
        "SELECT machine_id FROM command_history WHERE CAST(full_command AS text) LIKE '%with_machine_id%'",
        [],
        |r| r.get(0),
    )?;
    assert_eq!(machine_id, Some(99), "machine_id should be preserved through sync");

    Ok(())
}

#[test]
fn test_since_filter_removes_null_timestamps() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let source_db = temp_dir.path().join("source.db");
    let dest_db = temp_dir.path().join("dest.db");

    // Create a recent command in the source
    insert_test_command(&source_db, "echo recent", Some(1))?;

    // Insert a command with NULL timestamp directly into the source DB
    // (simulates imported bash history without timestamps)
    {
        let conn = rusqlite::Connection::open(&source_db)?;
        conn.execute(
            "INSERT INTO command_history (session_id, full_command, shellname, hostname, username)
             VALUES (99, CAST('echo null_ts' AS blob), 'bash', CAST('test-host' AS blob), CAST('test-user' AS blob))",
            [],
        )?;
    }

    // Verify source has both commands
    assert_eq!(count_commands(&source_db)?, 2);

    // Use stdin-stdout sync with --since to send only recent data.
    // --since should filter out both old AND null-timestamp commands.
    let server_args = vec![
        "--db".to_string(),
        dest_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--server".to_string(),
        "--no-secret-filter".to_string(),
    ];

    let client_args = vec![
        "--db".to_string(),
        source_db.to_str().unwrap().to_string(),
        "sync".to_string(),
        "--stdin-stdout".to_string(),
        "--send-only".to_string(),
        "--since".to_string(),
        "7".to_string(),
        "--no-secret-filter".to_string(),
    ];

    let (client, server) = spawn_sync_processes(client_args, server_args)?;
    let client_output = client.wait_with_output()?;
    let server_output = server.wait_with_output()?;

    assert!(
        client_output.status.success(),
        "Client failed: {}",
        String::from_utf8_lossy(&client_output.stderr)
    );
    assert!(
        server_output.status.success(),
        "Server failed: {}",
        String::from_utf8_lossy(&server_output.stderr)
    );

    // Dest should have the recent command but NOT the NULL-timestamp one
    let conn = rusqlite::Connection::open(&dest_db)?;
    let has_null_ts: bool = conn
        .query_row(
            "SELECT COUNT(*) FROM command_history WHERE CAST(full_command AS text) = 'echo null_ts'",
            [],
            |r| r.get::<_, i64>(0),
        )
        .map(|n| n > 0)?;
    assert!(!has_null_ts, "NULL-timestamp command should be filtered by --since");

    let has_recent: bool = conn
        .query_row(
            "SELECT COUNT(*) FROM command_history WHERE CAST(full_command AS text) = 'echo recent'",
            [],
            |r| r.get::<_, i64>(0),
        )
        .map(|n| n > 0)?;
    assert!(has_recent, "Recent command should have been synced");

    Ok(())
}

#[test]
fn test_directory_sync_handles_pre_migration_db() -> Result<()> {
    // Bug 5: ATTACHed databases without machine_id column should be migrated
    let temp_dir = TempDir::new()?;
    let sync_dir = temp_dir.path().join("sync_dir");
    std::fs::create_dir(&sync_dir)?;

    // Create a pre-migration database (no machine_id column)
    let old_db = sync_dir.join("old_peer.db");
    {
        let conn = rusqlite::Connection::open(&old_db)?;
        // Create schema WITHOUT machine_id (simulating pre-v1 database)
        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS command_history (
                id INTEGER PRIMARY KEY,
                session_id INTEGER NOT NULL,
                full_command BLOB NOT NULL,
                shellname TEXT NOT NULL,
                hostname BLOB,
                username BLOB,
                working_directory BLOB,
                exit_status INTEGER,
                start_unix_timestamp INTEGER,
                end_unix_timestamp INTEGER
            );
            CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT);",
        )?;
        conn.execute(
            "INSERT INTO command_history (session_id, full_command, shellname, hostname, username, start_unix_timestamp)
             VALUES (1, CAST('echo from_old_peer' AS blob), 'bash', CAST('old-host' AS blob), CAST('user' AS blob), 1000000)",
            [],
        )?;
    }

    let output_db = temp_dir.path().join("output.db");
    insert_test_command(&output_db, "echo existing", None)?;

    // Sync should succeed despite the old DB lacking machine_id
    let output = pxh_command()
        .args([
            "--db",
            output_db.to_str().unwrap(),
            "sync",
            "--no-secret-filter",
            sync_dir.to_str().unwrap(),
        ])
        .output()?;
    assert!(
        output.status.success(),
        "sync should handle pre-migration databases, got: {}",
        String::from_utf8_lossy(&output.stderr)
    );

    // Verify the old command was merged
    let conn = rusqlite::Connection::open(&output_db)?;
    let has_old: bool = conn
        .query_row(
            "SELECT COUNT(*) FROM command_history WHERE CAST(full_command AS text) LIKE '%from_old_peer%'",
            [],
            |r| r.get::<_, i64>(0),
        )
        .map(|n| n > 0)?;
    assert!(has_old, "command from pre-migration DB should be merged");

    Ok(())
}

// =============================================================================
// Incremental directory-sync (machine_id watermark) tests
// =============================================================================

fn set_local_machine_id(db_path: &Path, mid: u64) -> Result<()> {
    use rusqlite::Connection;
    let conn = Connection::open(db_path)?;
    conn.execute("CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value BLOB)", [])?;
    conn.execute(
        "INSERT OR REPLACE INTO settings (key, value) VALUES ('local_machine_id', ?)",
        [mid.to_string().as_bytes()],
    )?;
    Ok(())
}

fn get_watermark(db_path: &Path, mid: u64) -> Result<Option<i64>> {
    use rusqlite::Connection;
    let conn = Connection::open(db_path)?;
    let key = format!("sync_watermark_{mid}");
    let value: Option<Vec<u8>> =
        conn.query_row("SELECT value FROM settings WHERE key = ?", [&key], |r| r.get(0)).ok();
    Ok(value.and_then(|v| String::from_utf8(v).ok()?.parse::<i64>().ok()))
}

fn considered_count(stdout: &str) -> Option<i64> {
    let needle = "considered ";
    let start = stdout.find(needle)? + needle.len();
    let rest = &stdout[start..];
    let end = rest.find(' ')?;
    rest[..end].parse().ok()
}

#[test]
fn test_directory_sync_watermark_skips_seen_rows() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let sync_dir = temp_dir.path().join("sync_dir");
    std::fs::create_dir(&sync_dir)?;
    let src = sync_dir.join("src.db");
    let dst = temp_dir.path().join("dst.db");

    // Source: 3 rows, tagged with a machine_id so the watermark path activates.
    insert_test_command(&src, "echo a", None)?;
    insert_test_command(&src, "echo b", None)?;
    insert_test_command(&src, "echo c", None)?;
    set_local_machine_id(&src, 42)?;

    // First sync: scans all 3 rows, persists watermark.
    let out = pxh_command()
        .args(["--db", dst.to_str().unwrap(), "sync", sync_dir.to_str().unwrap()])
        .output()?;
    assert!(out.status.success(), "first sync failed: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert_eq!(considered_count(&stdout), Some(3), "first sync should consider 3 rows");
    let wm1 = get_watermark(&dst, 42)?.expect("watermark should be set after first sync");
    assert!(wm1 >= 3, "watermark {wm1} should cover all 3 source rows");

    // Add 2 more rows to the source.
    insert_test_command(&src, "echo d", None)?;
    insert_test_command(&src, "echo e", None)?;

    // Second sync: should only consider the 2 new rows above the watermark.
    let out = pxh_command()
        .args(["--db", dst.to_str().unwrap(), "sync", sync_dir.to_str().unwrap()])
        .output()?;
    assert!(out.status.success(), "second sync failed: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert_eq!(considered_count(&stdout), Some(2), "second sync should only consider new rows");
    let wm2 = get_watermark(&dst, 42)?.expect("watermark should advance after second sync");
    assert!(wm2 > wm1, "watermark should advance: {wm1} -> {wm2}");

    Ok(())
}

#[test]
fn test_directory_sync_falls_back_when_no_machine_id() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let sync_dir = temp_dir.path().join("sync_dir");
    std::fs::create_dir(&sync_dir)?;
    let src = sync_dir.join("src.db");
    let dst = temp_dir.path().join("dst.db");

    // Source with no local_machine_id setting -- old DB shape.
    insert_test_command(&src, "echo a", None)?;
    insert_test_command(&src, "echo b", None)?;

    // First sync still merges everything (full-scan fallback).
    let out = pxh_command()
        .args(["--db", dst.to_str().unwrap(), "sync", sync_dir.to_str().unwrap()])
        .output()?;
    assert!(out.status.success());
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert_eq!(considered_count(&stdout), Some(2));

    // No watermark should be stored for an unknown machine_id.
    use rusqlite::Connection;
    let conn = Connection::open(&dst)?;
    let watermarks: i64 = conn
        .query_row("SELECT COUNT(*) FROM settings WHERE key LIKE 'sync_watermark_%'", [], |r| {
            r.get(0)
        })
        .unwrap_or(0);
    assert_eq!(watermarks, 0, "no watermark should be stored without a source machine_id");

    Ok(())
}

#[test]
fn test_directory_sync_warns_on_duplicate_machine_ids() -> Result<()> {
    let temp_dir = TempDir::new()?;
    let sync_dir = temp_dir.path().join("sync_dir");
    std::fs::create_dir(&sync_dir)?;
    let src1 = sync_dir.join("src1.db");
    let src2 = sync_dir.join("src2.db");
    let dst = temp_dir.path().join("dst.db");

    insert_test_command(&src1, "echo from_src1", None)?;
    insert_test_command(&src2, "echo from_src2", None)?;
    set_local_machine_id(&src1, 99)?;
    set_local_machine_id(&src2, 99)?;

    let out = pxh_command()
        .args(["--db", dst.to_str().unwrap(), "sync", sync_dir.to_str().unwrap()])
        .output()?;
    assert!(out.status.success());
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("machine_id 99") && stderr.contains("clone"),
        "expected duplicate-machine_id warning, got stderr: {stderr}"
    );

    Ok(())
}