qrusty 0.22.0

// storage_tests.rs

//! Integration tests for the Storage layer
//!
//! These tests verify the complete storage functionality including
//! message persistence, priority ordering, locking, and cleanup.

use chrono::Utc;
use qrusty::message::{Message, Priority, PriorityOrdering, QueueConfig};
use qrusty::storage::{RenameQueueError, Storage};
use rocksdb::{Options, DB};
use std::sync::Arc;
use tempfile::TempDir;

/// Helper function to create a test storage instance
fn create_test_storage() -> (Storage, TempDir) {
    let temp_dir = TempDir::new().expect("Failed to create temp directory");
    let storage =
        Storage::new(temp_dir.path().to_str().unwrap()).expect("Failed to create test storage");
    (storage, temp_dir)
}

/// Helper function to create a test message
fn create_test_message(queue: &str, priority: u64, payload: &str) -> Message {
    Message {
        id: uuid::Uuid::new_v4().to_string(),
        queue: queue.to_string(),
        priority: Priority::Numeric(priority),
        payload: payload.to_string(),
        created_at: Utc::now(),
        locked_until: None,
        locked_by: None,
        retry_count: 0,
        max_retries: 3,
        payload_ref: None,
        payload_hash: None,
    }
}

#[test]
fn test_storage_new() {
    let temp_dir = TempDir::new().unwrap();
    let storage = Storage::new(temp_dir.path().to_str().unwrap());
    assert!(storage.is_ok());
}

// Verifies: PER-0001, PER-0002
#[test]
fn test_push_message() {
    let (storage, _temp_dir) = create_test_storage();
    let message = create_test_message("test_queue", 100, r#"{"test": "data"}"#);
    let message_id = message.id.clone();

    let result = storage.push(message);
    assert!(result.is_ok());
    assert_eq!(result.unwrap(), message_id);
}

// Verifies: DLV-0001, DLV-0002, API-0003
#[test]
fn test_push_and_pop_message() {
    let (storage, _temp_dir) = create_test_storage();
    let message = create_test_message("test_queue", 100, r#"{"test": "data"}"#);
    let message_id = message.id.clone();
    let payload = message.payload.clone();

    // Push the message
    storage.push(message).unwrap();

    // Pop the message
    let popped = storage.pop("test_queue", "consumer1", 30).unwrap();
    assert!(popped.is_some());

    let popped_msg = popped.unwrap();
    assert_eq!(popped_msg.id, message_id);
    assert_eq!(popped_msg.payload, payload);
    assert_eq!(popped_msg.retry_count, 1); // Should be incremented
    assert!(popped_msg.locked_until.is_some());
    assert_eq!(popped_msg.locked_by, Some("consumer1".to_string()));
}

// Verifies: SCH-0001, SCH-0002
#[test]
fn test_priority_ordering() {
    // Verifies: SCH-0001, SCH-0002
    let (storage, _temp_dir) = create_test_storage();

    // Push messages with different priorities
    let low_priority = create_test_message("test_queue", 10, "low");
    let high_priority = create_test_message("test_queue", 100, "high");
    let medium_priority = create_test_message("test_queue", 50, "medium");

    storage.push(low_priority).unwrap();
    storage.push(high_priority).unwrap();
    storage.push(medium_priority).unwrap();

    // Pop messages and verify order (highest priority first)
    let first = storage.pop("test_queue", "consumer1", 30).unwrap().unwrap();
    assert_eq!(first.payload, "high");
    assert_eq!(first.priority, Priority::Numeric(100));

    let second = storage.pop("test_queue", "consumer2", 30).unwrap().unwrap();
    assert_eq!(second.payload, "medium");
    assert_eq!(second.priority, Priority::Numeric(50));

    let third = storage.pop("test_queue", "consumer3", 30).unwrap().unwrap();
    assert_eq!(third.payload, "low");
    assert_eq!(third.priority, Priority::Numeric(10));
}

#[test]
fn test_fifo_ordering_ignores_priority() {
    // Verifies: SCH-0003
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::Fifo,
        ..Default::default()
    };
    storage.configure_queue("fifo_queue", config);

    let now = Utc::now();

    let mut first = create_test_message("fifo_queue", 999, "first");
    first.created_at = now - chrono::Duration::seconds(2);

    let mut second = create_test_message("fifo_queue", 0, "second");
    second.created_at = now - chrono::Duration::seconds(1);

    // Push in publish order. Priority values should not matter.
    storage.push(first).unwrap();
    storage.push(second).unwrap();

    let msg1 = storage.pop("fifo_queue", "consumer1", 30).unwrap().unwrap();
    assert_eq!(msg1.payload, "first");

    let msg2 = storage.pop("fifo_queue", "consumer2", 30).unwrap().unwrap();
    assert_eq!(msg2.payload, "second");
}

#[test]
fn test_fifo_preserves_position_on_nack_retry() {
    // Verifies: SCH-0004
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::Fifo,
        ..Default::default()
    };
    storage.configure_queue("fifo_retry_queue", config);

    let now = Utc::now();

    let mut first = create_test_message("fifo_retry_queue", 100, "first");
    first.created_at = now - chrono::Duration::seconds(2);
    let first_id = first.id.clone();

    let mut second = create_test_message("fifo_retry_queue", 100, "second");
    second.created_at = now - chrono::Duration::seconds(1);
    let second_id = second.id.clone();

    storage.push(first).unwrap();
    storage.push(second).unwrap();

    // Lock the first message.
    let locked = storage
        .pop("fifo_retry_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(locked.id, first_id);

    // Nack it (retries remain), making it available again.
    let success = storage
        .nack("fifo_retry_queue", &first_id, "consumer1")
        .unwrap();
    assert!(success);

    // Next pop should return the original first message again (not the later one).
    let next = storage
        .pop("fifo_retry_queue", "consumer2", 30)
        .unwrap()
        .unwrap();
    assert_eq!(next.id, first_id);

    // Then the second message should still be present.
    let later = storage
        .pop("fifo_retry_queue", "consumer3", 30)
        .unwrap()
        .unwrap();
    assert_eq!(later.id, second_id);
}

// Verifies: SYS-0001
#[test]
fn test_queue_isolation() {
    let (storage, _temp_dir) = create_test_storage();

    // Push messages to different queues
    let msg1 = create_test_message("queue1", 100, "queue1_data");
    let msg2 = create_test_message("queue2", 100, "queue2_data");

    storage.push(msg1).unwrap();
    storage.push(msg2).unwrap();

    // Pop from queue1 should only get queue1 message
    let popped1 = storage.pop("queue1", "consumer1", 30).unwrap().unwrap();
    assert_eq!(popped1.payload, "queue1_data");

    // Pop from queue2 should only get queue2 message
    let popped2 = storage.pop("queue2", "consumer2", 30).unwrap().unwrap();
    assert_eq!(popped2.payload, "queue2_data");

    // Pop from queue1 again should return None
    let empty = storage.pop("queue1", "consumer3", 30).unwrap();
    assert!(empty.is_none());
}

// Verifies: DLV-0001, DLV-0002
#[test]
fn test_message_locking() {
    let (storage, _temp_dir) = create_test_storage();
    let message = create_test_message("test_queue", 100, "test_data");

    storage.push(message).unwrap();

    // First consumer pops the message (locks it)
    let locked_msg = storage.pop("test_queue", "consumer1", 30).unwrap().unwrap();
    assert!(locked_msg.locked_until.is_some());
    assert_eq!(locked_msg.locked_by, Some("consumer1".to_string()));

    // Second consumer should not get the locked message
    let empty = storage.pop("test_queue", "consumer2", 30).unwrap();
    assert!(empty.is_none());
}

// Verifies: DLV-0003, API-0004
#[test]
fn test_ack_message() {
    let (storage, _temp_dir) = create_test_storage();
    let message = create_test_message("test_queue", 100, "test_data");
    let message_id = message.id.clone();

    storage.push(message).unwrap();

    // Pop and lock the message
    let _locked_msg = storage.pop("test_queue", "consumer1", 30).unwrap().unwrap();

    // Ack the message
    let ack_result = storage.ack("test_queue", &message_id, "consumer1").unwrap();
    assert!(ack_result);

    // Message should be gone
    let empty = storage.pop("test_queue", "consumer2", 30).unwrap();
    assert!(empty.is_none());
}

#[test]
fn test_ack_wrong_consumer() {
    let (storage, _temp_dir) = create_test_storage();
    let message = create_test_message("test_queue", 100, "test_data");
    let message_id = message.id.clone();

    storage.push(message).unwrap();

    // Pop and lock the message with consumer1
    storage.pop("test_queue", "consumer1", 30).unwrap();

    // Try to ack with wrong consumer
    let ack_result = storage.ack("test_queue", &message_id, "consumer2").unwrap();
    assert!(!ack_result);

    // Message should still be there (locked)
    let empty = storage.pop("test_queue", "consumer3", 30).unwrap();
    assert!(empty.is_none());
}

// Verifies: DLV-0004, DLV-0006, DLV-0007, API-0005
#[test]
fn test_nack_message_retry() {
    let (storage, _temp_dir) = create_test_storage();
    let message = create_test_message("test_queue", 100, "test_data");
    let message_id = message.id.clone();

    storage.push(message).unwrap();

    // Pop and lock the message
    storage.pop("test_queue", "consumer1", 30).unwrap();

    // Nack the message (should unlock for retry)
    let nack_result = storage
        .nack("test_queue", &message_id, "consumer1")
        .unwrap();
    assert!(nack_result);

    // Message should be available again
    let retry_msg = storage.pop("test_queue", "consumer2", 30).unwrap().unwrap();
    assert_eq!(retry_msg.id, message_id);
    assert_eq!(retry_msg.retry_count, 2); // Should be incremented again
    assert!(
        retry_msg.locked_until.is_none() || retry_msg.locked_by == Some("consumer2".to_string())
    );
}

// Verifies: DLV-0008, DLV-0009
#[test]
fn test_nack_max_retries_dead_letter() {
    let (storage, _temp_dir) = create_test_storage();
    let mut message = create_test_message("test_queue", 100, "test_data");
    message.max_retries = 2;
    message.retry_count = 2; // At max retries
    let message_id = message.id.clone();

    storage.push(message).unwrap();

    // Pop the message
    storage.pop("test_queue", "consumer1", 30).unwrap();

    // Nack should move to dead letter queue
    let nack_result = storage
        .nack("test_queue", &message_id, "consumer1")
        .unwrap();
    assert!(nack_result);

    // Message should not be available in original queue
    let empty = storage.pop("test_queue", "consumer2", 30).unwrap();
    assert!(empty.is_none());

    // Message should be in dead letter queue (we can't easily test this without
    // exposing DLQ functionality, but the nack succeeded which indicates it was moved)
}

#[test]
fn test_nack_wrong_consumer() {
    let (storage, _temp_dir) = create_test_storage();
    let message = create_test_message("test_queue", 100, "test_data");
    let message_id = message.id.clone();

    storage.push(message).unwrap();

    // Pop and lock with consumer1
    storage.pop("test_queue", "consumer1", 30).unwrap();

    // Try to nack with wrong consumer
    let nack_result = storage
        .nack("test_queue", &message_id, "consumer2")
        .unwrap();
    assert!(!nack_result);
}

#[test]
fn test_empty_queue_pop() {
    let (storage, _temp_dir) = create_test_storage();

    // Try to pop from empty queue
    let result = storage.pop("empty_queue", "consumer1", 30).unwrap();
    assert!(result.is_none());
}

#[test]
fn test_concurrent_operations() {
    let (storage, _temp_dir) = create_test_storage();
    let storage = Arc::new(storage);

    // Push multiple messages
    for i in 0..10 {
        let message = create_test_message("concurrent_queue", i * 10, &format!("data_{}", i));
        storage.push(message).unwrap();
    }

    // Spawn multiple consumers
    let mut handles = vec![];
    for i in 0..5 {
        let storage_clone = storage.clone();
        let consumer_id = format!("consumer_{}", i);

        let handle = std::thread::spawn(move || {
            let mut consumed = vec![];
            while let Some(msg) = storage_clone
                .pop("concurrent_queue", &consumer_id, 30)
                .unwrap()
            {
                consumed.push(msg);
                // Simulate some processing time
                std::thread::sleep(std::time::Duration::from_millis(10));
            }
            consumed
        });
        handles.push(handle);
    }

    // Wait for all consumers to finish
    let mut all_messages = vec![];
    for handle in handles {
        let consumed = handle.join().unwrap();
        all_messages.extend(consumed);
    }

    // Should have consumed all 10 messages
    assert_eq!(all_messages.len(), 10);

    // Verify no duplicates (each message consumed exactly once)
    let mut ids: Vec<String> = all_messages.iter().map(|m| m.id.clone()).collect();
    ids.sort();
    ids.dedup();
    assert_eq!(ids.len(), 10);
}

// Verifies: PER-0001, PER-0002
#[test]
fn test_message_persistence() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();

    let message_id = {
        // Create storage and add a message
        let storage = Storage::new(data_path).unwrap();
        let message = create_test_message("persistent_queue", 100, "persistent_data");
        let id = message.id.clone();
        storage.push(message).unwrap();
        id
    };

    // Recreate storage (simulating restart)
    let storage = Storage::new(data_path).unwrap();

    // Message should still be there
    let recovered_msg = storage
        .pop("persistent_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(recovered_msg.id, message_id);
    assert_eq!(recovered_msg.payload, "persistent_data");
}

#[test]
fn test_queue_configuration_max_first() {
    let (storage, _temp_dir) = create_test_storage();

    // Configure queue for max-first ordering (default behavior)
    let config = QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        ..Default::default()
    };
    storage.configure_queue("max_first_queue", config);

    // Push messages with different priorities
    let low_priority = create_test_message("max_first_queue", 10, "low");
    let high_priority = create_test_message("max_first_queue", 100, "high");
    let medium_priority = create_test_message("max_first_queue", 50, "medium");

    storage.push(low_priority).unwrap();
    storage.push(high_priority).unwrap();
    storage.push(medium_priority).unwrap();

    // Pop messages and verify order (highest priority first)
    let first = storage
        .pop("max_first_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(first.payload, "high");
    assert_eq!(first.priority, Priority::Numeric(100));

    let second = storage
        .pop("max_first_queue", "consumer2", 30)
        .unwrap()
        .unwrap();
    assert_eq!(second.payload, "medium");
    assert_eq!(second.priority, Priority::Numeric(50));

    let third = storage
        .pop("max_first_queue", "consumer3", 30)
        .unwrap()
        .unwrap();
    assert_eq!(third.payload, "low");
    assert_eq!(third.priority, Priority::Numeric(10));
}

// Verifies: SCH-0001, SCH-0002
#[test]
fn test_queue_configuration_min_first() {
    let (storage, _temp_dir) = create_test_storage();

    // Configure queue for min-first ordering (Unix-style)
    let config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        ..Default::default()
    };
    storage.configure_queue("min_first_queue", config);

    // Push messages with different priorities
    let low_priority = create_test_message("min_first_queue", 10, "low");
    let high_priority = create_test_message("min_first_queue", 100, "high");
    let medium_priority = create_test_message("min_first_queue", 50, "medium");

    storage.push(low_priority).unwrap();
    storage.push(high_priority).unwrap();
    storage.push(medium_priority).unwrap();

    // Pop messages and verify order (lowest priority first)
    let first = storage
        .pop("min_first_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(first.payload, "low");
    assert_eq!(first.priority, Priority::Numeric(10));

    let second = storage
        .pop("min_first_queue", "consumer2", 30)
        .unwrap()
        .unwrap();
    assert_eq!(second.payload, "medium");
    assert_eq!(second.priority, Priority::Numeric(50));

    let third = storage
        .pop("min_first_queue", "consumer3", 30)
        .unwrap()
        .unwrap();
    assert_eq!(third.payload, "high");
    assert_eq!(third.priority, Priority::Numeric(100));
}

#[test]
fn test_queue_configuration_default_behavior() {
    let (storage, _temp_dir) = create_test_storage();

    // Don't configure queue - should use default MaxFirst ordering
    let low_priority = create_test_message("default_queue", 10, "low");
    let high_priority = create_test_message("default_queue", 100, "high");

    storage.push(low_priority).unwrap();
    storage.push(high_priority).unwrap();

    // Should default to max-first behavior
    let first = storage
        .pop("default_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(first.payload, "high");
    assert_eq!(first.priority, Priority::Numeric(100));

    let second = storage
        .pop("default_queue", "consumer2", 30)
        .unwrap()
        .unwrap();
    assert_eq!(second.payload, "low");
    assert_eq!(second.priority, Priority::Numeric(10));
}

#[test]
fn test_mixed_queue_configurations() {
    let (storage, _temp_dir) = create_test_storage();

    // Configure different queues with different orderings
    let max_first_config = QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        ..Default::default()
    };
    let min_first_config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        ..Default::default()
    };

    storage.configure_queue("urgent_queue", max_first_config);
    storage.configure_queue("batch_queue", min_first_config);

    // Add messages to both queues
    storage
        .push(create_test_message("urgent_queue", 10, "urgent_low"))
        .unwrap();
    storage
        .push(create_test_message("urgent_queue", 100, "urgent_high"))
        .unwrap();
    storage
        .push(create_test_message("batch_queue", 10, "batch_low"))
        .unwrap();
    storage
        .push(create_test_message("batch_queue", 100, "batch_high"))
        .unwrap();

    // Urgent queue should return high priority first
    let urgent_first = storage
        .pop("urgent_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(urgent_first.payload, "urgent_high");
    assert_eq!(urgent_first.priority, Priority::Numeric(100));

    // Batch queue should return low priority first
    let batch_first = storage
        .pop("batch_queue", "consumer2", 30)
        .unwrap()
        .unwrap();
    assert_eq!(batch_first.payload, "batch_low");
    assert_eq!(batch_first.priority, Priority::Numeric(10));
}

#[test]
fn test_create_queue_method() {
    let (storage, _temp_dir) = create_test_storage();

    // Use create_queue method
    let config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        ..Default::default()
    };
    storage.create_queue("created_queue", config);

    // Verify the configuration was applied
    let retrieved_config = storage.get_queue_config("created_queue");
    assert_eq!(retrieved_config.ordering, PriorityOrdering::MinFirst);

    // Test actual message ordering
    storage
        .push(create_test_message("created_queue", 100, "high"))
        .unwrap();
    storage
        .push(create_test_message("created_queue", 10, "low"))
        .unwrap();

    let first = storage
        .pop("created_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(first.payload, "low");
    assert_eq!(first.priority, Priority::Numeric(10));
}

#[test]
fn test_get_queue_config() {
    let (storage, _temp_dir) = create_test_storage();

    // Test getting config for unconfigured queue (should return default)
    let default_config = storage.get_queue_config("unconfigured_queue");
    assert_eq!(default_config.ordering, PriorityOrdering::MaxFirst);

    // Configure a queue and retrieve its config
    let min_first_config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        ..Default::default()
    };
    storage.configure_queue("min_queue", min_first_config.clone());

    let retrieved_config = storage.get_queue_config("min_queue");
    assert_eq!(retrieved_config.ordering, PriorityOrdering::MinFirst);
}

#[test]
fn test_priority_ordering_edge_cases() {
    let (storage, _temp_dir) = create_test_storage();

    // Test with extreme priority values
    let min_first_config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        ..Default::default()
    };
    storage.configure_queue("edge_queue", min_first_config);

    // Test with 0 and max values
    storage
        .push(create_test_message("edge_queue", 0, "zero"))
        .unwrap();
    storage
        .push(create_test_message("edge_queue", u64::MAX, "max"))
        .unwrap();
    storage
        .push(create_test_message("edge_queue", 1, "one"))
        .unwrap();

    // In min-first, should get 0, then 1, then max
    let first = storage.pop("edge_queue", "consumer1", 30).unwrap().unwrap();
    assert_eq!(first.payload, "zero");
    assert_eq!(first.priority, Priority::Numeric(0));

    let second = storage.pop("edge_queue", "consumer2", 30).unwrap().unwrap();
    assert_eq!(second.payload, "one");
    assert_eq!(second.priority, Priority::Numeric(1));

    let third = storage.pop("edge_queue", "consumer3", 30).unwrap().unwrap();
    assert_eq!(third.payload, "max");
    assert_eq!(third.priority, Priority::Numeric(u64::MAX));
}

#[test]
fn test_concurrent_mixed_priority_queues() {
    let (storage, _temp_dir) = create_test_storage();
    let storage = Arc::new(storage);

    // Configure different queues
    let max_config = QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        ..Default::default()
    };
    let min_config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        ..Default::default()
    };

    storage.configure_queue("max_queue", max_config);
    storage.configure_queue("min_queue", min_config);

    // Push messages to both queues with same priorities but different expected orders
    for i in 0..10 {
        let priority = i * 10;
        storage
            .push(create_test_message(
                "max_queue",
                priority,
                &format!("max_{}", priority),
            ))
            .unwrap();
        storage
            .push(create_test_message(
                "min_queue",
                priority,
                &format!("min_{}", priority),
            ))
            .unwrap();
    }

    // Spawn consumers for both queues
    let max_storage = storage.clone();
    let min_storage = storage.clone();

    let max_handle = std::thread::spawn(move || {
        let mut consumed = vec![];
        while let Some(msg) = max_storage.pop("max_queue", "max_consumer", 30).unwrap() {
            consumed.push(msg.priority);
        }
        consumed
    });

    let min_handle = std::thread::spawn(move || {
        let mut consumed = vec![];
        while let Some(msg) = min_storage.pop("min_queue", "min_consumer", 30).unwrap() {
            consumed.push(msg.priority);
        }
        consumed
    });

    let max_priorities = max_handle.join().unwrap();
    let min_priorities = min_handle.join().unwrap();

    // Max queue should have descending order
    assert_eq!(
        max_priorities,
        vec![90, 80, 70, 60, 50, 40, 30, 20, 10, 0]
            .into_iter()
            .map(Priority::Numeric)
            .collect::<Vec<_>>()
    );

    // Min queue should have ascending order
    assert_eq!(
        min_priorities,
        vec![0, 10, 20, 30, 40, 50, 60, 70, 80, 90]
            .into_iter()
            .map(Priority::Numeric)
            .collect::<Vec<_>>()
    );
}

#[test]
fn test_timeout_monitor_unlock_expired_messages() {
    // Verifies: DLV-0005, PER-0005
    let (storage, _temp_dir) = create_test_storage();
    let storage = Arc::new(storage);

    // Push a message and lock it with a very short timeout
    let message = create_test_message("timeout_queue", 100, "timeout_test");
    let message_id = message.id.clone();

    storage.push(message).unwrap();

    // Pop the message with a 1-second timeout
    let locked_msg = storage
        .pop("timeout_queue", "consumer1", 1)
        .unwrap()
        .unwrap();

    assert!(locked_msg.locked_until.is_some());
    assert_eq!(locked_msg.locked_by, Some("consumer1".to_string()));

    // Wait for the lock to expire
    std::thread::sleep(std::time::Duration::from_secs(2));

    // Run the timeout monitor manually
    let unlocked_count = storage.unlock_expired_messages().unwrap();
    assert_eq!(unlocked_count, 1);

    // The message should now be available again
    let unlocked_msg = storage
        .pop("timeout_queue", "consumer2", 30)
        .unwrap()
        .unwrap();

    assert_eq!(unlocked_msg.id, message_id);
    assert_eq!(unlocked_msg.retry_count, 2); // Should be incremented again
    assert_eq!(unlocked_msg.locked_by, Some("consumer2".to_string()));
}

/// Verifies: DLV-0013 (post-restart stranded lock recovery)
///
/// After a server restart, `queue_counters.locked` is seeded from RocksDB
/// but the in-memory `locked_index` starts empty.  A message whose
/// `locked_until` was in the past at startup (or becomes so shortly after)
/// must still be discoverable by `unlock_expired_messages()`, because the
/// sticky `untracked_locks_possible` flag is initialised to `true`.
#[test]
fn test_unlock_expired_recovers_locks_after_restart() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();

    let message_id = {
        // Session 1: push and lock a message with a very short timeout.
        let storage = Storage::new(data_path).unwrap();
        let msg = create_test_message("restart_locks", 10, "stranded");
        let id = msg.id.clone();
        storage.push(msg).unwrap();
        let locked = storage
            .pop("restart_locks", "consumer1", 1)
            .unwrap()
            .unwrap();
        assert!(locked.locked_until.is_some());
        id
    };

    // Let the lock expire before the restart simulation.
    std::thread::sleep(std::time::Duration::from_secs(2));

    // Session 2: reopen storage.  The locked_index starts empty, so
    // without the sticky full-scan flag the stranded lock would be
    // invisible to unlock_expired_messages.
    let storage = Storage::new(data_path).unwrap();
    let unlocked_count = storage.unlock_expired_messages().unwrap();
    assert_eq!(
        unlocked_count, 1,
        "first sweep after restart must perform a full scan and recover stranded lock"
    );

    // The message should now be poppable again.
    let recovered = storage
        .pop("restart_locks", "consumer2", 30)
        .unwrap()
        .unwrap();
    assert_eq!(recovered.id, message_id);
    assert_eq!(recovered.locked_by, Some("consumer2".to_string()));
}

/// SYS-0018 diagnostic: diagnose_queue reports counter↔scan
/// discrepancies.  Inject drift by doing a re-pop on the OLD code
/// path simulation: artificially increment the locked counter via
/// the public ack/nack/pop sequence, then pre-drift verification
/// should be empty; for negative testing we force drift by directly
/// reseeding (which we'd later ALSO use to recover).
#[test]
fn test_diagnose_queue_reports_clean_state_after_normal_ops() {
    let (storage, _temp_dir) = create_test_storage();

    // Push 3, pop 1, ack it, pop 1 (still locked).
    for i in 0..3 {
        storage
            .push(create_test_message("diag_q", 100, &format!("p-{i}")))
            .unwrap();
    }
    let m1 = storage.pop("diag_q", "c1", 3600).unwrap().unwrap();
    storage.ack("diag_q", &m1.id, "c1").unwrap();
    storage.pop("diag_q", "c1", 3600).unwrap();

    let diag = storage.diagnose_queue("diag_q").unwrap();
    assert!(
        diag.discrepancies.is_empty(),
        "no drift expected: {:?}",
        diag.discrepancies
    );
    assert_eq!(diag.counter_available, 1);
    assert_eq!(diag.counter_locked, 1);
    assert_eq!(diag.scanned_total, 2); // m1 was acked, 2 remain
    assert_eq!(diag.scanned_locked_active, 1);
    assert_eq!(diag.scanned_available, 1);
}

#[test]
fn test_diagnose_queue_detects_injected_drift() {
    let (storage, _temp_dir) = create_test_storage();

    // Real state: 2 available, 0 locked.
    for i in 0..2 {
        storage
            .push(create_test_message("drifty", 100, &format!("p-{i}")))
            .unwrap();
    }

    // Simulate counter drift (e.g. from the pre-fix re-pop bug).
    storage.__test_set_queue_counters("drifty", 5, 3);

    let diag = storage.diagnose_queue("drifty").unwrap();
    assert_eq!(diag.counter_available, 5);
    assert_eq!(diag.counter_locked, 3);
    assert_eq!(diag.scanned_total, 2);
    assert_eq!(diag.scanned_available, 2);
    assert_eq!(diag.scanned_locked_active, 0);
    assert_eq!(diag.discrepancies.len(), 2);
    assert!(diag
        .discrepancies
        .iter()
        .any(|d| d.contains("counter.available")));
    assert!(diag
        .discrepancies
        .iter()
        .any(|d| d.contains("counter.locked")));
}

#[test]
fn test_reseed_queue_counters_repairs_drift() {
    let (storage, _temp_dir) = create_test_storage();

    storage
        .push(create_test_message("repair_q", 100, "p"))
        .unwrap();
    let m = storage.pop("repair_q", "c1", 3600).unwrap().unwrap();
    // Real state: 0 available, 1 locked.

    // Simulate drift.
    storage.__test_set_queue_counters("repair_q", 99, 99);
    let diag_before = storage.diagnose_queue("repair_q").unwrap();
    assert!(
        !diag_before.discrepancies.is_empty(),
        "drift should be detected"
    );

    // Reseed should bring counters back in sync with the scan.
    let (a, l) = storage.reseed_queue_counters("repair_q").unwrap();
    assert_eq!(a, 0);
    assert_eq!(l, 1);
    let diag_after = storage.diagnose_queue("repair_q").unwrap();
    assert!(
        diag_after.discrepancies.is_empty(),
        "drift should be repaired"
    );

    // Verify the queue still works after reseed.
    storage.ack("repair_q", &m.id, "c1").unwrap();
    let diag_final = storage.diagnose_queue("repair_q").unwrap();
    assert_eq!(diag_final.counter_available, 0);
    assert_eq!(diag_final.counter_locked, 0);
}

/// SYS-0018 regression: re-popping a message whose lock expired (before
/// `unlock_expired_messages` has swept it) must not double-count it as
/// locked.  The original pop did `available--, locked++` and the
/// counter still reflects that.  A second pop selecting the same
/// message via the expired-lock path should NOT increment locked
/// again — the message is just being re-locked, not newly locked.
///
/// Before the fix this caused observable counter drift: queues that
/// had any consumer-died-before-acking event would forever report
/// extra "locked" messages that weren't really locked, matching the
/// "queues with a few messages that never get processed" symptom.
#[test]
fn test_repop_of_expired_lock_does_not_double_count() {
    let (storage, _temp_dir) = create_test_storage();

    // Push one message.  After push: avail=1, locked=0.
    storage
        .push(create_test_message("drift_q", 100, "p"))
        .unwrap();
    let s = storage.get_queue_stats("drift_q").unwrap();
    assert_eq!((s.available, s.locked), (1, 0), "after push");

    // Pop with a 1-second timeout.  After pop: avail=0, locked=1.
    let m1 = storage.pop("drift_q", "c1", 1).unwrap().unwrap();
    let s = storage.get_queue_stats("drift_q").unwrap();
    assert_eq!((s.available, s.locked), (0, 1), "after first pop");

    // Wait past the lock timeout but do NOT call unlock_expired_messages.
    std::thread::sleep(std::time::Duration::from_secs(2));

    // Re-pop the same message via the expired-lock path.  The counter
    // already says (avail=0, locked=1) — that's still reality.  The
    // re-pop must not push it to (avail=0/sat, locked=2).
    let m2 = storage.pop("drift_q", "c2", 30).unwrap().unwrap();
    assert_eq!(m1.id, m2.id, "should be the same message");
    let s = storage.get_queue_stats("drift_q").unwrap();
    assert_eq!(
        (s.available, s.locked),
        (0, 1),
        "re-pop of expired lock must not double-count locked"
    );

    // Ack should bring counter back to (avail=0, locked=0).
    assert!(storage.ack("drift_q", &m2.id, "c2").unwrap());
    let s = storage.get_queue_stats("drift_q").unwrap();
    assert_eq!((s.available, s.locked), (0, 0), "after ack");
}

/// Verifies: DLV-0013 (cap-overflow stranded lock recovery)
///
/// When `pop()` skips a `locked_index` insert because the index is at
/// cap, the sticky `untracked_locks_possible` flag is set.  If the index
/// subsequently drops below cap, the next `unlock_expired_messages()`
/// call must still perform a full RocksDB scan to catch the skipped
/// lock — not skip the fallback just because the current index is no
/// longer at cap.
#[test]
fn test_unlock_expired_recovers_cap_overflow_lock() {
    // Force a tiny cap so we can overflow it with a single extra insert.
    // Safe because tests run in parallel but each uses its own Storage
    // constructor, which reads this env var at construction time.
    // We restore the original afterwards.
    let prev = std::env::var("QRUSTY_MAX_LOCKED_INDEX").ok();
    // Use a unique cap value so parallel tests don't race on the env var
    // in a way that matters; the cap is captured into the Storage struct.
    std::env::set_var("QRUSTY_MAX_LOCKED_INDEX", "1");

    let temp_dir = TempDir::new().unwrap();
    let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

    // Restore env immediately after construction so other parallel tests
    // are unaffected; the cap is already baked into `storage`.
    match prev {
        Some(v) => std::env::set_var("QRUSTY_MAX_LOCKED_INDEX", v),
        None => std::env::remove_var("QRUSTY_MAX_LOCKED_INDEX"),
    }

    let storage = Arc::new(storage);

    // Push two messages and lock both.  The first insert into
    // locked_index succeeds; the second hits the cap-1 limit and is
    // skipped — setting the sticky flag.
    storage
        .push(create_test_message("cap_queue", 100, "first"))
        .unwrap();
    storage
        .push(create_test_message("cap_queue", 90, "second"))
        .unwrap();

    let locked1 = storage.pop("cap_queue", "consumer1", 1).unwrap().unwrap();
    let locked2 = storage.pop("cap_queue", "consumer1", 1).unwrap().unwrap();
    assert!(locked1.locked_until.is_some());
    assert!(locked2.locked_until.is_some());

    // Drain the first lock via a successful ack so the index drops
    // below cap (to zero, well below the cap of 1).
    storage.ack("cap_queue", &locked1.id, "consumer1").unwrap();

    // Wait for locked2's lock to expire.  locked2's key was NEVER
    // inserted into the index, so it's only recoverable via the
    // sticky-flag-triggered full scan.
    std::thread::sleep(std::time::Duration::from_secs(2));

    // At this point the index is below cap but the sticky flag is set.
    // A naive sweeper would skip the full scan and leak the lock.
    let unlocked_count = storage.unlock_expired_messages().unwrap();
    assert_eq!(
        unlocked_count, 1,
        "sticky flag must force a full scan even after the index drops below cap"
    );

    // locked2 should now be poppable again.
    let recovered = storage.pop("cap_queue", "consumer2", 30).unwrap().unwrap();
    assert_eq!(recovered.id, locked2.id);
}

#[test]
fn test_allow_duplicates_default_true_allows_duplicate_payloads() {
    // Verifies: PER-0003
    let (storage, _temp_dir) = create_test_storage();

    // Queue is not explicitly configured; defaults should allow duplicates.
    let first = create_test_message("dup_default_queue", 100, "same");
    let second = create_test_message("dup_default_queue", 50, "same");

    storage.push(first).unwrap();
    let result = storage.push(second);
    assert!(result.is_ok());
}

#[test]
fn test_allow_duplicates_false_rejects_unlocked_duplicate_on_publish() {
    // Verifies: PER-0004
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        allow_duplicates: false,
        ..Default::default()
    };
    storage.configure_queue("no_dups_queue", config);

    let first = create_test_message("no_dups_queue", 100, "same");
    let second = create_test_message("no_dups_queue", 50, "same");

    storage.push(first).unwrap();
    let result = storage.push(second);
    assert!(result.is_err());
}

#[test]
fn test_allow_duplicates_false_transition_nack_unlocks_not_drops() {
    // Verifies DLV-0010 transition edge case:
    // When a queue switches from allow_duplicates=true to false while a locked
    // message has an identical unlocked duplicate, the old PER-0005 "drop-on-nack"
    // path is removed (DLV-0010).  Nacking the locked message now UNLOCKS it rather
    // than dropping it.  Both messages briefly coexist; new pushes of the same
    // payload are still blocked by the payload set.
    let (storage, _temp_dir) = create_test_storage();

    // Allow duplicates initially so we can create the required scenario.
    storage.configure_queue(
        "dups_then_disable_queue",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: true,
            ..Default::default()
        },
    );

    let now = Utc::now();
    let mut first = create_test_message("dups_then_disable_queue", 100, "same");
    first.created_at = now - chrono::Duration::seconds(2);
    let first_id = first.id.clone();

    let mut second = create_test_message("dups_then_disable_queue", 50, "same");
    second.created_at = now - chrono::Duration::seconds(1);

    storage.push(first).unwrap();
    storage.push(second).unwrap();

    // Lock the first (higher-priority) message so second remains unlocked.
    let locked = storage
        .pop("dups_then_disable_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(locked.id, first_id);

    // Switch to no-duplicates.  PER-0006 de-dupes unlocked messages only; since
    // there is only one unlocked message ("second"), nothing is removed.
    storage.configure_queue(
        "dups_then_disable_queue",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: false,
            ..Default::default()
        },
    );

    // Under DLV-0010 the drop-on-nack-due-to-duplicate path is removed.
    // Nacking the locked message must unlock it (return true), not drop it.
    let success = storage
        .nack("dups_then_disable_queue", &first_id, "consumer1")
        .unwrap();
    assert!(success);

    // Both messages coexist briefly as a transition artefact — both are poppable.
    let a = storage
        .pop("dups_then_disable_queue", "consumer2", 30)
        .unwrap();
    assert!(a.is_some());
    let b = storage
        .pop("dups_then_disable_queue", "consumer3", 30)
        .unwrap();
    assert!(b.is_some());
}

#[test]
fn test_configure_queue_disabling_duplicates_dedupes_existing_unlocked_messages() {
    // Verifies: PER-0006
    let (storage, _temp_dir) = create_test_storage();

    storage.configure_queue(
        "dedupe_on_disable_queue",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: true,
            ..Default::default()
        },
    );

    // Create duplicates while duplicates are allowed.
    let high = create_test_message("dedupe_on_disable_queue", 100, "same");
    let low = create_test_message("dedupe_on_disable_queue", 50, "same");
    storage.push(high).unwrap();
    storage.push(low).unwrap();

    // Now disable duplicates and expect de-dupe to remove the lower-precedence message.
    storage.configure_queue(
        "dedupe_on_disable_queue",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: false,
            ..Default::default()
        },
    );

    // Only the highest-precedence message should remain.
    let msg = storage
        .pop("dedupe_on_disable_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(msg.payload, "same");
    assert_eq!(msg.priority, Priority::Numeric(100));

    let acked = storage
        .ack("dedupe_on_disable_queue", &msg.id, "consumer1")
        .unwrap();
    assert!(acked);

    let empty = storage
        .pop("dedupe_on_disable_queue", "consumer2", 30)
        .unwrap();
    assert!(empty.is_none());
}

/// SYS-0018 + PER-0006 regression: after dropping duplicates on a
/// configure_queue(allow_duplicates: true → false) transition,
/// `queue_counters.available` shall reflect the post-dedupe count,
/// not the pre-dedupe count.  The configure_queue docstring calls
/// out that the rebuild is a one-shot O(queue_depth) cost; the
/// counters must stay correct through that rebuild.
#[test]
fn test_configure_queue_dedup_rebuild_updates_counters() {
    let (storage, _temp_dir) = create_test_storage();

    storage.configure_queue(
        "dedup_counter_queue",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: true,
            ..Default::default()
        },
    );

    // Push 3 unique + 2 duplicates (5 messages total, 3 distinct payloads).
    for i in 0..3 {
        storage
            .push(create_test_message(
                "dedup_counter_queue",
                100,
                &format!("p-{i}"),
            ))
            .unwrap();
    }
    for _ in 0..2 {
        storage
            .push(create_test_message("dedup_counter_queue", 100, "p-0"))
            .unwrap();
    }

    let pre = storage.get_queue_stats("dedup_counter_queue").unwrap();
    assert_eq!(
        pre.available, 5,
        "precondition: counters reflect all 5 inserted messages"
    );

    // Flip to no-duplicates — triggers the one-shot dedupe rebuild.
    storage.configure_queue(
        "dedup_counter_queue",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: false,
            ..Default::default()
        },
    );

    let post = storage.get_queue_stats("dedup_counter_queue").unwrap();
    assert_eq!(
        post.available, 3,
        "counters must reflect post-dedupe count (2 dupes removed), not pre-dedupe 5"
    );
    assert_eq!(post.locked, 0);

    // Verify via actual pop count that the DB matches the counter.
    let mut popped = 0;
    while storage
        .pop("dedup_counter_queue", "c1", 3600)
        .unwrap()
        .is_some()
    {
        popped += 1;
    }
    assert_eq!(popped, 3, "pop count must match post-dedupe counter");
}

#[test]
fn test_configure_queue_enabling_duplicates_does_not_modify_existing_messages() {
    // Verifies: PER-0007
    let (storage, _temp_dir) = create_test_storage();

    storage.configure_queue(
        "no_change_on_enable_queue",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: false,
            ..Default::default()
        },
    );

    // No duplicates needed here; we just want to ensure existing messages remain.
    let first = create_test_message("no_change_on_enable_queue", 100, "a");
    let second = create_test_message("no_change_on_enable_queue", 50, "b");
    storage.push(first).unwrap();
    storage.push(second).unwrap();

    // Enable duplicates: should not remove or rewrite messages.
    storage.configure_queue(
        "no_change_on_enable_queue",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: true,
            ..Default::default()
        },
    );

    // Both messages should still be consumable.
    let msg1 = storage
        .pop("no_change_on_enable_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    let acked1 = storage
        .ack("no_change_on_enable_queue", &msg1.id, "consumer1")
        .unwrap();
    assert!(acked1);

    let msg2 = storage
        .pop("no_change_on_enable_queue", "consumer2", 30)
        .unwrap()
        .unwrap();
    let acked2 = storage
        .ack("no_change_on_enable_queue", &msg2.id, "consumer2")
        .unwrap();
    assert!(acked2);

    let empty = storage
        .pop("no_change_on_enable_queue", "consumer3", 30)
        .unwrap();
    assert!(empty.is_none());
}

#[test]
fn test_timeout_monitor_no_expired_messages() {
    let (storage, _temp_dir) = create_test_storage();
    let storage = Arc::new(storage);

    // Push and lock a message with a long timeout
    let message = create_test_message("timeout_queue", 100, "long_timeout");
    storage.push(message).unwrap();

    // Pop with 60-second timeout
    storage
        .pop("timeout_queue", "consumer1", 60)
        .unwrap()
        .unwrap();

    // Run timeout monitor - should find no expired messages
    let unlocked_count = storage.unlock_expired_messages().unwrap();
    assert_eq!(unlocked_count, 0);

    // Message should still be locked
    let empty = storage.pop("timeout_queue", "consumer2", 30).unwrap();
    assert!(empty.is_none());
}

#[test]
fn test_timeout_monitor_multiple_expired_messages() {
    let (storage, _temp_dir) = create_test_storage();
    let storage = Arc::new(storage);

    // Push multiple messages and lock them with short timeouts
    for i in 0..5 {
        let message = create_test_message("batch_queue", i * 10, &format!("msg_{}", i));
        storage.push(message).unwrap();

        // Lock each message with 1-second timeout
        storage
            .pop("batch_queue", &format!("consumer_{}", i), 1)
            .unwrap()
            .unwrap();
    }

    // Wait for locks to expire
    std::thread::sleep(std::time::Duration::from_secs(2));

    // Run timeout monitor
    let unlocked_count = storage.unlock_expired_messages().unwrap();
    assert_eq!(unlocked_count, 5);

    // All messages should be available again
    let mut recovered_messages = Vec::new();
    for _ in 0..5 {
        if let Some(msg) = storage.pop("batch_queue", "recovery_consumer", 30).unwrap() {
            recovered_messages.push(msg);
        }
    }

    assert_eq!(recovered_messages.len(), 5);

    // All messages should have incremented retry counts
    for msg in recovered_messages {
        assert_eq!(msg.retry_count, 2);
        assert_eq!(msg.locked_by, Some("recovery_consumer".to_string()));
    }
}

#[test]
fn test_timeout_monitor_mixed_expired_and_valid() {
    let (storage, _temp_dir) = create_test_storage();
    let storage = Arc::new(storage);

    // Push and lock messages with different timeouts
    // Make short_timeout message have higher priority so it's popped first
    let short_msg = create_test_message("mixed_queue", 200, "short_timeout");
    let long_msg = create_test_message("mixed_queue", 100, "long_timeout");

    storage.push(short_msg.clone()).unwrap();
    storage.push(long_msg.clone()).unwrap();

    // Pop in priority order and lock with different timeouts
    // First pop will get the higher priority message (200 - "short_timeout")
    let first_locked = storage
        .pop("mixed_queue", "short_consumer", 1) // 1 second timeout
        .unwrap()
        .unwrap();
    assert_eq!(first_locked.payload, "short_timeout");

    // Second pop will get the lower priority message (100 - "long_timeout")
    let second_locked = storage
        .pop("mixed_queue", "long_consumer", 60) // 60 seconds timeout
        .unwrap()
        .unwrap();
    assert_eq!(second_locked.payload, "long_timeout");

    // Wait for short timeout to expire
    std::thread::sleep(std::time::Duration::from_secs(2));

    // Run timeout monitor - should only unlock the expired one (short_timeout)
    let unlocked_count = storage.unlock_expired_messages().unwrap();
    assert_eq!(unlocked_count, 1);

    // Only the short timeout message should be available
    let available_msg = storage
        .pop("mixed_queue", "recovery_consumer", 30)
        .unwrap()
        .unwrap();

    assert_eq!(available_msg.payload, "short_timeout");
    assert_eq!(available_msg.retry_count, 2);

    // The long timeout message should still be locked
    let empty = storage.pop("mixed_queue", "another_consumer", 30).unwrap();
    assert!(empty.is_none());
}

#[test]
fn test_timeout_monitor_acked_message_cleanup() {
    let (storage, _temp_dir) = create_test_storage();
    let storage = Arc::new(storage);

    // Push and lock a message
    let message = create_test_message("ack_queue", 100, "ack_test");
    let message_id = message.id.clone();

    storage.push(message).unwrap();
    storage.pop("ack_queue", "consumer1", 1).unwrap().unwrap();

    // Ack the message before timeout
    let ack_result = storage.ack("ack_queue", &message_id, "consumer1").unwrap();
    assert!(ack_result);

    // Wait for timeout period
    std::thread::sleep(std::time::Duration::from_secs(2));

    // Timeout monitor should handle gracefully (no panic, cleanup stale index entries)
    let unlocked_count = storage.unlock_expired_messages().unwrap();
    // The count might be 0 since the message was acked and removed
    // The important thing is that it doesn't panic or error
    assert!(unlocked_count == 0);
}

#[test]
fn test_queue_stats_empty_system() {
    let (storage, _temp_dir) = create_test_storage();

    // Test stats on empty system
    let all_stats = storage.get_all_queue_stats().unwrap();
    assert_eq!(all_stats.len(), 0);

    let queues = storage.list_queues().unwrap();
    assert_eq!(queues.len(), 0);
}

#[test]
fn test_queue_stats_single_queue() {
    let (storage, _temp_dir) = create_test_storage();

    // Add some messages to a single queue
    let msg1 = create_test_message("test_queue", 100, "msg1");
    let msg2 = create_test_message("test_queue", 50, "msg2");

    storage.push(msg1).unwrap();
    storage.push(msg2).unwrap();

    // Test queue-specific stats
    let stats = storage.get_queue_stats("test_queue").unwrap();
    assert_eq!(stats.name, "test_queue");
    assert_eq!(stats.available, 2);
    assert_eq!(stats.locked, 0);
    assert_eq!(stats.total, 2);

    // Test all-queue stats
    let all_stats = storage.get_all_queue_stats().unwrap();
    assert_eq!(all_stats.len(), 1);
    assert_eq!(all_stats[0].name, "test_queue");
    assert_eq!(all_stats[0].total, 2);

    // Test queue list
    let queues = storage.list_queues().unwrap();
    assert_eq!(queues, vec!["test_queue"]);
}

#[test]
fn test_queue_stats_multiple_queues() {
    let (storage, _temp_dir) = create_test_storage();

    // Add messages to multiple queues
    storage
        .push(create_test_message("orders", 100, "order1"))
        .unwrap();
    storage
        .push(create_test_message("orders", 50, "order2"))
        .unwrap();
    storage
        .push(create_test_message("notifications", 200, "notif1"))
        .unwrap();
    storage
        .push(create_test_message("analytics", 10, "analytics1"))
        .unwrap();

    // Test all-queue stats
    let all_stats = storage.get_all_queue_stats().unwrap();
    assert_eq!(all_stats.len(), 3);

    // Should be sorted alphabetically
    assert_eq!(all_stats[0].name, "analytics");
    assert_eq!(all_stats[1].name, "notifications");
    assert_eq!(all_stats[2].name, "orders");

    assert_eq!(all_stats[0].total, 1); // analytics
    assert_eq!(all_stats[1].total, 1); // notifications
    assert_eq!(all_stats[2].total, 2); // orders

    // Test individual queue stats
    let orders_stats = storage.get_queue_stats("orders").unwrap();
    assert_eq!(orders_stats.available, 2);
    assert_eq!(orders_stats.locked, 0);

    // Test queue list
    let queues = storage.list_queues().unwrap();
    assert_eq!(queues, vec!["analytics", "notifications", "orders"]);
}

#[test]
fn test_queue_stats_with_locked_messages() {
    let (storage, _temp_dir) = create_test_storage();

    // Add messages
    storage
        .push(create_test_message("work_queue", 100, "task1"))
        .unwrap();
    storage
        .push(create_test_message("work_queue", 50, "task2"))
        .unwrap();
    storage
        .push(create_test_message("work_queue", 25, "task3"))
        .unwrap();

    // Lock one message
    storage.pop("work_queue", "worker1", 30).unwrap().unwrap();

    // Test stats with locked message
    let stats = storage.get_queue_stats("work_queue").unwrap();
    assert_eq!(stats.available, 2);
    assert_eq!(stats.locked, 1);
    assert_eq!(stats.total, 3);

    // Lock another message
    storage.pop("work_queue", "worker2", 30).unwrap().unwrap();

    let stats = storage.get_queue_stats("work_queue").unwrap();
    assert_eq!(stats.available, 1);
    assert_eq!(stats.locked, 2);
    assert_eq!(stats.total, 3);

    // Test all-queue stats
    let all_stats = storage.get_all_queue_stats().unwrap();
    assert_eq!(all_stats.len(), 1);
    assert_eq!(all_stats[0].available, 1);
    assert_eq!(all_stats[0].locked, 2);
}

#[test]
fn test_queue_stats_after_ack_nack() {
    let (storage, _temp_dir) = create_test_storage();

    // Add messages
    let msg1 = create_test_message("process_queue", 100, "task1");
    let msg2 = create_test_message("process_queue", 50, "task2");
    let msg1_id = msg1.id.clone();
    let msg2_id = msg2.id.clone();

    storage.push(msg1).unwrap();
    storage.push(msg2).unwrap();

    // Initial stats
    let stats = storage.get_queue_stats("process_queue").unwrap();
    assert_eq!(stats.total, 2);
    assert_eq!(stats.available, 2);

    // Lock both messages
    storage
        .pop("process_queue", "worker1", 30)
        .unwrap()
        .unwrap();
    storage
        .pop("process_queue", "worker2", 30)
        .unwrap()
        .unwrap();

    let stats = storage.get_queue_stats("process_queue").unwrap();
    assert_eq!(stats.total, 2);
    assert_eq!(stats.locked, 2);
    assert_eq!(stats.available, 0);

    // Ack one message (should remove it)
    storage.ack("process_queue", &msg1_id, "worker1").unwrap();

    let stats = storage.get_queue_stats("process_queue").unwrap();
    assert_eq!(stats.total, 1);
    assert_eq!(stats.locked, 1);
    assert_eq!(stats.available, 0);

    // Nack the other message (should unlock it)
    storage.nack("process_queue", &msg2_id, "worker2").unwrap();

    let stats = storage.get_queue_stats("process_queue").unwrap();
    assert_eq!(stats.total, 1);
    assert_eq!(stats.locked, 0);
    assert_eq!(stats.available, 1);
}

#[test]
fn test_queue_stats_nonexistent_queue() {
    let (storage, _temp_dir) = create_test_storage();

    // Stats for non-existent queue should return zero counts
    let stats = storage.get_queue_stats("nonexistent").unwrap();
    assert_eq!(stats.name, "nonexistent");
    assert_eq!(stats.available, 0);
    assert_eq!(stats.locked, 0);
    assert_eq!(stats.total, 0);
}

#[test]
fn test_delete_queue() {
    let (storage, _temp_dir) = create_test_storage();

    // Create a queue and add some messages
    storage.create_queue("test_queue", QueueConfig::default());

    let msg1 = create_test_message("test_queue", 100, "payload1");
    let msg2 = create_test_message("test_queue", 200, "payload2");
    let msg3 = create_test_message("test_queue", 50, "payload3");

    storage.push(msg1).unwrap();
    storage.push(msg2).unwrap();
    storage.push(msg3).unwrap();

    // Verify messages exist
    let stats = storage.get_queue_stats("test_queue").unwrap();
    assert_eq!(stats.total, 3);

    // Delete the queue
    let deleted_count = storage.delete_queue("test_queue").unwrap();
    assert_eq!(deleted_count, 3);

    // Verify queue is empty and no messages can be consumed
    let stats_after = storage.get_queue_stats("test_queue").unwrap();
    assert_eq!(stats_after.total, 0);

    let result = storage.pop("test_queue", "consumer", 1).unwrap();
    assert!(result.is_none());
}

#[test]
fn test_delete_queue_with_locked_messages() {
    let (storage, _temp_dir) = create_test_storage();

    // Create a queue and add messages
    storage.create_queue("locked_queue", QueueConfig::default());

    let msg1 = create_test_message("locked_queue", 100, "payload1");
    let msg2 = create_test_message("locked_queue", 200, "payload2");

    storage.push(msg1).unwrap();
    storage.push(msg2).unwrap();

    // Lock one message by consuming it
    let _consumed = storage
        .pop("locked_queue", "consumer1", 30)
        .unwrap()
        .unwrap();

    // Verify one message is locked
    let stats = storage.get_queue_stats("locked_queue").unwrap();
    assert_eq!(stats.total, 2);
    assert_eq!(stats.locked, 1);
    assert_eq!(stats.available, 1);

    // Delete the queue (should delete both locked and unlocked messages)
    let deleted_count = storage.delete_queue("locked_queue").unwrap();
    assert_eq!(deleted_count, 2);

    // Verify everything is gone
    let stats_after = storage.get_queue_stats("locked_queue").unwrap();
    assert_eq!(stats_after.total, 0);
}

#[test]
fn test_purge_queue() {
    let (storage, _temp_dir) = create_test_storage();

    // Create a queue and add some messages
    storage.create_queue("purge_test", QueueConfig::default());

    let msg1 = create_test_message("purge_test", 100, "payload1");
    let msg2 = create_test_message("purge_test", 200, "payload2");
    let msg3 = create_test_message("purge_test", 50, "payload3");

    storage.push(msg1).unwrap();
    storage.push(msg2).unwrap();
    storage.push(msg3).unwrap();

    // Verify messages exist
    let stats = storage.get_queue_stats("purge_test").unwrap();
    assert_eq!(stats.total, 3);

    // Purge the queue
    let purged_count = storage.purge_queue("purge_test").unwrap();
    assert_eq!(purged_count, 3);

    // Verify queue is empty but still exists (config should remain)
    let stats_after = storage.get_queue_stats("purge_test").unwrap();
    assert_eq!(stats_after.total, 0);

    // Should be able to add new messages to the purged queue
    let new_msg = create_test_message("purge_test", 300, "new_payload");
    storage.push(new_msg).unwrap();

    let final_stats = storage.get_queue_stats("purge_test").unwrap();
    assert_eq!(final_stats.total, 1);
}

#[test]
fn test_purge_queue_preserves_config() {
    let (storage, _temp_dir) = create_test_storage();

    // Create a queue with specific configuration
    let custom_config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        ..Default::default()
    };
    storage.create_queue("config_test", custom_config.clone());

    let msg1 = create_test_message("config_test", 100, "payload1");
    let msg2 = create_test_message("config_test", 50, "payload2");

    storage.push(msg1).unwrap();
    storage.push(msg2).unwrap();

    // Verify min-first ordering
    let consumed1 = storage.pop("config_test", "consumer", 30).unwrap().unwrap();
    assert_eq!(consumed1.priority, Priority::Numeric(50)); // Lower priority comes first in MinFirst

    // Acknowledge the consumed message
    storage
        .ack("config_test", &consumed1.id, "consumer")
        .unwrap();

    // Purge remaining messages
    let purged_count = storage.purge_queue("config_test").unwrap();
    assert_eq!(purged_count, 1);

    // Add new messages and verify configuration is preserved
    let new_msg1 = create_test_message("config_test", 200, "new1");
    let new_msg2 = create_test_message("config_test", 150, "new2");

    storage.push(new_msg1).unwrap();
    storage.push(new_msg2).unwrap();

    // Should still follow min-first ordering
    let consumed_new = storage
        .pop("config_test", "consumer2", 30)
        .unwrap()
        .unwrap();
    assert_eq!(consumed_new.priority, Priority::Numeric(150)); // Lower priority should come first
}

#[test]
fn test_delete_nonexistent_queue() {
    let (storage, _temp_dir) = create_test_storage();

    // Deleting a non-existent queue should return 0 deleted messages
    let deleted_count = storage.delete_queue("nonexistent").unwrap();
    assert_eq!(deleted_count, 0);
}

#[test]
fn test_purge_nonexistent_queue() {
    let (storage, _temp_dir) = create_test_storage();

    // Purging a non-existent queue should return 0 purged messages
    let purged_count = storage.purge_queue("nonexistent").unwrap();
    assert_eq!(purged_count, 0);
}

#[test]
fn test_queue_isolation_delete_purge() {
    let (storage, _temp_dir) = create_test_storage();

    // Create two queues with messages
    storage.create_queue("queue1", QueueConfig::default());
    storage.create_queue("queue2", QueueConfig::default());

    let msg1_q1 = create_test_message("queue1", 100, "q1_payload1");
    let msg2_q1 = create_test_message("queue1", 200, "q1_payload2");
    let msg1_q2 = create_test_message("queue2", 150, "q2_payload1");
    let msg2_q2 = create_test_message("queue2", 250, "q2_payload2");

    storage.push(msg1_q1).unwrap();
    storage.push(msg2_q1).unwrap();
    storage.push(msg1_q2).unwrap();
    storage.push(msg2_q2).unwrap();

    // Verify both queues have messages
    let stats1 = storage.get_queue_stats("queue1").unwrap();
    let stats2 = storage.get_queue_stats("queue2").unwrap();
    assert_eq!(stats1.total, 2);
    assert_eq!(stats2.total, 2);

    // Delete queue1
    let deleted_count = storage.delete_queue("queue1").unwrap();
    assert_eq!(deleted_count, 2);

    // Verify queue1 is empty but queue2 is unaffected
    let stats1_after = storage.get_queue_stats("queue1").unwrap();
    let stats2_after = storage.get_queue_stats("queue2").unwrap();
    assert_eq!(stats1_after.total, 0);
    assert_eq!(stats2_after.total, 2);

    // Purge queue2
    let purged_count = storage.purge_queue("queue2").unwrap();
    assert_eq!(purged_count, 2);

    // Verify queue2 is now empty
    let stats2_final = storage.get_queue_stats("queue2").unwrap();
    assert_eq!(stats2_final.total, 0);
}

#[test]
fn test_has_duplicate_payload_excludes_message_id() {
    let (storage, _temp_dir) = create_test_storage();

    let message = create_test_message("dup_exclude_queue", 1, "same");
    let message_id = message.id.clone();
    storage.push(message).unwrap();

    // Excluding the only message should report no duplicates.
    let has_dupe = storage
        .has_duplicate_payload("dup_exclude_queue", "same", Some(&message_id))
        .unwrap();
    assert!(!has_dupe);
}

#[test]
fn test_allow_duplicates_false_blocks_push_when_message_is_locked() {
    // Verifies DLV-0010: a locked (in-flight) message's payload blocks a new push,
    // not just unlocked messages.
    let (storage, _temp_dir) = create_test_storage();
    storage.create_queue(
        "locked_dupe_queue",
        QueueConfig {
            allow_duplicates: false,
            ..Default::default()
        },
    );

    // Push and lock a message.
    let message_1 = create_test_message("locked_dupe_queue", 100, "dup_payload");
    storage.push(message_1).unwrap();
    storage
        .pop("locked_dupe_queue", "consumer_lock", 30)
        .unwrap()
        .unwrap();

    // Under DLV-0010, a second push with the same payload must be rejected even
    // though the first message is currently locked (in-flight).
    let message_2 = create_test_message("locked_dupe_queue", 90, "dup_payload");
    let push_res = storage.push(message_2);
    assert!(
        push_res.is_err(),
        "Push must be rejected while payload is locked"
    );
}

#[test]
fn test_stats_and_queue_listing_skip_invalid_and_corrupt_entries() {
    let temp_dir = TempDir::new().expect("Failed to create temp directory");
    let messages_path = format!("{}/messages", temp_dir.path().to_str().unwrap());

    // Seed the underlying RocksDB with some intentionally malformed entries.
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        let db = DB::open(&opts, &messages_path).expect("Failed to open raw test DB");

        // Invalid key format (no slashes).
        db.put(b"invalid_key", b"not_json").unwrap();

        // DLQ and queue_config entries should be ignored by stats.
        db.put(b"_dlq/somequeue/someid", b"not_json").unwrap();
        db.put(b"_queue_config/ignored_queue", b"not_json").unwrap();
        db.put(
            b"_queue_config/ok_queue",
            serde_json::to_vec(&QueueConfig::default()).unwrap(),
        )
        .unwrap();

        // Valid key format but corrupt JSON value (should warn+skip in stats).
        db.put(b"corrupt_queue/0/0/corrupt_id", b"not_json")
            .unwrap();

        // A valid message entry that should be counted.
        let ok_message = create_test_message("ok_queue", 1, "ok");
        db.put(
            b"ok_queue/0/0/ok_id",
            serde_json::to_vec(&ok_message).unwrap(),
        )
        .unwrap();
    }

    let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

    // Stats should succeed and include at least the valid queue.
    let all_stats = storage.get_all_queue_stats().unwrap();
    assert!(all_stats.iter().any(|s| s.name == "ok_queue"));

    // list_queues should include queues based on keys regardless of corrupt values,
    // and should not include DLQ keys.
    let queues = storage.list_queues().unwrap();
    assert!(queues.contains(&"ok_queue".to_string()));
    assert!(queues.contains(&"corrupt_queue".to_string()));
    assert!(!queues.iter().any(|q| q.starts_with("_dlq")));

    // Queue stats for the corrupt queue should not fail.
    let corrupt_stats = storage.get_queue_stats("corrupt_queue").unwrap();
    assert_eq!(corrupt_stats.total, 0);
}

#[test]
fn test_allow_duplicates_false_allows_same_payload_after_ack() {
    // Verifies: PER-0004
    // This test ensures that when a message is ACKed (removed from queue),
    // a new message with the same payload can be enqueued even when
    // allow_duplicates is false. This verifies that there is no persistent
    // index of payloads that prevents re-enqueueing after ACK.
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        allow_duplicates: false,
        ..Default::default()
    };
    storage.configure_queue("no_dups_ack_queue", config);

    // Create and push first message
    let first = create_test_message("no_dups_ack_queue", 100, "same_payload");
    let first_id = first.id.clone();
    storage.push(first).unwrap();

    // Verify duplicate is rejected while message is unlocked and available
    let duplicate_while_available = create_test_message("no_dups_ack_queue", 100, "same_payload");
    let result_while_available = storage.push(duplicate_while_available);
    assert!(
        result_while_available.is_err(),
        "Should reject duplicate while original is available (unlocked)"
    );

    // Pop the message (locks it)
    let locked = storage
        .pop("no_dups_ack_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(locked.id, first_id);

    // ACK the message (removes it from queue)
    let ack_result = storage
        .ack("no_dups_ack_queue", &first_id, "consumer1")
        .unwrap();
    assert!(ack_result, "ACK should succeed");

    // Now try to push a new message with the same payload - should succeed!
    // This is the key test: no persistent index should prevent re-enqueueing
    let second = create_test_message("no_dups_ack_queue", 100, "same_payload");
    let second_id = second.id.clone();
    let result = storage.push(second);
    assert!(
        result.is_ok(),
        "Should allow message with same payload after ACK: {:?}",
        result
    );

    // Verify we can pop the new message
    let popped = storage
        .pop("no_dups_ack_queue", "consumer2", 30)
        .unwrap()
        .unwrap();
    assert_eq!(popped.id, second_id);
    assert_eq!(popped.payload, "same_payload");
}

#[test]
fn test_allow_duplicates_false_allows_same_payload_after_nack_to_dlq() {
    // Verifies: PER-0004, DLV-0004
    // This test ensures that when a message is NACKed and moved to DLQ,
    // a new message with the same payload can be enqueued even when
    // allow_duplicates is false.
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        allow_duplicates: false,
        ..Default::default()
    };
    storage.configure_queue("no_dups_nack_queue", config);

    // Create and push first message with max_retries=0 so it goes to DLQ on first nack
    let mut first = create_test_message("no_dups_nack_queue", 100, "same_payload");
    first.max_retries = 0;
    first.retry_count = 0;
    let first_id = first.id.clone();
    storage.push(first).unwrap();

    // Pop the message (locks it and increments retry_count to 1)
    let locked = storage
        .pop("no_dups_nack_queue", "consumer1", 30)
        .unwrap()
        .unwrap();
    assert_eq!(locked.id, first_id);
    assert_eq!(locked.retry_count, 1); // Should be incremented by pop

    // NACK the message (should move to DLQ since retry_count >= max_retries)
    let nack_result = storage
        .nack("no_dups_nack_queue", &first_id, "consumer1")
        .unwrap();
    assert!(nack_result, "NACK should succeed");

    // Verify queue is empty (message moved to DLQ, not in main queue)
    let empty = storage.pop("no_dups_nack_queue", "consumer2", 30).unwrap();
    assert!(empty.is_none(), "Queue should be empty after DLQ move");

    // Now try to push a new message with the same payload - should succeed!
    let second = create_test_message("no_dups_nack_queue", 100, "same_payload");
    let second_id = second.id.clone();
    let result = storage.push(second);
    assert!(
        result.is_ok(),
        "Should allow message with same payload after NACK to DLQ: {:?}",
        result
    );

    // Verify we can pop the new message
    let popped = storage
        .pop("no_dups_nack_queue", "consumer3", 30)
        .unwrap()
        .unwrap();
    assert_eq!(popped.id, second_id);
    assert_eq!(popped.payload, "same_payload");
}

// ---------------------------------------------------------------------------
// Batch ACK tests
// ---------------------------------------------------------------------------

#[test]
fn test_batch_ack_all_found() {
    let (storage, _temp_dir) = create_test_storage();

    // Push and lock three messages
    let mut ids: Vec<String> = Vec::new();
    for i in 0..3 {
        let m = create_test_message("batch_q", 100, &format!("payload-{}", i));
        let id = m.id.clone();
        storage.push(m).unwrap();
        ids.push(id);
    }

    // Pop all three to lock them
    for _ in 0..3 {
        storage.pop("batch_q", "consumer1", 3600).unwrap();
    }

    let result = storage.batch_ack("batch_q", "consumer1", &ids).unwrap();

    assert_eq!(result.acked.len(), 3);
    assert_eq!(result.not_found.len(), 0);

    // Queue should be empty
    let none = storage.pop("batch_q", "consumer1", 30).unwrap();
    assert!(none.is_none());
}

#[test]
fn test_batch_ack_partial_not_found() {
    let (storage, _temp_dir) = create_test_storage();

    let msg = create_test_message("batch_q", 100, "real");
    let real_id = msg.id.clone();
    storage.push(msg).unwrap();
    storage.pop("batch_q", "consumer1", 3600).unwrap();

    let ids = vec![real_id.clone(), "nonexistent-id".to_string()];
    let result = storage.batch_ack("batch_q", "consumer1", &ids).unwrap();

    assert_eq!(result.acked, vec![real_id]);
    assert_eq!(result.not_found, vec!["nonexistent-id".to_string()]);
}

#[test]
fn test_batch_ack_wrong_consumer_not_acked() {
    let (storage, _temp_dir) = create_test_storage();

    let msg = create_test_message("batch_q", 100, "data");
    let id = msg.id.clone();
    storage.push(msg).unwrap();
    storage.pop("batch_q", "consumer1", 3600).unwrap();

    // Try to ack with a different consumer
    let result = storage
        .batch_ack("batch_q", "wrong-consumer", std::slice::from_ref(&id))
        .unwrap();

    assert_eq!(result.acked.len(), 0);
    assert_eq!(result.not_found, vec![id]);
}

#[test]
fn test_batch_ack_empty_list() {
    let (storage, _temp_dir) = create_test_storage();

    let result = storage.batch_ack("batch_q", "consumer1", &[]).unwrap();

    assert_eq!(result.acked.len(), 0);
    assert_eq!(result.not_found.len(), 0);
}

#[test]
fn test_batch_ack_removes_from_locked_index() {
    let (storage, _temp_dir) = create_test_storage();

    let msg = create_test_message("batch_q", 100, "data");
    let id = msg.id.clone();
    storage.push(msg).unwrap();
    let locked = storage.pop("batch_q", "consumer1", 3600).unwrap().unwrap();
    assert_eq!(locked.id, id);

    storage
        .batch_ack("batch_q", "consumer1", std::slice::from_ref(&id))
        .unwrap();

    // Timeout monitor should find nothing to unlock
    let unlocked = storage.unlock_expired_messages().unwrap();
    assert_eq!(unlocked, 0, "locked_index should be clean after batch_ack");
}

// ---------------------------------------------------------------------------
// Batch NACK tests
// ---------------------------------------------------------------------------

#[test]
fn test_batch_nack_all_unlocked_for_retry() {
    let (storage, _temp_dir) = create_test_storage();

    let mut ids: Vec<String> = Vec::new();
    for i in 0..3 {
        let m = create_test_message("batch_q", 100, &format!("p-{}", i));
        let id = m.id.clone();
        storage.push(m).unwrap();
        ids.push(id);
    }

    for _ in 0..3 {
        storage.pop("batch_q", "consumer1", 3600).unwrap();
    }

    let result = storage.batch_nack("batch_q", "consumer1", &ids).unwrap();

    assert_eq!(result.unlocked.len(), 3);
    assert_eq!(result.dead_lettered.len(), 0);
    assert_eq!(result.dropped.len(), 0);
    assert_eq!(result.not_found.len(), 0);

    // All three should be available again
    for _ in 0..3 {
        let m = storage.pop("batch_q", "consumer2", 30).unwrap();
        assert!(m.is_some());
    }
}

#[test]
fn test_batch_nack_exceeded_retries_moves_to_dlq() {
    let (storage, _temp_dir) = create_test_storage();

    // max_retries = 1, so after one pop (retry_count = 1) it goes to DLQ on nack
    let mut msg = create_test_message("batch_q", 100, "exhausted");
    msg.max_retries = 1;
    let id = msg.id.clone();
    storage.push(msg).unwrap();
    storage.pop("batch_q", "consumer1", 3600).unwrap();

    let result = storage.batch_nack("batch_q", "consumer1", &[id]).unwrap();

    assert_eq!(result.dead_lettered.len(), 1);
    assert_eq!(result.unlocked.len(), 0);

    // Queue should be empty (message in DLQ)
    let none = storage.pop("batch_q", "consumer2", 30).unwrap();
    assert!(none.is_none());
}

#[test]
fn test_batch_nack_no_duplicates_push_rejected_while_locked() {
    // Verifies DLV-0010: the scenario that used to trigger "drop on nack" is now
    // impossible — the second push is rejected because the first is in-flight.
    // Consequently, the `dropped` result field should always be empty.
    let (storage, _temp_dir) = create_test_storage();

    storage.configure_queue(
        "no_dup_q",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: false,
            ..Default::default()
        },
    );

    let mut msg_a = create_test_message("no_dup_q", 100, "same-payload");
    msg_a.max_retries = 10;
    let id_a = msg_a.id.clone();
    storage.push(msg_a).unwrap();
    storage.pop("no_dup_q", "consumer1", 3600).unwrap();

    // Under DLV-0010 the second push is rejected — making the "drop on nack" scenario
    // impossible to construct.
    let mut msg_b = create_test_message("no_dup_q", 50, "same-payload");
    msg_b.max_retries = 10;
    assert!(
        storage.push(msg_b).is_err(),
        "Push of duplicate payload must fail while first message is locked"
    );

    // NACK A: no duplicate exists, so it should be unlocked for retry.
    let result = storage
        .batch_nack("no_dup_q", "consumer1", &[id_a])
        .unwrap();

    assert_eq!(result.unlocked.len(), 1, "should unlock for retry");
    assert_eq!(
        result.dropped.len(),
        0,
        "drop path is unreachable under DLV-0010"
    );

    // A should be available again.
    let a = storage.pop("no_dup_q", "consumer2", 30).unwrap();
    assert!(a.is_some());
}

#[test]
fn test_batch_nack_wrong_consumer_not_nacked() {
    let (storage, _temp_dir) = create_test_storage();

    let msg = create_test_message("batch_q", 100, "data");
    let id = msg.id.clone();
    storage.push(msg).unwrap();
    storage.pop("batch_q", "consumer1", 3600).unwrap();

    let result = storage
        .batch_nack("batch_q", "wrong-consumer", std::slice::from_ref(&id))
        .unwrap();

    assert_eq!(result.not_found, vec![id]);
    assert_eq!(result.unlocked.len(), 0);
}

#[test]
fn test_batch_nack_empty_list() {
    let (storage, _temp_dir) = create_test_storage();

    let result = storage.batch_nack("batch_q", "consumer1", &[]).unwrap();

    assert_eq!(result.unlocked.len(), 0);
    assert_eq!(result.dead_lettered.len(), 0);
    assert_eq!(result.dropped.len(), 0);
    assert_eq!(result.not_found.len(), 0);
}

// ---------------------------------------------------------------------------
// DLV-0014 — batch_ack / batch_nack fast-path regression tests
//
// Before DLV-0014 was extended to cover batch ops, batch_ack/batch_nack did
// a full queue prefix scan even to locate a handful of ids.  On queues with
// millions of messages this dominated latency.  The fast path uses the
// `locked_id_index` to resolve ids in O(batch_size).  These tests verify
// that both the fast and fallback paths deliver identical semantics.
// ---------------------------------------------------------------------------

/// Clear the sticky `untracked_locks_possible` flag by running a full sweep
/// when there are no untracked locks.  Storage is constructed with the flag
/// set to true, so this is how tests exercise the fast path.
fn clear_sticky_flag(storage: &Storage) {
    let _ = storage.unlock_expired_messages().unwrap();
}

#[test]
fn test_batch_ack_fast_path_after_sticky_cleared() {
    let (storage, _temp_dir) = create_test_storage();
    clear_sticky_flag(&storage);

    // Push three, lock all three.
    let mut ids = Vec::new();
    for i in 0..3 {
        let m = create_test_message("fast_q", 100, &format!("p-{i}"));
        ids.push(m.id.clone());
        storage.push(m).unwrap();
    }
    for _ in 0..3 {
        storage.pop("fast_q", "c1", 3600).unwrap().unwrap();
    }

    let result = storage.batch_ack("fast_q", "c1", &ids).unwrap();
    assert_eq!(result.acked.len(), 3);
    assert!(result.not_found.is_empty());
}

#[test]
fn test_batch_ack_fast_path_returns_not_found_without_scan() {
    let (storage, _temp_dir) = create_test_storage();
    clear_sticky_flag(&storage);

    let bogus_ids = vec!["nonexistent-1".to_string(), "nonexistent-2".to_string()];

    let result = storage.batch_ack("any_queue", "c1", &bogus_ids).unwrap();
    // No messages in the queue, no matching index entries — not_found
    // without ever scanning.
    assert!(result.acked.is_empty());
    assert_eq!(result.not_found.len(), 2);
}

#[test]
fn test_batch_ack_fallback_when_sticky_flag_set() {
    // Force a cap of 1 so a second pop() fails to insert into
    // locked_index and sets the sticky flag.
    let prev = std::env::var("QRUSTY_MAX_LOCKED_INDEX").ok();
    std::env::set_var("QRUSTY_MAX_LOCKED_INDEX", "1");

    let temp_dir = TempDir::new().unwrap();
    let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

    match prev {
        Some(v) => std::env::set_var("QRUSTY_MAX_LOCKED_INDEX", v),
        None => std::env::remove_var("QRUSTY_MAX_LOCKED_INDEX"),
    }

    storage
        .push(create_test_message("overflow_q", 100, "first"))
        .unwrap();
    storage
        .push(create_test_message("overflow_q", 90, "second"))
        .unwrap();

    let m1 = storage.pop("overflow_q", "c1", 3600).unwrap().unwrap();
    let m2 = storage.pop("overflow_q", "c1", 3600).unwrap().unwrap();

    // m2 is cap-overflow and is not in the secondary index, so the fast
    // path alone would leave it out.  The fallback prefix scan must find
    // and ack it.
    let ids = vec![m1.id.clone(), m2.id.clone()];
    let result = storage.batch_ack("overflow_q", "c1", &ids).unwrap();
    assert_eq!(
        result.acked.len(),
        2,
        "batch_ack fallback must find cap-overflow locks"
    );
    assert!(result.not_found.is_empty());
}

#[test]
fn test_batch_nack_fast_path_after_sticky_cleared() {
    let (storage, _temp_dir) = create_test_storage();
    clear_sticky_flag(&storage);

    let mut ids = Vec::new();
    for i in 0..3 {
        let m = create_test_message("fast_q", 100, &format!("p-{i}"));
        ids.push(m.id.clone());
        storage.push(m).unwrap();
    }
    for _ in 0..3 {
        storage.pop("fast_q", "c1", 3600).unwrap().unwrap();
    }

    let result = storage.batch_nack("fast_q", "c1", &ids).unwrap();
    assert_eq!(result.unlocked.len(), 3);
    assert!(result.dead_lettered.is_empty());
    assert!(result.not_found.is_empty());
}

#[test]
fn test_batch_nack_fallback_when_sticky_flag_set() {
    let prev = std::env::var("QRUSTY_MAX_LOCKED_INDEX").ok();
    std::env::set_var("QRUSTY_MAX_LOCKED_INDEX", "1");

    let temp_dir = TempDir::new().unwrap();
    let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

    match prev {
        Some(v) => std::env::set_var("QRUSTY_MAX_LOCKED_INDEX", v),
        None => std::env::remove_var("QRUSTY_MAX_LOCKED_INDEX"),
    }

    storage
        .push(create_test_message("overflow_q", 100, "first"))
        .unwrap();
    storage
        .push(create_test_message("overflow_q", 90, "second"))
        .unwrap();

    let m1 = storage.pop("overflow_q", "c1", 3600).unwrap().unwrap();
    let m2 = storage.pop("overflow_q", "c1", 3600).unwrap().unwrap();

    let ids = vec![m1.id.clone(), m2.id.clone()];
    let result = storage.batch_nack("overflow_q", "c1", &ids).unwrap();
    assert_eq!(
        result.unlocked.len(),
        2,
        "batch_nack fallback must find cap-overflow locks"
    );
}

#[test]
fn test_ack_fast_path_cleans_stale_secondary_index_entry() {
    // finalize_ack_by_key cleans up the secondary-index entry when the
    // underlying RocksDB record is missing.  Exercise that branch
    // directly by injecting a stale entry.
    let (storage, _temp_dir) = create_test_storage();
    clear_sticky_flag(&storage);

    let ghost_id = "ghost-id";
    let ghost_key = "some_queue/ghost-key";
    storage.__test_insert_locked_id_index_entry("some_queue", ghost_id, ghost_key);

    assert!(!storage.ack("some_queue", ghost_id, "c1").unwrap());
    assert!(
        !storage.__test_locked_id_index_has("some_queue", ghost_id),
        "ack fast path must clean stale secondary-index entries"
    );
}

#[test]
fn test_nack_fast_path_cleans_stale_secondary_index_entry() {
    let (storage, _temp_dir) = create_test_storage();
    clear_sticky_flag(&storage);

    let ghost_id = "ghost-id";
    let ghost_key = "some_queue/ghost-key";
    storage.__test_insert_locked_id_index_entry("some_queue", ghost_id, ghost_key);

    assert!(!storage.nack("some_queue", ghost_id, "c1").unwrap());
    assert!(
        !storage.__test_locked_id_index_has("some_queue", ghost_id),
        "nack fast path must clean stale secondary-index entries"
    );
}

#[test]
fn test_renew_fast_path_cleans_stale_secondary_index_entry() {
    let (storage, _temp_dir) = create_test_storage();
    clear_sticky_flag(&storage);

    let ghost_id = "ghost-id";
    let ghost_key = "some_queue/ghost-key";
    storage.__test_insert_locked_id_index_entry("some_queue", ghost_id, ghost_key);

    assert!(!storage.renew("some_queue", ghost_id, "c1", 30).unwrap());
    assert!(
        !storage.__test_locked_id_index_has("some_queue", ghost_id),
        "renew fast path must clean stale secondary-index entries"
    );
}

#[test]
fn test_batch_ack_fast_path_cleans_stale_secondary_index_entries() {
    let (storage, _temp_dir) = create_test_storage();
    clear_sticky_flag(&storage);

    // Inject a secondary-index entry that points at a storage key which
    // doesn't exist in RocksDB.  This simulates the only error mode the
    // fast-path stale branch can defend against: an (id, key) pair that
    // survived beyond its underlying message.
    let ghost_id = "ghost-id";
    let ghost_key = "some_queue/ghost-key";
    storage.__test_insert_locked_id_index_entry("some_queue", ghost_id, ghost_key);
    assert!(
        storage.__test_locked_id_index_has("some_queue", ghost_id),
        "precondition: stale entry is present"
    );

    let result = storage
        .batch_ack("some_queue", "c1", &[ghost_id.to_string()])
        .unwrap();
    assert!(result.acked.is_empty());
    assert_eq!(result.not_found, vec![ghost_id.to_string()]);
    assert!(
        !storage.__test_locked_id_index_has("some_queue", ghost_id),
        "batch_ack fast path must clean stale secondary-index entries"
    );
}

#[test]
fn test_batch_nack_fast_path_cleans_stale_secondary_index_entries() {
    let (storage, _temp_dir) = create_test_storage();
    clear_sticky_flag(&storage);

    let ghost_id = "ghost-id";
    let ghost_key = "some_queue/ghost-key";
    storage.__test_insert_locked_id_index_entry("some_queue", ghost_id, ghost_key);

    let result = storage
        .batch_nack("some_queue", "c1", &[ghost_id.to_string()])
        .unwrap();
    assert!(result.unlocked.is_empty());
    assert!(result.dead_lettered.is_empty());
    assert_eq!(result.not_found, vec![ghost_id.to_string()]);
    assert!(
        !storage.__test_locked_id_index_has("some_queue", ghost_id),
        "batch_nack fast path must clean stale secondary-index entries"
    );
}

// ---------------------------------------------------------------------------
// DLV-0014 — rename_queue locked_index cleanup regression tests
//
// `rename_queue` previously did an O(total_locked_across_all_queues)
// retain pass on `locked_index`.  The fix collects the source queue's
// old keys during its single prefix scan and removes them by exact key,
// leaving unrelated queues' entries untouched.
// ---------------------------------------------------------------------------

/// DLV-0014: `rename_queue` used to call `locked_index.retain()`, which
/// scans every entry across every queue.  The fix collects the source
/// queue's old keys during the single prefix scan and removes them by
/// exact-key deletion.  Before the fix, injecting an unrelated
/// locked-index entry and renaming another queue would have *also*
/// walked that unrelated entry.  This test verifies it isn't.
#[test]
fn test_rename_queue_does_not_touch_unrelated_locked_id_index_entries() {
    let (storage, _temp_dir) = create_test_storage();

    // Lock a real message in queue `other` so it has a secondary-index entry.
    storage
        .push(create_test_message("other", 100, "keep"))
        .unwrap();
    let other = storage.pop("other", "c1", 3600).unwrap().unwrap();
    assert!(
        storage.__test_locked_id_index_has("other", &other.id),
        "precondition: unrelated queue's lock is tracked"
    );

    // Lock a real message in queue `from` so there is something to move.
    storage
        .push(create_test_message("from_q", 100, "move"))
        .unwrap();
    let moved = storage.pop("from_q", "c1", 3600).unwrap().unwrap();

    storage.rename_queue("from_q", "to_q").unwrap();

    // Unrelated queue's secondary-index entry is untouched — same
    // (queue, id) still resolves.
    assert!(
        storage.__test_locked_id_index_has("other", &other.id),
        "rename must not touch unrelated queues' secondary-index entries"
    );

    // Source queue's entries were moved under the new queue name.
    assert!(
        !storage.__test_locked_id_index_has("from_q", &moved.id),
        "old queue name must be fully cleared from the secondary index"
    );
    assert!(
        storage.__test_locked_id_index_has("to_q", &moved.id),
        "new queue name must contain the rekeyed entry"
    );
}

#[test]
fn test_rename_queue_leaves_unrelated_queue_locks_alone() {
    let (storage, _temp_dir) = create_test_storage();

    // Unrelated queue with an active lock.
    storage
        .push(create_test_message("unrelated", 100, "keep"))
        .unwrap();
    let unrelated = storage.pop("unrelated", "c1", 3600).unwrap().unwrap();

    // Source queue with its own active lock.
    storage
        .push(create_test_message("from_q", 100, "move"))
        .unwrap();
    let moved = storage.pop("from_q", "c1", 3600).unwrap().unwrap();

    storage.rename_queue("from_q", "to_q").unwrap();

    // The unrelated queue's lock must still be acknowledgeable by its
    // original consumer — which only works if its `locked_index` entry
    // and the secondary index were untouched by the rename.
    assert!(storage.ack("unrelated", &unrelated.id, "c1").unwrap());

    // And the moved lock must be acknowledgeable under the new queue
    // name (the secondary index was rebuilt and the primary index got
    // the new key).
    assert!(storage.ack("to_q", &moved.id, "c1").unwrap());
}

// ---------------------------------------------------------------------------
// In-memory payload set tests (PER-0008, PER-0009, PER-0010, DLV-0010)
//
// Helpers shared by this section
// ---------------------------------------------------------------------------

fn no_dup_config() -> QueueConfig {
    QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        allow_duplicates: false,
        ..Default::default()
    }
}

fn allow_dup_config() -> QueueConfig {
    QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        allow_duplicates: true,
        ..Default::default()
    }
}

// ---------------------------------------------------------------------------
// DLV-0010: A locked message's payload must block a new push of the same
// payload.  This is the primary behavioral change introduced by the in-memory
// set — previously the duplicate check skipped locked messages.
// ---------------------------------------------------------------------------

// Verifies: DLV-0010, PER-0008
#[test]
fn test_pset_blocks_push_while_locked() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", no_dup_config());

    let m1 = create_test_message("nd", 100, "payload-x");
    storage.push(m1).unwrap();
    let _locked = storage.pop("nd", "c1", 3600).unwrap().unwrap();

    // Same payload — must be rejected even though original is locked (in-flight).
    let m2 = create_test_message("nd", 100, "payload-x");
    let result = storage.push(m2);
    assert!(
        result.is_err(),
        "Push of duplicate payload must be rejected while original is locked"
    );
}

// ---------------------------------------------------------------------------
// PER-0010 — push / ack cycle: ack removes payload so re-push is allowed.
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_allows_push_after_ack() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", no_dup_config());

    let m1 = create_test_message("nd", 100, "payload-reuse");
    storage.push(m1).unwrap();
    let locked = storage.pop("nd", "c1", 3600).unwrap().unwrap();
    storage.ack("nd", &locked.id, "c1").unwrap();

    let m2 = create_test_message("nd", 100, "payload-reuse");
    assert!(
        storage.push(m2).is_ok(),
        "Push of same payload must succeed after ack"
    );
}

// ---------------------------------------------------------------------------
// PER-0010 — nack-retry: payload stays in set (message returns to queue).
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_nack_retry_keeps_payload_blocked() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", no_dup_config());

    let m1 = create_test_message("nd", 100, "payload-y");
    storage.push(m1).unwrap();
    let locked = storage.pop("nd", "c1", 3600).unwrap().unwrap();
    // retry_count=0 < max_retries=3 → unlocked for retry
    storage.nack("nd", &locked.id, "c1").unwrap();

    // Payload must still be blocked (message is back in queue).
    let m2 = create_test_message("nd", 100, "payload-y");
    assert!(
        storage.push(m2).is_err(),
        "Push of duplicate payload must be rejected after nack-retry"
    );
}

// ---------------------------------------------------------------------------
// PER-0010 — nack-DLQ: payload removed from set, re-push allowed.
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_nack_dlq_allows_repush() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", no_dup_config());

    let mut m1 = create_test_message("nd", 100, "payload-z");
    m1.max_retries = 0; // immediate DLQ on first nack
    storage.push(m1).unwrap();
    let locked = storage.pop("nd", "c1", 3600).unwrap().unwrap();
    storage.nack("nd", &locked.id, "c1").unwrap(); // → DLQ

    let m2 = create_test_message("nd", 100, "payload-z");
    assert!(
        storage.push(m2).is_ok(),
        "Push of same payload must succeed after message is dead-lettered"
    );
}

// ---------------------------------------------------------------------------
// PER-0009 — startup reconstruction: unlocked messages are included.
// ---------------------------------------------------------------------------

// Verifies: PER-0009
#[test]
fn test_pset_startup_reconstruction_blocks_push() {
    let temp_dir = TempDir::new().unwrap();
    let path = temp_dir.path().to_str().unwrap();

    // Session 1: push a message and leave it unlocked.
    {
        let storage = Storage::new(path).unwrap();
        storage.configure_queue("nd", no_dup_config());
        let m = create_test_message("nd", 100, "persisted-payload");
        storage.push(m).unwrap();
    }

    // Session 2: reopen — set must be reconstructed from DB.
    {
        let storage = Storage::new(path).unwrap();
        let m2 = create_test_message("nd", 100, "persisted-payload");
        assert!(
            storage.push(m2).is_err(),
            "Startup reconstruction must block duplicate for unlocked message"
        );
    }
}

// ---------------------------------------------------------------------------
// PER-0009 — startup reconstruction includes locked messages (DLV-0010
// extends the invariant to locked messages too).
// ---------------------------------------------------------------------------

// Verifies: PER-0009, DLV-0010
#[test]
fn test_pset_startup_reconstruction_includes_locked() {
    let temp_dir = TempDir::new().unwrap();
    let path = temp_dir.path().to_str().unwrap();

    // Session 1: push and lock a message (simulate consumer crashed).
    {
        let storage = Storage::new(path).unwrap();
        storage.configure_queue("nd", no_dup_config());
        let m = create_test_message("nd", 100, "locked-payload");
        storage.push(m).unwrap();
        let _locked = storage.pop("nd", "c1", 3600).unwrap().unwrap();
        // Storage goes out of scope without ack/nack — lock is abandoned in DB.
    }

    // Session 2: locked message's payload must still block a new push.
    {
        let storage = Storage::new(path).unwrap();
        let m2 = create_test_message("nd", 100, "locked-payload");
        assert!(
            storage.push(m2).is_err(),
            "Startup reconstruction must block duplicate for locked message"
        );
    }
}

// ---------------------------------------------------------------------------
// PER-0010 — purge clears the set; same payload can be re-pushed.
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_purge_clears_set() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", no_dup_config());

    let m1 = create_test_message("nd", 100, "purge-payload");
    storage.push(m1).unwrap();

    // Baseline: duplicate is blocked.
    let m_check = create_test_message("nd", 100, "purge-payload");
    assert!(storage.push(m_check).is_err());

    storage.purge_queue("nd").unwrap();

    let m2 = create_test_message("nd", 100, "purge-payload");
    assert!(
        storage.push(m2).is_ok(),
        "Same payload must be allowed after purge"
    );
}

// ---------------------------------------------------------------------------
// PER-0010 — delete_queue clears the set; re-creating queue and re-pushing
// the same payload must succeed.
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_delete_clears_set() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", no_dup_config());

    let m1 = create_test_message("nd", 100, "delete-payload");
    storage.push(m1).unwrap();

    storage.delete_queue("nd").unwrap();
    storage.configure_queue("nd", no_dup_config());

    let m2 = create_test_message("nd", 100, "delete-payload");
    assert!(
        storage.push(m2).is_ok(),
        "Same payload must be allowed after queue delete + recreate"
    );
}

// ---------------------------------------------------------------------------
// PER-0010 — configure false→true: removing the no-dup constraint allows
// payloads that were previously blocked.
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_disable_constraint_drops_set() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", no_dup_config());

    let m1 = create_test_message("nd", 100, "shared-payload");
    storage.push(m1).unwrap();

    // Switch to allow duplicates.
    storage.configure_queue("nd", allow_dup_config());

    let m2 = create_test_message("nd", 100, "shared-payload");
    assert!(
        storage.push(m2).is_ok(),
        "Same payload must be allowed after switching to allow_duplicates=true"
    );
}

// ---------------------------------------------------------------------------
// PER-0010 — configure true→false: the set is built from existing messages,
// blocking future pushes of the same payloads.
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_enable_constraint_builds_set() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", allow_dup_config());

    // Push while allow_duplicates=true — this is fine.
    let m1 = create_test_message("nd", 100, "existing-payload");
    storage.push(m1).unwrap();

    // Switch to disallow duplicates.  Existing messages are now in the set.
    storage.configure_queue("nd", no_dup_config());

    let m2 = create_test_message("nd", 100, "existing-payload");
    assert!(
        storage.push(m2).is_err(),
        "Existing payload must be in set after enabling allow_duplicates=false"
    );
}

// ---------------------------------------------------------------------------
// PER-0010 — batch_ack removes payloads from set; re-push of each is allowed.
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_batch_ack_removes_payloads() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", no_dup_config());

    let payloads = ["ba1", "ba2", "ba3"];
    let mut ids = Vec::new();
    for p in &payloads {
        let m = create_test_message("nd", 100, p);
        storage.push(m).unwrap();
        let locked = storage.pop("nd", "c1", 3600).unwrap().unwrap();
        ids.push(locked.id);
    }

    storage.batch_ack("nd", "c1", &ids).unwrap();

    for p in &payloads {
        let m = create_test_message("nd", 100, p);
        assert!(
            storage.push(m).is_ok(),
            "Payload {} must be re-pushable after batch_ack",
            p
        );
    }
}

// ---------------------------------------------------------------------------
// PER-0010 — batch_nack with DLQ removes payloads; re-push allowed.
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_batch_nack_dlq_removes_payloads() {
    let (storage, _dir) = create_test_storage();
    storage.configure_queue("nd", no_dup_config());

    let payloads = ["bn1", "bn2"];
    let mut ids = Vec::new();
    for p in &payloads {
        let mut m = create_test_message("nd", 100, p);
        m.max_retries = 0;
        storage.push(m).unwrap();
        let locked = storage.pop("nd", "c1", 3600).unwrap().unwrap();
        ids.push(locked.id);
    }

    let result = storage.batch_nack("nd", "c1", &ids).unwrap();
    assert_eq!(result.dead_lettered.len(), 2);

    for p in &payloads {
        let m = create_test_message("nd", 100, p);
        assert!(
            storage.push(m).is_ok(),
            "Payload {} must be re-pushable after batch_nack DLQ",
            p
        );
    }
}

// ---------------------------------------------------------------------------
// PER-0010 — unlock_expired with retry keeps payload in set.
// ---------------------------------------------------------------------------

// Verifies: PER-0010
#[test]
fn test_pset_unlock_expired_retry_keeps_payload_blocked() {
    let temp_dir = TempDir::new().unwrap();
    let path = temp_dir.path().to_str().unwrap();
    let past = Utc::now() - chrono::Duration::seconds(60);

    // Session 1: push and lock a message, then close storage so the file lock
    // is released before we open the raw DB.
    let locked_id = {
        let storage = Storage::new(path).unwrap();
        storage.configure_queue("nd", no_dup_config());
        let m = create_test_message("nd", 100, "exp-payload");
        storage.push(m).unwrap();
        let locked = storage.pop("nd", "c1", 3600).unwrap().unwrap();
        locked.id
        // storage is dropped here — file lock released
    };

    // Patch the stored message's locked_until to the past so it looks expired.
    {
        let mut opts = Options::default();
        opts.create_if_missing(false);
        let db = DB::open(&opts, format!("{}/messages", path)).unwrap();
        let prefix = "nd/";
        let iter = db.prefix_iterator(prefix.as_bytes());
        for item in iter {
            let (key, value) = item.unwrap();
            if !String::from_utf8_lossy(&key).starts_with(prefix) {
                break;
            }
            let mut m: Message = serde_json::from_slice(&value).unwrap();
            if m.id == locked_id {
                m.locked_until = Some(past);
                db.put(&key, serde_json::to_vec(&m).unwrap()).unwrap();
                break;
            }
        }
        // db dropped here — file lock released
    }

    // Session 2: open fresh storage.  Startup reconstruction includes the
    // expired-lock message (PER-0009 covers locked messages).  Pop treats
    // expired locks as available, so we can consume it.  The payload stays
    // in the set because the message is still in the queue (retry, not DLQ).
    {
        let storage = Storage::new(path).unwrap();

        // pop() returns expired-lock messages as available.
        let popped = storage.pop("nd", "c2", 30).unwrap();
        assert!(
            popped.is_some(),
            "Expired-lock message must be available via pop"
        );

        // The message was re-locked to c2 — payload is still in the queue.
        // A new push with the same payload must be rejected.
        let m2 = create_test_message("nd", 100, "exp-payload");
        assert!(
            storage.push(m2).is_err(),
            "Payload must stay blocked while the re-locked message is in the queue"
        );
    }
}

// ---------------------------------------------------------------------------
// rename_queue edge cases — covers storage.rs L33-53, L213-227
// ---------------------------------------------------------------------------

#[test]
fn test_rename_queue_same_name_returns_ok() {
    let (storage, _tmp) = create_test_storage();
    let msg = create_test_message("rq_same", 1, r#"{"x":1}"#);
    storage.push(msg).unwrap();

    // from == to → early return Ok(())
    let result = storage.rename_queue("rq_same", "rq_same");
    assert!(
        result.is_ok(),
        "rename to same name must succeed: {result:?}"
    );
}

#[test]
fn test_rename_queue_not_found_returns_error() {
    let (storage, _tmp) = create_test_storage();

    let result = storage.rename_queue("does_not_exist", "also_new");
    assert!(
        matches!(result, Err(RenameQueueError::NotFound)),
        "expected NotFound, got {result:?}"
    );
}

#[test]
fn test_rename_queue_already_exists_returns_error() {
    let (storage, _tmp) = create_test_storage();
    let msg1 = create_test_message("rq_from", 1, r#"{"a":1}"#);
    let msg2 = create_test_message("rq_to", 1, r#"{"b":2}"#);
    storage.push(msg1).unwrap();
    storage.push(msg2).unwrap();

    let result = storage.rename_queue("rq_from", "rq_to");
    assert!(
        matches!(result, Err(RenameQueueError::AlreadyExists)),
        "expected AlreadyExists, got {result:?}"
    );
}

#[test]
fn test_rename_queue_empty_name_returns_storage_error() {
    let (storage, _tmp) = create_test_storage();

    // empty `from`
    let result = storage.rename_queue("", "some_name");
    assert!(
        matches!(result, Err(RenameQueueError::Storage(_))),
        "expected Storage error for empty from, got {result:?}"
    );

    // empty `to`
    let result2 = storage.rename_queue("some_name", "");
    assert!(
        matches!(result2, Err(RenameQueueError::Storage(_))),
        "expected Storage error for empty to, got {result2:?}"
    );

    // whitespace-only names
    let result3 = storage.rename_queue("  ", "name");
    assert!(
        matches!(result3, Err(RenameQueueError::Storage(_))),
        "expected Storage error for whitespace name, got {result3:?}"
    );
}

#[test]
fn test_rename_queue_error_display() {
    // RenameQueueError::NotFound
    let e = RenameQueueError::NotFound;
    assert_eq!(e.to_string(), "queue not found");
    assert!(std::error::Error::source(&e).is_none());

    // RenameQueueError::AlreadyExists
    let e = RenameQueueError::AlreadyExists;
    assert_eq!(e.to_string(), "target queue already exists");
    assert!(std::error::Error::source(&e).is_none());

    // RenameQueueError::Storage — Display includes inner message
    let inner = anyhow::anyhow!("disk full");
    let e = RenameQueueError::Storage(inner);
    assert!(e.to_string().contains("storage error"));
    assert!(std::error::Error::source(&e).is_some());

    // From<anyhow::Error> conversion
    let anyhow_err = anyhow::anyhow!("oops");
    let converted: RenameQueueError = anyhow_err.into();
    assert!(matches!(converted, RenameQueueError::Storage(_)));
}

// ---------------------------------------------------------------------------
// Text priority (lexicographic ordering) tests
// ---------------------------------------------------------------------------

use qrusty::message::PriorityKind;

fn create_text_priority_message(queue: &str, priority: &str, payload: &str) -> Message {
    Message {
        id: uuid::Uuid::new_v4().to_string(),
        queue: queue.to_string(),
        priority: Priority::Text(priority.to_string()),
        payload: payload.to_string(),
        created_at: Utc::now(),
        locked_until: None,
        locked_by: None,
        retry_count: 0,
        max_retries: 3,
        payload_ref: None,
        payload_hash: None,
    }
}

#[test]
fn test_text_priority_min_first() {
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("text_min", config);

    // Push in non-sorted order
    storage
        .push(create_text_priority_message("text_min", "charlie", "c"))
        .unwrap();
    storage
        .push(create_text_priority_message("text_min", "alpha", "a"))
        .unwrap();
    storage
        .push(create_text_priority_message("text_min", "bravo", "b"))
        .unwrap();

    // MinFirst: should pop in ascending lexicographic order
    let first = storage.pop("text_min", "c1", 30).unwrap().unwrap();
    assert_eq!(first.payload, "a");
    assert_eq!(first.priority, Priority::Text("alpha".to_string()));

    let second = storage.pop("text_min", "c2", 30).unwrap().unwrap();
    assert_eq!(second.payload, "b");
    assert_eq!(second.priority, Priority::Text("bravo".to_string()));

    let third = storage.pop("text_min", "c3", 30).unwrap().unwrap();
    assert_eq!(third.payload, "c");
    assert_eq!(third.priority, Priority::Text("charlie".to_string()));
}

#[test]
fn test_text_priority_max_first() {
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("text_max", config);

    storage
        .push(create_text_priority_message("text_max", "alpha", "a"))
        .unwrap();
    storage
        .push(create_text_priority_message("text_max", "charlie", "c"))
        .unwrap();
    storage
        .push(create_text_priority_message("text_max", "bravo", "b"))
        .unwrap();

    // MaxFirst: should pop in descending lexicographic order
    let first = storage.pop("text_max", "c1", 30).unwrap().unwrap();
    assert_eq!(first.payload, "c");
    assert_eq!(first.priority, Priority::Text("charlie".to_string()));

    let second = storage.pop("text_max", "c2", 30).unwrap().unwrap();
    assert_eq!(second.payload, "b");
    assert_eq!(second.priority, Priority::Text("bravo".to_string()));

    let third = storage.pop("text_max", "c3", 30).unwrap().unwrap();
    assert_eq!(third.payload, "a");
    assert_eq!(third.priority, Priority::Text("alpha".to_string()));
}

#[test]
fn test_text_priority_fifo() {
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::Fifo,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("text_fifo", config);

    let now = Utc::now();

    let mut m1 = create_text_priority_message("text_fifo", "zzz", "first");
    m1.created_at = now - chrono::Duration::seconds(2);
    let mut m2 = create_text_priority_message("text_fifo", "aaa", "second");
    m2.created_at = now - chrono::Duration::seconds(1);

    storage.push(m1).unwrap();
    storage.push(m2).unwrap();

    // Fifo ignores priority text — arrival order
    let first = storage.pop("text_fifo", "c1", 30).unwrap().unwrap();
    assert_eq!(first.payload, "first");

    let second = storage.pop("text_fifo", "c2", 30).unwrap().unwrap();
    assert_eq!(second.payload, "second");
}

#[test]
fn test_text_priority_kind_mismatch_rejected() {
    let (storage, _temp_dir) = create_test_storage();

    // Configure as text priority queue
    let config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("text_only", config);

    // Pushing a numeric priority should fail
    let msg = create_test_message("text_only", 42, "nope");
    let err = storage.push(msg).unwrap_err();
    assert!(err.to_string().contains("Priority kind mismatch"));

    // And vice versa: numeric queue rejects text priority
    storage.configure_queue("num_only", QueueConfig::default());
    let msg = create_text_priority_message("num_only", "hello", "nope");
    let err = storage.push(msg).unwrap_err();
    assert!(err.to_string().contains("Priority kind mismatch"));
}

#[test]
fn test_text_priority_validation() {
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("val_q", config);

    // Empty string rejected
    let msg = create_text_priority_message("val_q", "", "x");
    let err = storage.push(msg).unwrap_err();
    assert!(err.to_string().contains("empty"));

    // String with '/' accepted (encoded transparently in key)
    let msg = create_text_priority_message("val_q", "a/b", "slash_ok");
    storage.push(msg).unwrap();

    // Long strings accepted (no length limit)
    let long = "x".repeat(200);
    let msg = create_text_priority_message("val_q", &long, "ok");
    storage.push(msg).unwrap();
}

#[test]
fn test_text_priority_max_first_prefix_ordering() {
    // Verify strings with shared prefixes sort correctly under MaxFirst
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("pfx", config);

    for key in &["a", "ab", "abc", "b", "ba"] {
        storage
            .push(create_text_priority_message("pfx", key, key))
            .unwrap();
    }

    // MaxFirst descending: ba, b, abc, ab, a
    let order: Vec<String> = {
        let mut v = Vec::new();
        for i in 0..5 {
            let m = storage.pop("pfx", &format!("c{}", i), 30).unwrap().unwrap();
            v.push(m.payload);
        }
        v
    };
    assert_eq!(order, vec!["ba", "b", "abc", "ab", "a"]);
}

// Verifies: PER-0013
#[test]
fn test_delete_queue_removes_dlq_entries() {
    let (storage, _temp_dir) = create_test_storage();

    storage.create_queue("dlq_test", QueueConfig::default());

    // Push a message with max_retries=0 so nack sends it straight to DLQ
    let mut msg = create_test_message("dlq_test", 100, "will_dlq");
    msg.max_retries = 0;
    let msg_id = msg.id.clone();
    storage.push(msg).unwrap();

    // Pop then nack → message goes to DLQ
    let consumed = storage.pop("dlq_test", "c1", 30).unwrap().unwrap();
    assert_eq!(consumed.id, msg_id);
    storage.nack("dlq_test", &msg_id, "c1").unwrap();

    // Queue has no live messages but DLQ entry exists
    let stats = storage.get_queue_stats("dlq_test").unwrap();
    assert_eq!(stats.total, 0);

    // Delete the queue
    storage.delete_queue("dlq_test").unwrap();

    // queue_exists must return false (DLQ entries cleaned up)
    assert!(!storage.queue_exists("dlq_test").unwrap());

    // Re-creating the queue must succeed
    storage.create_queue("dlq_test", QueueConfig::default());
    assert!(storage.queue_exists("dlq_test").unwrap());
}

// Verifies: PER-0013
#[test]
fn test_delete_queue_with_dlq_allows_recreate_via_api() {
    let (storage, _temp_dir) = create_test_storage();

    storage.create_queue("recreate_test", QueueConfig::default());

    // Push message with max_retries=0 → nack → DLQ
    let mut msg = create_test_message("recreate_test", 50, "dlq_payload");
    msg.max_retries = 0;
    let msg_id = msg.id.clone();
    storage.push(msg).unwrap();
    storage.pop("recreate_test", "c1", 30).unwrap().unwrap();
    storage.nack("recreate_test", &msg_id, "c1").unwrap();

    // Delete and verify clean slate
    storage.delete_queue("recreate_test").unwrap();
    assert!(!storage.queue_exists("recreate_test").unwrap());

    // Push to the re-created queue should work
    storage.create_queue("recreate_test", QueueConfig::default());
    let msg2 = create_test_message("recreate_test", 100, "fresh_msg");
    storage.push(msg2).unwrap();
    let stats = storage.get_queue_stats("recreate_test").unwrap();
    assert_eq!(stats.total, 1);
}

/// Verifies: SCH-0002
#[test]
fn test_text_priority_with_slash_min_first() {
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("slash_q", config);

    // Publish messages with path-like priorities containing /
    let msg1 = create_text_priority_message("slash_q", "/usr/local/bin", "bin_payload");
    storage.push(msg1).unwrap();

    let msg2 = create_text_priority_message("slash_q", "/usr/local/share", "share_payload");
    storage.push(msg2).unwrap();

    let msg3 = create_text_priority_message("slash_q", "/etc/config", "etc_payload");
    storage.push(msg3).unwrap();

    // MinFirst: lexicographic ascending, so /etc < /usr/local/bin < /usr/local/share
    let m = storage.pop("slash_q", "c1", 30).unwrap().unwrap();
    assert_eq!(m.payload, "etc_payload");
    assert_eq!(m.priority, Priority::Text("/etc/config".to_string()));
    storage.ack("slash_q", &m.id, "c1").unwrap();

    let m = storage.pop("slash_q", "c1", 30).unwrap().unwrap();
    assert_eq!(m.payload, "bin_payload");
    assert_eq!(m.priority, Priority::Text("/usr/local/bin".to_string()));
    storage.ack("slash_q", &m.id, "c1").unwrap();

    let m = storage.pop("slash_q", "c1", 30).unwrap().unwrap();
    assert_eq!(m.payload, "share_payload");
    assert_eq!(m.priority, Priority::Text("/usr/local/share".to_string()));
    storage.ack("slash_q", &m.id, "c1").unwrap();
}

/// Verifies: SCH-0002
#[test]
fn test_text_priority_with_slash_max_first() {
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MaxFirst,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("slash_max_q", config);

    let msg1 = create_text_priority_message("slash_max_q", "/a/first", "first_payload");
    storage.push(msg1).unwrap();

    let msg2 = create_text_priority_message("slash_max_q", "/z/last", "last_payload");
    storage.push(msg2).unwrap();

    // MaxFirst: lexicographic descending, so /z > /a
    let m = storage.pop("slash_max_q", "c1", 30).unwrap().unwrap();
    assert_eq!(m.payload, "last_payload");
    assert_eq!(m.priority, Priority::Text("/z/last".to_string()));
    storage.ack("slash_max_q", &m.id, "c1").unwrap();

    let m = storage.pop("slash_max_q", "c1", 30).unwrap().unwrap();
    assert_eq!(m.payload, "first_payload");
    assert_eq!(m.priority, Priority::Text("/a/first".to_string()));
}

/// Verifies: SCH-0002
#[test]
fn test_text_priority_slash_does_not_corrupt_queue_stats() {
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("stats_slash_q", config);

    let msg1 = create_text_priority_message("stats_slash_q", "/a/b/c/d", "payload1");
    storage.push(msg1).unwrap();

    let msg2 = create_text_priority_message("stats_slash_q", "simple", "payload2");
    storage.push(msg2).unwrap();

    // Stats should show 2 messages in stats_slash_q, not be confused by / in keys
    let stats = storage.get_queue_stats("stats_slash_q").unwrap();
    assert_eq!(stats.total, 2);
    assert_eq!(stats.available, 2);

    // list_queues should show exactly one queue
    let queues = storage.list_queues().unwrap();
    assert!(queues.contains(&"stats_slash_q".to_string()));

    // get_all_queue_stats should also work correctly
    let all_stats = storage.get_all_queue_stats().unwrap();
    let qs = all_stats
        .iter()
        .find(|s| s.name == "stats_slash_q")
        .unwrap();
    assert_eq!(qs.total, 2);
}

/// Verifies: SCH-0002
#[test]
fn test_text_priority_with_null_byte() {
    let (storage, _temp_dir) = create_test_storage();

    let config = QueueConfig {
        ordering: PriorityOrdering::MinFirst,
        priority_kind: PriorityKind::Text,
        ..Default::default()
    };
    storage.configure_queue("null_q", config);

    // Null bytes should be accepted (encoded transparently in key)
    let msg = create_text_priority_message("null_q", "before\0after", "null_payload");
    storage.push(msg).unwrap();

    let m = storage.pop("null_q", "c1", 30).unwrap().unwrap();
    assert_eq!(m.priority, Priority::Text("before\0after".to_string()));
    assert_eq!(m.payload, "null_payload");
}

// ---------------------------------------------------------------------------
// Payload store integration (PER-0016, PER-0018)
// ---------------------------------------------------------------------------

/// Verifies that the payload round-trips through the mmap-based PayloadStore:
/// push externalizes the payload, pop resolves it, ack cleans up dedup correctly.
// Verifies: PER-0016, PER-0018
#[test]
fn test_payload_store_push_pop_ack_roundtrip() {
    let (storage, _temp_dir) = create_test_storage();

    let payload_text = "hello from the payload store";
    let msg = create_test_message("ps_queue", 50, payload_text);
    let msg_id = msg.id.clone();

    storage.push(msg).unwrap();

    // Pop should resolve the payload from the store.
    let popped = storage.pop("ps_queue", "consumer-1", 30).unwrap().unwrap();
    assert_eq!(popped.id, msg_id);
    assert_eq!(popped.payload, payload_text);

    // Ack should succeed (dedup hash cleanup works with externalized payloads).
    let acked = storage.ack("ps_queue", &msg_id, "consumer-1").unwrap();
    assert!(acked);
}

/// Verifies that duplicate detection works correctly with externalized payloads.
// Verifies: PER-0008, PER-0014, PER-0016
#[test]
fn test_payload_store_duplicate_detection_with_externalized_payloads() {
    let (storage, _temp_dir) = create_test_storage();

    storage.create_queue(
        "ps_nodup",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: false,
            ..Default::default()
        },
    );

    let msg1 = create_test_message("ps_nodup", 10, "unique-payload");
    storage.push(msg1).unwrap();

    // Second push with same payload should be rejected.
    let msg2 = create_test_message("ps_nodup", 20, "unique-payload");
    let result = storage.push(msg2);
    assert!(result.is_err());
    assert!(result.unwrap_err().to_string().contains("Duplicate"));
}

/// Verifies that after ack, the same payload can be pushed again (dedup hash
/// removal works correctly with externalized payload_hash field).
// Verifies: PER-0010, PER-0016
#[test]
fn test_payload_store_ack_allows_repush_of_same_payload() {
    let (storage, _temp_dir) = create_test_storage();

    storage.create_queue(
        "ps_repush",
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: false,
            ..Default::default()
        },
    );

    // Push, pop, ack.
    let msg = create_test_message("ps_repush", 10, "repush-me");
    let id = storage.push(msg).unwrap();
    storage.pop("ps_repush", "c1", 30).unwrap();
    storage.ack("ps_repush", &id, "c1").unwrap();

    // Should be able to push the same payload again.
    let msg2 = create_test_message("ps_repush", 10, "repush-me");
    assert!(storage.push(msg2).is_ok());
}

/// Verifies payload integrity survives a Storage restart (close + reopen).
// Verifies: PER-0018
#[test]
fn test_payload_store_survives_restart() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();
    let msg_id;

    {
        let storage = Storage::new(data_path).unwrap();
        let msg = create_test_message("ps_restart", 10, "persistent-payload");
        msg_id = msg.id.clone();
        storage.push(msg).unwrap();
    }
    // Storage dropped — RocksDB closed.

    {
        let storage = Storage::new(data_path).unwrap();
        let popped = storage.pop("ps_restart", "c1", 30).unwrap().unwrap();
        assert_eq!(popped.id, msg_id);
        assert_eq!(popped.payload, "persistent-payload");
    }
}

/// Verifies nack-retry round-trip: payload is still readable after unlock.
// Verifies: PER-0016, DLV-0004
#[test]
fn test_payload_store_nack_retry_preserves_payload() {
    let (storage, _temp_dir) = create_test_storage();

    let msg = create_test_message("ps_nack", 10, "nack-me");
    storage.push(msg).unwrap();

    let popped = storage.pop("ps_nack", "c1", 30).unwrap().unwrap();
    assert_eq!(popped.payload, "nack-me");

    // Nack for retry.
    storage.nack("ps_nack", &popped.id, "c1").unwrap();

    // Pop again — payload should still be readable.
    let popped2 = storage.pop("ps_nack", "c2", 30).unwrap().unwrap();
    assert_eq!(popped2.id, popped.id);
    assert_eq!(popped2.payload, "nack-me");
}

/// Regression test for stale-mmap bug: push many messages after a restart,
/// then pop them all — every payload must be intact, not empty.
// Verifies: PER-0016, PER-0018
#[test]
fn test_payload_store_push_pop_many_after_restart() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();

    // Phase 1: push some messages and close.
    {
        let storage = Storage::new(data_path).unwrap();
        for i in 0..10 {
            let msg = create_test_message("ps_restart_many", i, &format!("old-payload-{}", i));
            storage.push(msg).unwrap();
        }
    }

    // Phase 2: reopen, push MORE messages (these append to the active
    // segment file that was mmap'd at startup), then pop all of them.
    {
        let storage = Storage::new(data_path).unwrap();
        for i in 10..50 {
            let msg = create_test_message("ps_restart_many", i, &format!("new-payload-{}", i));
            storage.push(msg).unwrap();
        }

        // Pop all 50 and verify every payload is non-empty and correct.
        let mut popped = Vec::new();
        while let Some(msg) = storage.pop("ps_restart_many", "c1", 30).unwrap() {
            assert!(
                !msg.payload.is_empty(),
                "payload for message {} must not be empty (stale mmap bug)",
                msg.id
            );
            popped.push(msg);
        }
        assert_eq!(popped.len(), 50, "should pop all 50 messages");
    }
}

/// Verifies old-format messages (inline payload, no payload_ref) are NOT
/// corrupted by any code path: push, pop, nack-retry, pop again, ack.
// Verifies: PER-0018
#[test]
fn test_old_format_messages_survive_full_lifecycle() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();

    // Simulate a pre-0.19.0 message: write directly to RocksDB with
    // inline payload and no payload_ref/payload_hash fields.
    let old_msg_json = r#"{
        "id": "old-msg-001",
        "queue": "old_queue",
        "priority": 50,
        "payload": "{\"file\":\"/path/to/book.epub\"}",
        "created_at": "2026-01-01T00:00:00Z",
        "locked_until": null,
        "locked_by": null,
        "retry_count": 0,
        "max_retries": 3
    }"#;

    {
        // Write the old-format message directly to RocksDB, bypassing
        // push() which would externalize the payload.
        let mut opts = Options::default();
        opts.create_if_missing(true);
        let db = DB::open(&opts, format!("{}/messages", data_path)).unwrap();

        // Also write a queue config so Storage recognises the queue.
        let config = QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            ..Default::default()
        };
        let config_key = "_queue_config/old_queue";
        db.put(config_key.as_bytes(), serde_json::to_vec(&config).unwrap())
            .unwrap();

        let key = "old_queue/18446744073709551565/0000000000000000/old-msg-001";
        db.put(key.as_bytes(), old_msg_json.as_bytes()).unwrap();
    }

    // Reopen (simulates upgrade to 0.19.x).
    {
        let storage = Storage::new(data_path).unwrap();

        // Pop: must return the inline payload intact.
        let msg = storage.pop("old_queue", "c1", 30).unwrap().unwrap();
        assert_eq!(msg.id, "old-msg-001");
        assert_eq!(msg.payload, r#"{"file":"/path/to/book.epub"}"#);

        // Nack for retry.
        storage.nack("old_queue", "old-msg-001", "c1").unwrap();

        // Pop again: payload must still be intact.
        let msg2 = storage.pop("old_queue", "c2", 30).unwrap().unwrap();
        assert_eq!(msg2.payload, r#"{"file":"/path/to/book.epub"}"#);

        // Ack: clean up.
        storage.ack("old_queue", "old-msg-001", "c2").unwrap();
    }
}

/// Verifies mixed old-format and new-format messages coexist: old messages
/// keep inline payloads, new messages use the payload store.
// Verifies: PER-0016, PER-0018
#[test]
fn test_mixed_old_and_new_format_messages() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();

    // Phase 1: write old-format message directly to RocksDB.
    {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        let db = DB::open(&opts, format!("{}/messages", data_path)).unwrap();

        let config = QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            ..Default::default()
        };
        db.put(
            "_queue_config/mixed_q".as_bytes(),
            serde_json::to_vec(&config).unwrap(),
        )
        .unwrap();

        let old_json = r#"{
            "id": "old-001",
            "queue": "mixed_q",
            "priority": 100,
            "payload": "old-inline-payload",
            "created_at": "2026-01-01T00:00:00Z",
            "locked_until": null,
            "locked_by": null,
            "retry_count": 0,
            "max_retries": 3
        }"#;
        let key = "mixed_q/18446744073709551515/0000000000000000/old-001";
        db.put(key.as_bytes(), old_json.as_bytes()).unwrap();
    }

    // Phase 2: reopen, push a new-format message via the API.
    {
        let storage = Storage::new(data_path).unwrap();
        let new_msg = create_test_message("mixed_q", 200, "new-externalized-payload");
        storage.push(new_msg).unwrap();

        // Pop both: the higher-priority new message comes first (priority 200
        // > 100 in MaxFirst), then the old message.
        let msg1 = storage.pop("mixed_q", "c1", 30).unwrap().unwrap();
        assert_eq!(msg1.payload, "new-externalized-payload");

        let msg2 = storage.pop("mixed_q", "c1", 30).unwrap().unwrap();
        assert_eq!(msg2.id, "old-001");
        assert_eq!(msg2.payload, "old-inline-payload");
    }
}

/// After a restart with missing segment files, the integrity scan must delete
/// the broken messages and subsequent pushes + pops must work normally.
// Verifies: PER-0019
#[test]
fn test_broken_payload_ref_cleanup_and_fresh_push() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();
    let big = "x".repeat(5000); // above externalize threshold

    // Push messages that will become broken (large enough to be externalized).
    {
        let storage = Storage::new(data_path).unwrap();
        storage
            .push(create_test_message("broken_q", 100, &big))
            .unwrap();
        storage
            .push(create_test_message("broken_q", 50, &big))
            .unwrap();
    }

    // Remove segment files to simulate the scenario where payloads were
    // stored in RocksDB (QRUSTY_PAYLOAD_STORE=disabled) and segment files
    // never existed for these messages.
    let payloads_dir = temp_dir.path().join("payloads");
    std::fs::remove_dir_all(&payloads_dir).unwrap();

    // Reopen — integrity scan deletes broken messages, fresh segment created.
    let storage = Storage::new(data_path).unwrap();

    // Broken messages should be gone.
    let popped = storage.pop("broken_q", "c1", 30).unwrap();
    assert!(
        popped.is_none(),
        "broken messages should have been purged by integrity scan"
    );

    // Push a fresh message and verify it works end-to-end.
    storage
        .push(create_test_message("broken_q", 10, "fresh-message"))
        .unwrap();
    let popped = storage.pop("broken_q", "c1", 30).unwrap().unwrap();
    assert_eq!(popped.payload, "fresh-message");
}

/// When ALL messages in a queue have broken payload_refs, pop() must return
/// None (empty queue) after deleting them all.
// Verifies: PER-0019
#[test]
fn test_pop_returns_none_when_all_payloads_broken() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();

    {
        let storage = Storage::new(data_path).unwrap();
        let big = "x".repeat(5000); // above externalize threshold
        let msg = create_test_message("all_broken_q", 100, &big);
        storage.push(msg).unwrap();
    }

    // Delete segment files.
    let payloads_dir = temp_dir.path().join("payloads");
    for entry in std::fs::read_dir(&payloads_dir).unwrap() {
        let entry = entry.unwrap();
        if entry.file_name().to_string_lossy().starts_with("payload_") {
            std::fs::remove_file(entry.path()).unwrap();
        }
    }

    let storage = Storage::new(data_path).unwrap();
    let popped = storage.pop("all_broken_q", "c1", 30).unwrap();
    assert!(
        popped.is_none(),
        "pop must return None when all payloads are broken"
    );
}

/// The startup integrity scan must delete messages with broken payload_refs
/// so they don't clog queues.
// Verifies: PER-0019
#[test]
fn test_integrity_scan_deletes_broken_messages() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();

    // Push messages with externalized payloads (above threshold).
    {
        let big = "x".repeat(5000);
        let storage = Storage::new(data_path).unwrap();
        storage
            .push(create_test_message("scan_q", 100, &big))
            .unwrap();
        storage
            .push(create_test_message("scan_q", 50, &big))
            .unwrap();
    }

    // Delete segment files.
    let payloads_dir = temp_dir.path().join("payloads");
    for entry in std::fs::read_dir(&payloads_dir).unwrap() {
        let entry = entry.unwrap();
        if entry.file_name().to_string_lossy().starts_with("payload_") {
            std::fs::remove_file(entry.path()).unwrap();
        }
    }

    // Reopen — initialize_cache runs integrity scan which should delete broken msgs.
    let storage = Storage::new(data_path).unwrap();

    // Queue should be empty — broken messages were purged during startup.
    let stats = storage.get_queue_stats("scan_q").unwrap();
    assert_eq!(
        stats.available, 0,
        "broken messages should have been purged by integrity scan"
    );
}

/// Regression: a locked message whose payload segment was deleted must
/// decrement `locked` (not `available`) during the startup integrity
/// scan.  Previously it always decremented `available`, leaving a
/// phantom `locked=1` that made the queue appear permanently stuck.
// Verifies: PER-0019
#[test]
fn test_integrity_scan_decrements_locked_for_locked_broken_message() {
    let temp_dir = TempDir::new().unwrap();
    let data_path = temp_dir.path().to_str().unwrap();

    // Session 1: push a message with externalized payload, then lock it.
    {
        let big = "x".repeat(5000);
        let storage = Storage::new(data_path).unwrap();
        storage
            .push(create_test_message("phantom_q", 100, &big))
            .unwrap();
        // Pop to lock.
        let msg = storage.pop("phantom_q", "consumer1", 600).unwrap().unwrap();
        assert!(msg.locked_until.is_some());

        let stats = storage.get_queue_stats("phantom_q").unwrap();
        assert_eq!(stats.locked, 1);
        assert_eq!(stats.available, 0);
    }

    // Delete the payload segment files to simulate corruption.
    let payloads_dir = temp_dir.path().join("payloads");
    for entry in std::fs::read_dir(&payloads_dir).unwrap() {
        let entry = entry.unwrap();
        if entry.file_name().to_string_lossy().starts_with("payload_") {
            std::fs::remove_file(entry.path()).unwrap();
        }
    }

    // Session 2: reopen — integrity scan deletes the broken message.
    let storage = Storage::new(data_path).unwrap();

    // BOTH counters must be zero.  Before the fix, locked stayed at 1
    // because the scan always decremented available.
    let stats = storage.get_queue_stats("phantom_q").unwrap();
    assert_eq!(
        stats.locked, 0,
        "locked counter must be decremented when a locked message with a broken payload is deleted"
    );
    assert_eq!(stats.available, 0);
}

/// Stress test: push and pop 1000 messages rapidly, verify every payload.
// Verifies: PER-0016
#[test]
fn test_payload_store_stress_push_pop_1000() {
    let (storage, _temp_dir) = create_test_storage();

    let payloads: Vec<String> = (0..1000)
        .map(|i| format!("stress-payload-{:04}-{}", i, "x".repeat(100)))
        .collect();

    for (i, payload) in payloads.iter().enumerate() {
        let msg = create_test_message("stress_q", (1000 - i) as u64, payload);
        storage.push(msg).unwrap();
    }

    for expected_payload in &payloads {
        let msg = storage.pop("stress_q", "c1", 30).unwrap().unwrap();
        assert!(
            !msg.payload.is_empty(),
            "payload must not be empty in stress test"
        );
        assert_eq!(&msg.payload, expected_payload);
        storage.ack("stress_q", &msg.id, "c1").unwrap();
    }
}

/// After payload compaction, messages in the hot tier must still be readable.
/// Compaction rewrites segment files and deletes old ones — if the hot tier
/// holds stale payload_refs pointing to deleted segments, pop returns empty
/// payloads.
// Verifies: PER-0017, PER-0020
#[test]
fn test_compaction_does_not_break_hot_tier() {
    let (storage, _temp_dir) = create_test_storage();

    // Push messages with payloads above the externalization threshold (4096).
    let payloads: Vec<String> = (0..10)
        .map(|i| format!("payload-{}-{}", i, "x".repeat(5000)))
        .collect();
    for (i, payload) in payloads.iter().enumerate() {
        let msg = create_test_message("compact_q", (100 - i) as u64, payload);
        storage.push(msg).unwrap();
    }

    // Verify messages are in the hot tier by popping one.
    let popped = storage.pop("compact_q", "c1", 30).unwrap().unwrap();
    assert_eq!(popped.payload, payloads[0]);
    storage.ack("compact_q", &popped.id, "c1").unwrap();

    // Run compaction — this rewrites segment files and deletes old ones.
    let updated = storage.compact_payloads().unwrap();
    assert!(updated > 0, "compaction should have updated references");

    // Pop the remaining messages — they must still have valid payloads.
    // If the hot tier has stale payload_refs pointing to deleted segments,
    // this will fail.
    for (i, expected) in payloads.iter().enumerate().skip(1) {
        let popped = storage.pop("compact_q", "c1", 30).unwrap();
        assert!(popped.is_some(), "message {} should be available", i);
        let popped = popped.unwrap();
        assert_eq!(
            popped.payload, *expected,
            "message {} has wrong payload after compaction",
            i
        );
        storage.ack("compact_q", &popped.id, "c1").unwrap();
    }
}

/// Payloads smaller than the externalization threshold should stay inline
/// in RocksDB (no payload_ref).
// Verifies: PER-0020
#[test]
fn test_small_payloads_stay_inline() {
    let (storage, _temp_dir) = create_test_storage();

    // Push a small payload (well under the default 4096 threshold).
    let small = create_test_message("inline_q", 1, "tiny");
    storage.push(small).unwrap();

    // Pop it — payload_ref should be None (inline).
    let popped = storage.pop("inline_q", "c1", 30).unwrap().unwrap();
    assert_eq!(popped.payload, "tiny");
    assert!(
        popped.payload_ref.is_none(),
        "small payloads should stay inline, not externalized"
    );
}

/// Payloads at or above the externalization threshold should be externalized.
// Verifies: PER-0020
#[test]
fn test_large_payloads_are_externalized() {
    let (storage, _temp_dir) = create_test_storage();

    // Push a payload larger than the default 4096 threshold.
    let large_payload = "x".repeat(5000);
    let large = create_test_message("extern_q", 1, &large_payload);
    storage.push(large).unwrap();

    // Pop it — payload should be resolved correctly and payload_ref should
    // have been set (we can't check it directly after resolve clears it,
    // but the payload must be intact).
    let popped = storage.pop("extern_q", "c1", 30).unwrap().unwrap();
    assert_eq!(popped.payload, large_payload);
}

/// When all externalized payloads are acked (zero live refs), compact_payloads
/// must still clean up old segment files and their mmap regions.
// Verifies: PER-0017, PER-0021
#[test]
fn test_compact_with_zero_live_refs_cleans_segments() {
    let (storage, temp_dir) = create_test_storage();

    // Push large payloads to create segment files.
    let big = "x".repeat(5000);
    for i in 0..5 {
        let msg = create_test_message("zero_ref_q", (100 - i) as u64, &big);
        storage.push(msg).unwrap();
    }

    // Verify segment files exist.
    let payloads_dir = temp_dir.path().join("payloads");
    let segment_count_before = std::fs::read_dir(&payloads_dir)
        .unwrap()
        .filter(|e| {
            e.as_ref()
                .unwrap()
                .file_name()
                .to_string_lossy()
                .starts_with("payload_")
        })
        .count();
    assert!(segment_count_before > 0, "should have segment files");

    // Pop and ack all messages — no live refs remain.
    for _ in 0..5 {
        let popped = storage.pop("zero_ref_q", "c1", 30).unwrap().unwrap();
        storage.ack("zero_ref_q", &popped.id, "c1").unwrap();
    }

    // Run compaction with zero live refs.
    let updated = storage.compact_payloads().unwrap();
    assert_eq!(updated, 0, "no references to update when all acked");

    // Payload store disk usage should be zero (only a fresh empty active
    // segment file may remain).
    let store = storage.payload_store().unwrap();
    assert_eq!(
        store.disk_usage_bytes(),
        0,
        "disk usage should be 0 after cleaning empty segments"
    );

    // The old segment files (with data) should be gone.  Only a fresh
    // empty payload_0000.dat may exist as the reset active segment.
    let remaining_bytes: u64 = std::fs::read_dir(&payloads_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.file_name().to_string_lossy().starts_with("payload_"))
        .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0))
        .sum();
    assert_eq!(
        remaining_bytes, 0,
        "no payload data should remain on disk after cleanup"
    );
}

// ---------------------------------------------------------------------------
// force_unlock_queue tests (API-0014)
// ---------------------------------------------------------------------------

/// Verifies: API-0014 — force-unlock removes all locks on a queue.
#[test]
fn test_force_unlock_queue_unlocks_all_locked_messages() {
    let (storage, _temp_dir) = create_test_storage();

    // Push 4 messages and lock all of them.
    for i in 0..4 {
        storage
            .push(create_test_message("fu_q", 10 + i, &format!("msg-{}", i)))
            .unwrap();
    }
    let mut locked_ids = Vec::new();
    for _ in 0..4 {
        let msg = storage.pop("fu_q", "consumer1", 300).unwrap().unwrap();
        locked_ids.push(msg.id.clone());
    }

    // Verify all are locked.
    let stats = storage.get_queue_stats("fu_q").unwrap();
    assert_eq!(stats.locked, 4);
    assert_eq!(stats.available, 0);

    // Force-unlock.
    let unlocked = storage.force_unlock_queue_sync("fu_q").unwrap();
    assert_eq!(unlocked, 4);

    // Counters should reflect the unlock.
    let stats = storage.get_queue_stats("fu_q").unwrap();
    assert_eq!(stats.locked, 0);
    assert_eq!(stats.available, 4);

    // All messages should be poppable again.
    for _ in 0..4 {
        assert!(storage.pop("fu_q", "consumer2", 300).unwrap().is_some());
    }
}

/// Verifies: API-0014 — force-unlock on empty queue returns 0.
#[test]
fn test_force_unlock_queue_empty_returns_zero() {
    let (storage, _temp_dir) = create_test_storage();
    let unlocked = storage.force_unlock_queue_sync("nonexistent").unwrap();
    assert_eq!(unlocked, 0);
}

/// Verifies: API-0014 — force-unlock only unlocks locked messages, not available ones.
#[test]
fn test_force_unlock_queue_mixed_locked_and_available() {
    let (storage, _temp_dir) = create_test_storage();

    // Push 5 messages, lock only 3.
    for i in 0..5 {
        storage
            .push(create_test_message("fu_mix", 10 + i, &format!("msg-{}", i)))
            .unwrap();
    }
    for _ in 0..3 {
        storage.pop("fu_mix", "consumer1", 300).unwrap();
    }

    let stats = storage.get_queue_stats("fu_mix").unwrap();
    assert_eq!(stats.locked, 3);
    assert_eq!(stats.available, 2);

    let unlocked = storage.force_unlock_queue_sync("fu_mix").unwrap();
    assert_eq!(unlocked, 3);

    let stats = storage.get_queue_stats("fu_mix").unwrap();
    assert_eq!(stats.locked, 0);
    assert_eq!(stats.available, 5);
}

/// Verifies: API-0014 — force-unlock on one queue does not affect another.
#[test]
fn test_force_unlock_queue_isolation() {
    let (storage, _temp_dir) = create_test_storage();

    // Lock messages on two queues.
    for q in &["fu_iso_a", "fu_iso_b"] {
        for i in 0..2 {
            storage
                .push(create_test_message(q, 10 + i, &format!("{}-msg-{}", q, i)))
                .unwrap();
        }
        for _ in 0..2 {
            storage.pop(q, "consumer1", 300).unwrap();
        }
    }

    // Force-unlock only queue A.
    let unlocked = storage.force_unlock_queue_sync("fu_iso_a").unwrap();
    assert_eq!(unlocked, 2);

    // Queue B should still have its locks.
    let stats_b = storage.get_queue_stats("fu_iso_b").unwrap();
    assert_eq!(stats_b.locked, 2);
    assert_eq!(stats_b.available, 0);
}