qrusty 0.20.8

// src/storage.rs

//! # Storage Layer
//!
//! This module provides the persistent storage implementation for Qrusty using RocksDB.
//! It handles message persistence, priority ordering, and message lifecycle management.
//!
//! ## Key Design Decisions
//!
//! - **Priority Ordering**: Messages are stored with inverted priority keys for natural sorting
//! - **Key Format**: `queue_name/priority_inverted/timestamp/uuid` ensures correct ordering
//! - **Atomic Operations**: All storage operations are atomic via RocksDB transactions
//! - **Lock Management**: In-memory index tracks locked messages for timeout monitoring
//! - **Dead Letter Queue**: Failed messages are moved to `_dlq/original_queue/message_id`

use anyhow::Result;
use chrono::{DateTime, Utc};
use rocksdb::{BlockBasedOptions, Cache, Options, WriteBatch, DB};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
use std::sync::Arc;
use std::sync::{Mutex, RwLock};

use crate::message::{
    BatchAckResult, BatchNackResult, Message, Priority, PriorityOrdering, QueueConfig,
};
use crate::payload_store::PayloadStore;

/// Computes a 128-bit xxh3 hash of a payload string, returned as a
/// fixed-size byte array.  Used for O(1) duplicate detection with
/// bounded memory: storing `[u8; 16]` instead of the full payload
/// string reduces per-entry overhead from kilobytes to 16 bytes.
///
/// Collision probability is ~2⁻⁶⁴ at the birthday bound — negligible
/// for any realistic queue depth.
// Implements: PER-0014
fn hash_payload(payload: &str) -> [u8; 16] {
    xxhash_rust::xxh3::xxh3_128(payload.as_bytes()).to_ne_bytes()
}

/// Internal pad width for byte-complement hex encoding of text priority keys.
/// All keys are padded to at least this width with `\x00` before complementing
/// so that shorter strings sort after longer ones sharing the same prefix.
const TEXT_PRIORITY_PAD: usize = 512;

/// Percent-encodes characters that conflict with the storage key separator.
///
/// The key format uses `/` as a field separator (`queue/priority/ts/uuid`),
/// so literal `/` inside the text priority must be escaped.  We also escape
/// `%` (the escape character itself) and `\0` (null byte) for safety.
///
/// This encoding preserves the original lexicographic ordering among encoded
/// strings because `%` (0x25) sorts lower than `/` (0x2F), and both encoded
/// forms (`%25` for `%`, `%2F` for `/`) maintain their relative byte order.
fn encode_text_priority(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for b in s.bytes() {
        match b {
            b'%' => out.push_str("%25"),
            b'/' => out.push_str("%2F"),
            0 => out.push_str("%00"),
            _ => out.push(b as char),
        }
    }
    out
}

/// Encodes a string as byte-complemented hex for MaxFirst lexicographic ordering.
fn byte_complement_hex(s: &str) -> String {
    let bytes = s.as_bytes();
    let pad_len = TEXT_PRIORITY_PAD.max(bytes.len());
    let mut buf = vec![0u8; pad_len];
    buf[..bytes.len()].copy_from_slice(bytes);
    buf.iter().map(|b| format!("{:02x}", b ^ 0xFF)).collect()
}

#[derive(Debug)]
pub enum RenameQueueError {
    NotFound,
    AlreadyExists,
    Storage(anyhow::Error),
}

impl std::fmt::Display for RenameQueueError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RenameQueueError::NotFound => write!(f, "queue not found"),
            RenameQueueError::AlreadyExists => write!(f, "target queue already exists"),
            RenameQueueError::Storage(err) => write!(f, "storage error: {err}"),
        }
    }
}

impl std::error::Error for RenameQueueError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            RenameQueueError::Storage(err) => Some(err.as_ref()),
            _ => None,
        }
    }
}

impl From<anyhow::Error> for RenameQueueError {
    fn from(value: anyhow::Error) -> Self {
        RenameQueueError::Storage(value)
    }
}

/// Per-queue available/locked message counts cached in memory.
///
/// Implements: SYS-0018
///
/// Seeded once from a full DB scan at startup and updated incrementally
/// by every state-changing storage operation.  All reads of queue stats
/// (`get_queue_stats`, `get_all_queue_stats`, `list_queues`) go through
/// this cache, avoiding any RocksDB scan.
#[derive(Debug, Clone, Default)]
pub struct QueueCounts {
    pub available: u64,
    pub locked: u64,
}

/// Secondary index mapping `(queue, message_id) -> storage_key` for O(1)
/// lookup of currently-locked messages.  Implements: DLV-0014.
pub type LockedIdIndex = HashMap<String, HashMap<String, String>>;

/// Persistent storage layer for messages using RocksDB.
///
/// The Storage struct provides a high-performance, ACID-compliant storage backend
/// for the priority queue system. It uses RocksDB for persistence and maintains
/// an in-memory index for tracking locked messages.
///
/// # Architecture
///
/// - **Primary Storage**: RocksDB database for message persistence
/// - **Lock Index**: In-memory HashMap tracking message locks and timeouts
/// - **Key Ordering**: Custom key format ensures priority-based message retrieval
/// - **Compression**: LZ4 compression for storage efficiency
/// - **Durability**: Write-ahead logging (WAL) enabled for crash recovery
///
/// # Examples
///
/// ```rust,no_run
/// use qrusty::storage::Storage;
/// use std::sync::Arc;
///
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// // Initialize storage
/// let storage = Storage::new("/data/qrusty")?;
///
/// // Storage is thread-safe and can be shared via Arc
/// let storage = Arc::new(storage);
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct Storage {
    /// Main RocksDB instance for message persistence
    /// Thread-safe via Arc, supports concurrent reads/writes
    messages: Arc<DB>,

    /// In-memory index mapping message keys to lock expiration times
    /// Used by timeout_monitor to efficiently find expired locks
    /// Format: "queue/priority/timestamp/uuid" -> `DateTime<Utc>`
    locked_index: Arc<RwLock<HashMap<String, DateTime<Utc>>>>,

    /// Secondary index mapping (queue, message_id) -> storage_key for
    /// O(1) lookup of locked messages in `ack`, `nack`, and `renew`.
    /// Without this, those paths do a full prefix scan of the queue
    /// and deserialize every entry until the target is found — fine
    /// on small queues, catastrophic on queues with millions of
    /// messages (e.g. a renew for an already-acked message walks the
    /// entire prefix before returning `Ok(false)`).
    ///
    /// Populated on `pop()` when a message becomes locked.  Removed on
    /// `ack`, `nack`, `unlock_expired_messages`, and `force_unlock_queue`.
    /// Shares the same `locked_index_cap` limit and the
    /// `untracked_locks_possible` sticky-fallback semantics: an index
    /// miss with the flag set falls back to the full prefix scan.
    ///
    /// Outer key: queue name.  Inner key: message_id.  Value: storage key.
    /// The two-level layout makes `force_unlock_queue_sync` cleanup O(1)
    /// in queues rather than O(total locked messages).
    ///
    /// Implements: DLV-0014
    locked_id_index: Arc<RwLock<LockedIdIndex>>,

    /// Queue configurations mapping queue names to their settings
    /// Format: queue_name -> QueueConfig
    /// Used to determine priority ordering when storing/retrieving messages
    queue_configs: Arc<RwLock<HashMap<String, QueueConfig>>>,

    /// Per-queue payload hash sets for O(1) duplicate detection (PER-0008).
    /// Only populated for queues whose `allow_duplicates` is `false`.
    /// Includes payloads of both locked and unlocked messages (DLV-0010).
    /// Stores 128-bit xxh3 hashes instead of full payload strings (PER-0014).
    /// Format: queue_name -> set of payload hashes currently in the queue
    payload_sets: Arc<RwLock<HashMap<String, HashSet<[u8; 16]>>>>,

    /// Per-queue mutexes that serialise `pop()` calls within each queue.
    ///
    /// RocksDB does not provide read-modify-write atomicity at the application
    /// level: two concurrent `pop()` callers can both observe the same message
    /// as unlocked and lock it, causing double-delivery.  Holding this per-queue
    /// mutex for the duration of a `pop()` prevents the race (DLV-0011).
    /// Cross-queue parallelism is preserved — only the mutex for the affected
    /// queue is acquired.
    pop_locks: Arc<RwLock<HashMap<String, Arc<Mutex<()>>>>>,

    /// In-memory cache of per-queue available/locked counts (SYS-0018).
    ///
    /// Seeded once from a full scan in `new()`, then updated incrementally by
    /// push, pop, ack, nack, batch ops, unlock_expired, delete, and purge.
    /// Query methods read exclusively from this cache.
    queue_counters: Arc<Mutex<HashMap<String, QueueCounts>>>,

    /// Per-queue hot tier of the highest-priority **available** messages
    /// (SYS-0020).  Pop operations serve from here first, falling back to a
    /// RocksDB prefix scan only when the hot tier is empty.  When the tier
    /// drops below `hot_tier_refill_threshold` messages, a background refill
    /// tops it up to `hot_tier_capacity` from RocksDB.
    ///
    /// Key: storage key (same format as RocksDB).  Value: serialised message
    /// bytes.  Using `Vec<u8>` avoids deserialising messages until they are
    /// actually popped.
    #[allow(clippy::type_complexity)]
    hot_tier: Arc<RwLock<HashMap<String, BTreeMap<String, Vec<u8>>>>>,

    /// External payload store (PER-0016).  When present, message payloads
    /// are stored in append-only mmap'd files and only a `PayloadRef` is
    /// kept in RocksDB.  The OS page cache decides which payloads stay
    /// resident.
    payload_store: Option<Arc<PayloadStore>>,

    /// Maximum number of available messages to keep in each queue's hot tier.
    /// Configurable via `QRUSTY_HOT_TIER_SIZE` (default 1000).
    // Implements: SYS-0021
    pub hot_tier_capacity: usize,

    /// When a queue's hot tier drops to this count, trigger a refill.
    /// Configurable via `QRUSTY_REFILL_THRESHOLD` (default 250).
    // Implements: SYS-0021
    hot_tier_refill_threshold: usize,

    /// Maximum entries in the locked_index before inserts are skipped.
    /// Messages beyond this cap are still locked in RocksDB; the timeout
    /// monitor's fallback full scan will catch their expiration.
    /// Configurable via `QRUSTY_MAX_LOCKED_INDEX` (default 500_000).
    locked_index_cap: usize,

    /// Sticky flag: set to true whenever an insertion into `locked_index`
    /// is skipped because the index is at cap.  The next
    /// `unlock_expired_messages` call will perform a full RocksDB scan to
    /// find expired locks that are not tracked in the index, even if the
    /// index has since dropped below cap.  Cleared after a successful
    /// full scan completes.
    ///
    /// Implements: DLV-0013
    ///
    /// Without this flag, the window between (a) `pop()` skipping the
    /// index insert while at cap and (b) the index dropping below cap
    /// before the next sweep would leave locks stranded forever, because
    /// the sweeper would only do a full scan when currently at cap.
    untracked_locks_possible: Arc<AtomicBool>,

    /// RocksDB block cache handle, retained so we can dynamically resize
    /// it under memory pressure (e.g. shrink from 256 MB to 64 MB).
    /// Wrapped in Arc<Mutex<>> because `Cache::set_capacity` takes `&mut self`
    /// and `Storage` derives `Clone`.
    rocksdb_cache: Arc<Mutex<Cache>>,

    /// The configured (full) block cache size in bytes, used to restore
    /// capacity after pressure subsides.
    rocksdb_cache_capacity: usize,

    /// Minimum payload size in bytes before externalizing to the payload
    /// store.  Payloads smaller than this stay inline in RocksDB.
    /// Configurable via `QRUSTY_EXTERNALIZE_MIN_BYTES` (default 4096).
    externalize_min_bytes: usize,
}

/// Internal target for batch_nack: a located message plus its storage key.
struct BatchNackTarget {
    key: String,
    msg: Message,
}

impl Storage {
    // Implements: PER-0006, PER-0007
    fn dedupe_unlocked_messages_by_payload(&self, queue_name: &str) -> Result<usize> {
        let prefix = format!("{}/", queue_name);
        let now = Utc::now();

        // Use payload hashes (16 bytes each) instead of full payload strings
        // to avoid temporarily inflating memory during reconfiguration.
        let mut seen_hashes: HashSet<[u8; 16]> = HashSet::new();
        let mut keys_to_delete: Vec<Vec<u8>> = Vec::new();
        let mut keys_to_delete_str: Vec<String> = Vec::new();

        let iter = self.messages.prefix_iterator(prefix.as_bytes());
        for item in iter {
            let (key, value) = item?;
            let key_str = String::from_utf8_lossy(&key);

            if !key_str.starts_with(&prefix) {
                break;
            }

            // Ignore internal keys.
            if key_str.starts_with("_queue_config/") || key_str.starts_with("_dlq/") {
                continue;
            }

            let msg: Message = serde_json::from_slice(&value)?;

            // Only de-dupe unlocked/available messages.
            if let Some(locked_until) = msg.locked_until {
                if locked_until > now {
                    continue;
                }
            }

            let h = Self::get_payload_hash(&msg);
            if !seen_hashes.insert(h) {
                keys_to_delete.push(key.to_vec());
                keys_to_delete_str.push(key_str.to_string());
            }
        }

        if keys_to_delete.is_empty() {
            return Ok(0);
        }

        let mut batch = WriteBatch::default();
        for key in &keys_to_delete {
            batch.delete(key);
        }
        self.messages.write(batch)?;

        // Best-effort cleanup: avoid the timeout monitor chasing deleted keys.
        {
            let mut locked_index = self.locked_index.write().unwrap();
            for key in keys_to_delete_str {
                locked_index.remove(&key);
            }
        }

        // SYS-0018: dedupe only removes unlocked/available messages, so
        // `available -= N`.  The locked count is not affected because
        // locked messages were skipped during the scan.
        let removed = keys_to_delete.len();
        {
            let mut counters = self.queue_counters.lock().unwrap();
            if let Some(entry) = counters.get_mut(queue_name) {
                entry.available = entry.available.saturating_sub(removed as u64);
            }
        }

        Ok(removed)
    }

    pub fn queue_exists(&self, queue_name: &str) -> Result<bool> {
        if self.queue_configs.read().unwrap().contains_key(queue_name) {
            return Ok(true);
        }

        let prefix = format!("{}/", queue_name);
        if let Some(item) = self.messages.prefix_iterator(prefix.as_bytes()).next() {
            let (key, _value) = item?;
            let key_str = String::from_utf8_lossy(&key);
            if key_str.starts_with(&prefix) {
                return Ok(true);
            }
        }

        let dlq_prefix = format!("_dlq/{}/", queue_name);
        if let Some(item) = self.messages.prefix_iterator(dlq_prefix.as_bytes()).next() {
            let (key, _value) = item?;
            let key_str = String::from_utf8_lossy(&key);
            if key_str.starts_with(&dlq_prefix) {
                return Ok(true);
            }
        }

        Ok(false)
    }

    /// Renames a queue and moves all messages (and DLQ entries) to the new name.
    ///
    /// This is an atomic-ish best effort: it uses RocksDB write batches, but the
    /// overall operation may span multiple batches for large queues.
    pub fn rename_queue(&self, from: &str, to: &str) -> std::result::Result<(), RenameQueueError> {
        if from.trim().is_empty() || to.trim().is_empty() {
            return Err(RenameQueueError::Storage(anyhow::anyhow!(
                "queue names cannot be empty"
            )));
        }

        if from == to {
            return Ok(());
        }

        if !self.queue_exists(from)? {
            return Err(RenameQueueError::NotFound);
        }
        if self.queue_exists(to)? {
            return Err(RenameQueueError::AlreadyExists);
        }

        // Preserve configuration on rename.
        let existing_config = self.get_queue_config(from);

        let now = Utc::now();
        let mut batch = WriteBatch::default();
        // (old_key, new_key, locked_until) for still-locked messages re-keyed
        // into the new queue.  Collecting old keys here lets us remove the
        // exact entries from `locked_index` below without retain-scanning
        // the entire cross-queue map.
        let mut moved_locked: Vec<(String, String, DateTime<Utc>)> = Vec::new();
        // DLV-0014: (message_id, new_storage_key) for still-locked
        // messages that got re-keyed into the new queue.  Used below
        // to rebuild the secondary index under the new queue name.
        let mut moved_locked_ids: Vec<(String, String)> = Vec::new();
        let mut ops_in_batch: usize = 0;

        let prefix_from = format!("{}/", from);
        let iter = self.messages.prefix_iterator(prefix_from.as_bytes());
        for item in iter {
            let (key, value) = item.map_err(|e| RenameQueueError::Storage(e.into()))?;
            let key_str = String::from_utf8_lossy(&key);

            if !key_str.starts_with(&prefix_from) {
                break;
            }
            if key_str.starts_with("_queue_config/") || key_str.starts_with("_dlq/") {
                continue;
            }

            let old_key = key_str.to_string();
            let mut msg: Message = serde_json::from_slice(&value).map_err(|e| {
                RenameQueueError::Storage(anyhow::anyhow!("deserialize message failed: {e}"))
            })?;
            msg.queue = to.to_string();

            let new_key = self.generate_message_key(&msg, &existing_config);
            let new_value = serde_json::to_vec(&msg).map_err(|e| {
                RenameQueueError::Storage(anyhow::anyhow!("serialize message failed: {e}"))
            })?;

            batch.put(new_key.as_bytes(), &new_value);
            batch.delete(&key);
            ops_in_batch += 2;

            if let Some(locked_until) = msg.locked_until {
                if locked_until > now {
                    moved_locked.push((old_key, new_key.clone(), locked_until));
                    moved_locked_ids.push((msg.id.clone(), new_key.clone()));
                }
            }

            if ops_in_batch >= 2000 {
                self.messages
                    .write(batch)
                    .map_err(|e| RenameQueueError::Storage(e.into()))?;
                batch = WriteBatch::default();
                ops_in_batch = 0;
            }
        }

        // Move DLQ entries
        let dlq_prefix_from = format!("_dlq/{}/", from);
        let dlq_iter = self.messages.prefix_iterator(dlq_prefix_from.as_bytes());
        for item in dlq_iter {
            let (key, value) = item.map_err(|e| RenameQueueError::Storage(e.into()))?;
            let key_str = String::from_utf8_lossy(&key);
            if !key_str.starts_with(&dlq_prefix_from) {
                break;
            }

            // Preserve message id suffix if present
            let suffix = key_str
                .strip_prefix(&dlq_prefix_from)
                .unwrap_or_default()
                .to_string();

            let mut msg: Message = serde_json::from_slice(&value).map_err(|e| {
                RenameQueueError::Storage(anyhow::anyhow!("deserialize DLQ message failed: {e}"))
            })?;
            msg.queue = to.to_string();
            let new_value = serde_json::to_vec(&msg).map_err(|e| {
                RenameQueueError::Storage(anyhow::anyhow!("serialize DLQ message failed: {e}"))
            })?;

            let new_key = format!("_dlq/{}/{}", to, suffix);
            batch.put(new_key.as_bytes(), &new_value);
            batch.delete(&key);
            ops_in_batch += 2;

            if ops_in_batch >= 2000 {
                self.messages
                    .write(batch)
                    .map_err(|e| RenameQueueError::Storage(e.into()))?;
                batch = WriteBatch::default();
                ops_in_batch = 0;
            }
        }

        // Update queue config keys (persisted)
        let config_key_from = format!("_queue_config/{}", from);
        let config_key_to = format!("_queue_config/{}", to);
        let config_json = serde_json::to_vec(&existing_config).map_err(|e| {
            RenameQueueError::Storage(anyhow::anyhow!("serialize config failed: {e}"))
        })?;

        batch.delete(config_key_from.as_bytes());
        batch.put(config_key_to.as_bytes(), &config_json);
        ops_in_batch += 2;

        // Flush remaining batch
        if ops_in_batch > 0 {
            self.messages
                .write(batch)
                .map_err(|e| RenameQueueError::Storage(e.into()))?;
        }

        // Update in-memory configs
        {
            let mut configs = self.queue_configs.write().unwrap();
            configs.remove(from);
            configs.insert(to.to_string(), existing_config);
        }

        // Update locked_index: remove the exact old keys we re-keyed and
        // insert the new ones.  Precisely targeted rather than a
        // retain() pass so rename cost is O(locked_in_from_queue)
        // instead of O(total_locked_across_all_queues).
        {
            let mut index = self.locked_index.write().unwrap();
            for (old_key, new_key, until) in moved_locked {
                index.remove(&old_key);
                index.insert(new_key, until);
            }
        }

        // DLV-0014: rebuild the secondary index under the new queue
        // name.  Drop the entire `from` sub-map and insert the moved
        // (msg_id, new_storage_key) pairs into the `to` sub-map.
        {
            let mut id_index = self.locked_id_index.write().unwrap();
            id_index.remove(from);
            if !moved_locked_ids.is_empty() {
                let entry = id_index.entry(to.to_string()).or_default();
                for (msg_id, new_key) in moved_locked_ids {
                    entry.insert(msg_id, new_key);
                }
            }
        }

        // Move payload set to the new queue name (PER-0010).
        {
            let mut sets = self.payload_sets.write().unwrap();
            if let Some(set) = sets.remove(from) {
                sets.insert(to.to_string(), set);
            }
        }

        // SYS-0018: move counter entry to new name.
        {
            let mut counters = self.queue_counters.lock().unwrap();
            if let Some(counts) = counters.remove(from) {
                counters.insert(to.to_string(), counts);
            }
        }

        // Rebuild hot tier for the new queue name (keys contain queue name
        // as a prefix, so we can't simply move the BTreeMap).
        {
            self.hot_tier.write().unwrap().remove(from);
            self.refill_hot_tier(to);
        }

        Ok(())
    }

    /// Opens RocksDB and loads queue configurations, but does **not** yet
    /// rebuild the payload-dedup sets or seed the queue counters.  Those two
    /// full-database scans are deferred to [`Storage::initialize_cache`] so
    /// that the HTTP server can start accepting connections (and serving
    /// `/health`) while the heavy work runs in the background.
    ///
    /// Call [`Storage::new`] when you need a fully-ready instance in one step
    /// (e.g. in tests).
    // Implements: SYS-0019
    pub fn open(data_path: &str) -> Result<Self> {
        let mut opts = Options::default();
        opts.create_if_missing(true);
        opts.set_compression_type(rocksdb::DBCompressionType::Lz4);

        // Cap open SST file handles.  Each open file holds a table reader
        // with index/filter metadata in memory.  With millions of keys and
        // hundreds of SST files, a high limit causes unbounded memory growth.
        // 256 is enough for good read performance; RocksDB re-opens files
        // as needed (slightly slower on cache-miss reads).
        let max_open_files: i32 = std::env::var("ROCKSDB_MAX_OPEN_FILES")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(128);
        opts.set_max_open_files(max_open_files);

        // Use direct I/O for flush and compaction output to bypass the
        // kernel page cache.  This prevents compaction from inflating
        // the page cache with transient SST data that competes with
        // the container memory limit (PER-0015).
        opts.set_use_direct_io_for_flush_and_compaction(true);

        // Enable WAL for durability
        opts.set_manual_wal_flush(false);
        opts.set_use_fsync(true); // Use fsync for better durability

        // Cap write-side memory: each memtable up to 16 MB, at most 2 active.
        // This bounds write-buffer RAM to ~32 MB per column family.
        let write_buffer_mb: usize = std::env::var("ROCKSDB_WRITE_BUFFER_MB")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(16);
        opts.set_write_buffer_size(write_buffer_mb * 1024 * 1024);
        opts.set_max_write_buffer_number(2);

        // Cap the RocksDB block cache (read-only cache of decompressed SST
        // blocks).  This bounds memory without affecting durability — data is
        // already persisted via WAL + fsync.  Default 64 MB; override with
        // ROCKSDB_CACHE_MB.
        // Implements: PER-0015
        let cache_mb: usize = std::env::var("ROCKSDB_CACHE_MB")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(128);
        let cache = Cache::new_lru_cache(cache_mb * 1024 * 1024);
        let mut block_opts = BlockBasedOptions::default();
        block_opts.set_block_cache(&cache);
        // Force index and filter blocks into the block cache instead of
        // letting them grow unbounded in separate memory.  Without this,
        // index/filter blocks for 10M+ keys can consume several GB.
        block_opts.set_cache_index_and_filter_blocks(true);
        // Pin L0 index/filter blocks in the cache so they aren't evicted
        // under pressure (L0 is read on every query).
        block_opts.set_pin_l0_filter_and_index_blocks_in_cache(true);
        opts.set_block_based_table_factory(&block_opts);
        tracing::info!(
            "RocksDB block cache: {} MB (includes index/filter)",
            cache_mb
        );

        let messages = Arc::new(DB::open(&opts, format!("{}/messages", data_path))?);

        // Hot tier configuration (SYS-0021).
        let hot_tier_capacity: usize = std::env::var("QRUSTY_HOT_TIER_SIZE")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(1000);
        let hot_tier_refill_threshold: usize = std::env::var("QRUSTY_REFILL_THRESHOLD")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(250);
        tracing::info!(
            "Hot tier: capacity={}, refill_threshold={}",
            hot_tier_capacity,
            hot_tier_refill_threshold
        );

        let locked_index_cap: usize = std::env::var("QRUSTY_MAX_LOCKED_INDEX")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(500_000);

        // Open the payload store for mmap-based payload storage (PER-0016).
        // Set QRUSTY_PAYLOAD_STORE=disabled to force inline payloads.
        // Even when disabled, existing payload_ref messages will be logged
        // as errors on read (payload data is unrecoverable without the store).
        let payload_store_disabled = std::env::var("QRUSTY_PAYLOAD_STORE")
            .ok()
            .map(|v| v == "disabled")
            .unwrap_or(false);
        let payload_store = if payload_store_disabled {
            tracing::warn!(
                "Payload store DISABLED via QRUSTY_PAYLOAD_STORE=disabled. \
                 New payloads will be stored inline in RocksDB."
            );
            None
        } else {
            match PayloadStore::open(&std::path::Path::new(data_path).join("payloads")) {
                Ok(ps) => {
                    tracing::info!("Payload store opened at {}/payloads", data_path);
                    Some(Arc::new(ps))
                }
                Err(e) => {
                    tracing::warn!("Payload store disabled: {}", e);
                    None
                }
            }
        };

        let rocksdb_cache_capacity = cache_mb * 1024 * 1024;

        let storage = Self {
            messages,
            locked_index: Arc::new(RwLock::new(HashMap::new())),
            locked_id_index: Arc::new(RwLock::new(HashMap::new())),
            queue_configs: Arc::new(RwLock::new(HashMap::new())),
            payload_sets: Arc::new(RwLock::new(HashMap::new())),
            pop_locks: Arc::new(RwLock::new(HashMap::new())),
            queue_counters: Arc::new(Mutex::new(HashMap::new())),
            hot_tier: Arc::new(RwLock::new(HashMap::new())),
            hot_tier_capacity,
            hot_tier_refill_threshold,
            payload_store,
            locked_index_cap,
            // Start as true so the first sweep after startup does a full
            // scan, picking up any locks that persisted across restarts
            // without being in the (now empty) in-memory index.
            untracked_locks_possible: Arc::new(AtomicBool::new(true)),
            rocksdb_cache: Arc::new(Mutex::new(cache)),
            rocksdb_cache_capacity,
            externalize_min_bytes: std::env::var("QRUSTY_EXTERNALIZE_MIN_BYTES")
                .ok()
                .and_then(|v| v.parse().ok())
                .unwrap_or(4096),
        };

        // Load persisted queue configurations (fast: only config keys).
        let load_result = {
            #[cfg(test)]
            {
                if std::env::var("QRUSTY_TEST_FORCE_LOAD_QUEUE_CONFIGS_ERROR").is_ok() {
                    Err(anyhow::anyhow!("forced load_queue_configs error"))
                } else {
                    storage.load_queue_configs()
                }
            }

            #[cfg(not(test))]
            {
                storage.load_queue_configs()
            }
        };

        if let Err(e) = load_result {
            tracing::warn!("Failed to load queue configurations: {}", e);
        }

        Ok(storage)
    }

    /// Rebuilds the payload-dedup sets and seeds the queue counters by
    /// performing full database scans.  This is the slow part of startup and
    /// is intended to be called from a background blocking task after the HTTP
    /// server is already listening, so that `/health` remains responsive while
    /// the work is in progress.
    // Implements: SYS-0019
    ///
    /// After this method returns the storage instance is fully operational.
    pub fn initialize_cache(&self) -> Result<()> {
        tracing::info!("Storage cache initialisation starting (payload sets + queue counters)…");

        // Reconstruct payload sets from DB for all no-dup queues (PER-0009).
        if let Err(e) = self.load_payload_sets() {
            tracing::warn!("Failed to load payload sets: {}", e);
        }

        // Seed in-memory queue counters from a single full scan (SYS-0018).
        if let Err(e) = self.seed_queue_counters() {
            tracing::warn!("Failed to seed queue counters: {}", e);
        }

        // Flush after counter seeding to free the write buffer before the
        // next phase.
        let _ = self.messages.flush();

        // PER-0019: purge broken payload refs.  Does NOT re-inline payloads
        // to avoid inflating the RocksDB write buffer with full payloads
        // (which can push the process past the container memory limit).
        self.check_payload_integrity();
        let _ = self.messages.flush();

        // Hot tiers are NOT seeded at startup.  pop() lazily fills them
        // on first access via refill_hot_tier(), avoiding the large
        // memory spike of loading hot_tier_capacity × N queues × message
        // bytes during initialisation.  The first few pops per queue will
        // fall back to the RocksDB prefix scan slow path; after that the
        // hot tier is populated and subsequent pops are fast.

        tracing::info!("Storage cache initialisation complete — server is fully ready.");
        Ok(())
    }

    /// Convenience wrapper used by tests: opens and fully initialises storage
    /// in one synchronous call.  Production code should use [`Storage::open`]
    /// followed by [`Storage::initialize_cache`] in a background task.
    #[cfg_attr(not(test), allow(dead_code))]
    pub fn new(data_path: &str) -> Result<Self> {
        let storage = Self::open(data_path)?;
        storage.initialize_cache()?;
        Ok(storage)
    }

    /// Reconstructs the in-memory payload sets for every no-dup queue by scanning
    /// RocksDB on startup.  Includes both locked and unlocked messages (PER-0009,
    /// DLV-0010).  Called synchronously from `new()` after `load_queue_configs()`.
    // Implements: PER-0009
    fn load_payload_sets(&self) -> Result<()> {
        let configs = self
            .queue_configs
            .try_read()
            .map_err(|_| anyhow::anyhow!("queue_configs lock contention in load_payload_sets"))?;
        let mut sets = self
            .payload_sets
            .try_write()
            .map_err(|_| anyhow::anyhow!("payload_sets lock contention in load_payload_sets"))?;

        for (queue_name, config) in configs.iter() {
            if config.allow_duplicates {
                continue;
            }
            let prefix = format!("{}/", queue_name);
            let mut set: HashSet<[u8; 16]> = HashSet::new();
            let iter = self.messages.prefix_iterator(prefix.as_bytes());
            for item in iter {
                let (key, value) = item?;
                let key_str = String::from_utf8_lossy(&key);
                if !key_str.starts_with(&prefix) {
                    break;
                }
                // Exclude internal entries that share the queue prefix.
                if key_str.starts_with("_queue_config/") || key_str.starts_with("_dlq/") {
                    continue;
                }
                if let Ok(msg) = serde_json::from_slice::<Message>(&value) {
                    set.insert(Self::get_payload_hash(&msg));
                }
            }
            sets.insert(queue_name.clone(), set);
        }
        Ok(())
    }

    /// Seeds the in-memory queue counters by scanning all messages once (SYS-0018).
    ///
    /// Called from `new()` at startup.  After this, all counter updates are
    /// incremental — no further full scans are needed during normal operation.
    // Implements: SYS-0018
    fn seed_queue_counters(&self) -> Result<()> {
        let now = Utc::now();
        let mut counters: HashMap<String, QueueCounts> = HashMap::new();

        let iter = self.messages.iterator(rocksdb::IteratorMode::Start);
        for item in iter {
            let (key, value) = item?;
            let key_str = String::from_utf8_lossy(&key);

            // Skip internal keys.
            if key_str.starts_with("_dlq/") || key_str.starts_with("_queue_config/") {
                continue;
            }

            let parts: Vec<&str> = key_str.split('/').collect();
            if parts.len() < 4 {
                continue;
            }

            let queue_name = parts[0].to_string();
            let entry = counters.entry(queue_name).or_default();

            match serde_json::from_slice::<Message>(&value) {
                Ok(msg) => {
                    if msg.locked_until.is_some_and(|lu| lu > now) {
                        entry.locked += 1;
                    } else {
                        entry.available += 1;
                    }
                }
                Err(e) => {
                    tracing::warn!("seed_queue_counters: skipping key {}: {}", key_str, e);
                }
            }
        }

        // Also include configured queues that currently have no messages.
        let configs = self.queue_configs.read().unwrap();
        for queue_name in configs.keys() {
            counters.entry(queue_name.clone()).or_default();
        }

        *self.queue_counters.lock().unwrap() = counters;
        Ok(())
    }

    /// Scans all messages for broken payload references and logs a summary.
    /// Also re-inlines any resolvable externalized payloads back into RocksDB
    /// so they don't depend on the payload store for reads.
    // Implements: PER-0019
    fn check_payload_integrity(&self) {
        let mut total_refs = 0u64;
        let mut resolvable = 0u64;
        let mut broken = 0u64;
        let mut deleted = 0u64;

        let iter = self.messages.iterator(rocksdb::IteratorMode::Start);
        for item in iter {
            let (key, value) = match item {
                Ok(kv) => kv,
                Err(_) => continue,
            };
            let key_str = String::from_utf8_lossy(&key);
            if key_str.starts_with("_queue_config/") {
                continue;
            }
            let is_dlq = key_str.starts_with("_dlq/");

            let mut msg: Message = match serde_json::from_slice(&value) {
                Ok(m) => m,
                Err(_) => continue,
            };

            if msg.payload_ref.is_some() && msg.payload.is_empty() {
                total_refs += 1;
                if self.resolve_payload(&mut msg) {
                    resolvable += 1;
                    // NOTE: we intentionally do NOT re-inline payloads here.
                    // Re-inlining writes full payload bytes back into RocksDB,
                    // inflating the write buffer and potentially pushing the
                    // process past the container memory limit during startup.
                    // The payload_ref stays — payloads are resolved at pop time.
                } else {
                    broken += 1;
                    // PER-0019: delete messages with unrecoverable payloads
                    // so they don't clog the queue on every consume attempt.
                    if self.messages.delete(&key).is_ok() {
                        deleted += 1;
                        // DLQ entries don't participate in dedup sets or
                        // counters, so only update those for normal messages.
                        if !is_dlq {
                            let queue = &msg.queue;
                            let mut sets = self.payload_sets.write().unwrap();
                            if let Some(set) = sets.get_mut(queue) {
                                set.remove(&Self::get_payload_hash(&msg));
                            }
                            // SYS-0018: decrement the correct counter
                            // depending on whether the deleted message
                            // was locked or available.  Previously this
                            // always decremented `available`, which left
                            // a phantom `locked=1` when the deleted
                            // message was actually locked — making the
                            // queue appear stuck forever.
                            let now = chrono::Utc::now();
                            let mut counters = self.queue_counters.lock().unwrap();
                            if let Some(entry) = counters.get_mut(queue) {
                                if msg.locked_until.is_some_and(|lu| lu > now) {
                                    entry.locked = entry.locked.saturating_sub(1);
                                } else {
                                    entry.available = entry.available.saturating_sub(1);
                                }
                            }
                        }
                    }
                }
            }
        }

        if total_refs > 0 {
            tracing::info!(
                total_refs = total_refs,
                resolvable = resolvable,
                broken = broken,
                deleted = deleted,
                "Payload integrity check complete"
            );
            if broken > 0 {
                tracing::error!(
                    broken = broken,
                    deleted = deleted,
                    "BROKEN payload references found — messages with \
                     unrecoverable payloads have been deleted"
                );
            }
        }
    }

    /// Refills a single queue's hot tier from RocksDB up to `hot_tier_capacity`.
    ///
    /// Only **available** (unlocked or lock-expired) messages are loaded.
    /// Called when the hot tier drops below `hot_tier_refill_threshold` after
    /// a pop, or during rename/reconfiguration.
    // Implements: SYS-0020
    fn refill_hot_tier(&self, queue: &str) {
        let now = Utc::now();
        let prefix = format!("{}/", queue);
        let mut tier = BTreeMap::new();
        let iter = self.messages.prefix_iterator(prefix.as_bytes());

        for item in iter {
            let (key, value) = match item {
                Ok(kv) => kv,
                Err(_) => break,
            };
            let key_str = String::from_utf8_lossy(&key);
            if !key_str.starts_with(&prefix) {
                break;
            }
            // Skip locked messages.
            if let Ok(msg) = serde_json::from_slice::<Message>(&value) {
                if msg.locked_until.is_some_and(|lu| lu > now) {
                    continue;
                }
            } else {
                continue;
            }
            tier.insert(key_str.into_owned(), value.to_vec());
            if tier.len() >= self.hot_tier_capacity {
                break;
            }
        }

        self.hot_tier
            .write()
            .unwrap()
            .insert(queue.to_string(), tier);
    }

    /// Shrinks all hot tiers to `target_size` entries, evicting the
    /// lowest-priority messages.  Called by the memory monitor when
    /// memory pressure is detected (SYS-0022).
    #[allow(dead_code)]
    pub fn shrink_hot_tiers(&self, target_size: usize) {
        let mut tiers = self.hot_tier.write().unwrap();
        for (_, tier) in tiers.iter_mut() {
            while tier.len() > target_size {
                tier.pop_last(); // remove lowest-priority (last key)
            }
        }
    }

    /// Light memory release for **Warning** pressure: flush RocksDB
    /// memtables and shrink the block cache.  Does NOT touch the hot tier
    /// (clearing it is counter-productive — it forces cache-miss storms
    /// that re-inflate the block cache immediately).
    pub fn release_memory_warning(&self) {
        // Flush memtables to disk so their RAM can be freed.
        if let Err(e) = self.messages.flush() {
            tracing::warn!("RocksDB flush during pressure release failed: {}", e);
        }
        // Shrink block cache to 25% of configured capacity.
        let reduced = self.rocksdb_cache_capacity / 4;
        self.rocksdb_cache.lock().unwrap().set_capacity(reduced);
        // Reclaim per-queue structures for empty queues and shrink
        // over-allocated containers (SYS-0025).
        self.release_memory_for_empty_queues();
        // Compact RocksDB to reclaim tombstone space from acked/deleted
        // messages.  Uses None..None for full range (SYS-0025).
        self.messages.compact_range(None::<&[u8]>, None::<&[u8]>);
        tracing::info!(
            "Memory release (warning): flushed RocksDB, shrunk block cache to {} MB, compacted range, reclaimed idle structures",
            reduced / (1024 * 1024)
        );
    }

    /// Aggressive memory release for **Critical** pressure: everything in
    /// `release_memory_warning` plus shrinking hot tiers and compacting
    /// payloads.
    pub fn release_memory_critical(&self) {
        // All Warning-level actions first (flush, shrink cache to 25%,
        // reclaim empty queues, shrink_to_fit, compact_range).
        self.release_memory_warning();
        // Then escalate: block cache to absolute minimum.
        self.rocksdb_cache.lock().unwrap().set_capacity(1024 * 1024);
        // Shrink (don't clear) hot tiers to 10% of capacity.
        let target = std::cmp::max(self.hot_tier_capacity / 10, 1);
        self.shrink_hot_tiers(target);
        tracing::info!(
            "Memory release (critical): block cache → 1 MB, hot tiers → {}",
            target
        );
    }

    /// Reclaims memory from idle per-queue structures (SYS-0025).
    ///
    /// For empty queues (available=0, locked=0):
    ///   - Clears + shrinks dedup HashSets
    ///   - Removes hot tier entries
    ///   - Removes stale locked_index entries
    ///
    /// For all queues:
    ///   - Calls shrink_to_fit() on dedup HashSets
    ///
    /// Also shrinks the global locked_index HashMap.
    // Implements: SYS-0025
    pub fn release_memory_for_empty_queues(&self) {
        // Collect empty queue names from counters.
        let empty_queues: Vec<String> = {
            let counters = self.queue_counters.lock().unwrap();
            counters
                .iter()
                .filter(|(_, c)| c.available == 0 && c.locked == 0)
                .map(|(name, _)| name.clone())
                .collect()
        };

        // Clean up empty queues.
        if !empty_queues.is_empty() {
            // Remove hot tier entries for empty queues.
            {
                let mut tiers = self.hot_tier.write().unwrap();
                for q in &empty_queues {
                    tiers.remove(q);
                }
            }

            // Clear + shrink dedup sets for empty queues.
            {
                let mut sets = self.payload_sets.write().unwrap();
                for q in &empty_queues {
                    if let Some(set) = sets.get_mut(q) {
                        set.clear();
                        set.shrink_to_fit();
                    }
                }
            }

            // Remove locked_index entries belonging to empty queues.
            {
                let mut li = self.locked_index.write().unwrap();
                li.retain(|key, _| {
                    let queue = key.split('/').next().unwrap_or("");
                    !empty_queues.iter().any(|eq| eq == queue)
                });
            }

            tracing::info!(
                "Memory reclaim: cleaned {} empty queue(s): {:?}",
                empty_queues.len(),
                empty_queues
            );
        }

        // Shrink all dedup sets (including non-empty ones).
        {
            let mut sets = self.payload_sets.write().unwrap();
            for (_, set) in sets.iter_mut() {
                set.shrink_to_fit();
            }
        }

        // Shrink the global locked_index.
        {
            let mut li = self.locked_index.write().unwrap();
            li.shrink_to_fit();
        }
    }

    /// Restores RocksDB block cache to its full configured capacity.
    /// Called when memory pressure subsides.
    pub fn restore_cache_capacity(&self) {
        self.rocksdb_cache
            .lock()
            .unwrap()
            .set_capacity(self.rocksdb_cache_capacity);
        tracing::info!(
            "Memory restored: block cache → {} MB",
            self.rocksdb_cache_capacity / (1024 * 1024)
        );
    }

    /// Returns extended storage metrics for the /stats endpoint.
    pub fn extended_metrics(&self) -> serde_json::Value {
        let hot_tier_sizes: std::collections::HashMap<String, usize> = {
            let tiers = self.hot_tier.read().unwrap();
            tiers.iter().map(|(q, t)| (q.clone(), t.len())).collect()
        };

        let locked_index_size = self.locked_index.read().unwrap().len();

        let payload_store_disk_bytes = self
            .payload_store
            .as_ref()
            .map(|s| s.disk_usage_bytes())
            .unwrap_or(0);

        serde_json::json!({
            "hot_tier_sizes": hot_tier_sizes,
            "locked_index_size": locked_index_size,
            "payload_store_disk_bytes": payload_store_disk_bytes
        })
    }

    /// Returns a memory breakdown for inclusion in pressure log lines.
    /// Shows where memory is allocated so operators can see whether release
    /// actions are effective (SYS-0022).
    pub fn memory_breakdown(&self) -> std::collections::HashMap<&'static str, usize> {
        let mut m = std::collections::HashMap::new();
        m.insert(
            "block_cache_capacity_mb",
            self.rocksdb_cache_capacity / (1024 * 1024),
        );
        let hot_tier_entries: usize = self
            .hot_tier
            .read()
            .unwrap()
            .values()
            .map(|t| t.len())
            .sum();
        m.insert("hot_tier_entries", hot_tier_entries);
        let dedup_entries: usize = self
            .payload_sets
            .read()
            .unwrap()
            .values()
            .map(|s| s.len())
            .sum();
        m.insert("dedup_set_entries", dedup_entries);
        m.insert(
            "locked_index_entries",
            self.locked_index.read().unwrap().len(),
        );
        m
    }

    /// Returns a reference to the payload store, if available.
    pub fn payload_store(&self) -> Option<&Arc<PayloadStore>> {
        self.payload_store.as_ref()
    }

    /// Runs payload store compaction: collects all live PayloadRefs from
    /// RocksDB, rewrites them to a new segment, and updates references.
    // Implements: PER-0017
    pub fn compact_payloads(&self) -> Result<usize> {
        let store = match self.payload_store {
            Some(ref s) => s,
            None => return Ok(0),
        };

        // Collect all live PayloadRefs from RocksDB.
        let mut live_refs = Vec::new();
        let mut ref_keys: Vec<(Vec<u8>, crate::payload_store::PayloadRef)> = Vec::new();
        let iter = self.messages.iterator(rocksdb::IteratorMode::Start);
        for item in iter {
            let (key, value) = item?;
            let key_str = String::from_utf8_lossy(&key);
            if key_str.starts_with("_queue_config/") || key_str.starts_with("_dlq/") {
                continue;
            }
            if let Ok(msg) = serde_json::from_slice::<Message>(&value) {
                if let Some(pref) = msg.payload_ref {
                    live_refs.push(pref.clone());
                    ref_keys.push((key.to_vec(), pref));
                }
            }
        }

        // PER-0021: Even with zero live refs, call compact() so it can
        // clean up old segment files and mmap regions.
        let ref_map = store.compact(&live_refs)?;

        if live_refs.is_empty() {
            return Ok(0);
        }

        // Update RocksDB references atomically.
        let mut batch = WriteBatch::default();
        let mut updated = 0;
        for (key, old_ref) in &ref_keys {
            if let Some(new_ref) = ref_map.get(&(old_ref.file_id, old_ref.offset)) {
                if let Ok(Some(value)) = self.messages.get(key) {
                    if let Ok(mut msg) = serde_json::from_slice::<Message>(&value) {
                        msg.payload_ref = Some(new_ref.clone());
                        if let Ok(new_value) = serde_json::to_vec(&msg) {
                            batch.put(key, new_value);
                            updated += 1;
                        }
                    }
                }
            }
        }
        if !batch.is_empty() {
            self.messages.write(batch)?;
        }

        // PER-0020: clear hot tiers so they get repopulated from RocksDB
        // with the new payload_refs.  Without this, cached messages still
        // reference the old (now-deleted) segment files.
        {
            let mut tiers = self.hot_tier.write().unwrap();
            tiers.clear();
        }

        tracing::info!("Payload compaction: {} references updated", updated);
        Ok(updated)
    }

    /// Loads persisted queue configurations from the database into memory.
    fn load_queue_configs(&self) -> Result<()> {
        let iter = self.messages.iterator(rocksdb::IteratorMode::Start);

        for item in iter {
            let (key, value) = item?;
            let key_str = String::from_utf8_lossy(&key);

            // Look for queue config keys
            if !key_str.starts_with("_queue_config/") {
                continue;
            }

            let queue_name = &key_str["_queue_config/".len()..];
            if queue_name.is_empty() {
                continue;
            }

            let Ok(config) = serde_json::from_slice::<QueueConfig>(&value) else {
                continue;
            };

            // Use blocking write since this is called during initialization
            if let Ok(mut configs) = self.queue_configs.try_write() {
                configs.insert(queue_name.to_string(), config);
            }
        }

        Ok(())
    }

    /// Creates or updates the configuration for a queue.
    ///
    /// This method sets behavioral settings for a queue.
    ///
    /// Queue ordering (the queue "type") is immutable after a queue exists. If a queue
    /// already exists, attempts to change `ordering` are ignored and the existing ordering
    /// is preserved.
    ///
    /// # Arguments
    ///
    /// * `queue_name` - Name of the queue to configure
    /// * `config` - Configuration settings for the queue
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::{storage::Storage, message::{QueueConfig, PriorityOrdering}};
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// // Create a min-first priority queue
    /// let config = QueueConfig {
    ///     ordering: PriorityOrdering::MinFirst,
    ///     ..Default::default()
    /// };
    /// storage.configure_queue("priority_queue", config);
    ///
    /// // Create a max-first priority queue (default)
    /// let config = QueueConfig {
    ///     ordering: PriorityOrdering::MaxFirst,
    ///     ..Default::default()
    /// };
    /// storage.configure_queue("urgent_queue", config);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Queue Configuration Persistence
    ///
    /// Queue configurations are currently stored in memory and will be reset on
    /// restart. In a production deployment, you may want to persist these configurations
    /// to storage or configure them during application startup.
    ///
    /// # Performance — dedup rebuild cost
    ///
    /// When toggling `allow_duplicates` from `true` to `false`, the queue's
    /// `payload_sets` entry must be rebuilt from scratch because no dedup
    /// set was maintained during the permissive phase.  This rebuild is
    /// an inherent O(queue_depth) prefix scan with payload-hash work per
    /// message — unavoidable without pre-computing the set eagerly.  It
    /// is a **one-shot cost per config change**, not a per-request cost:
    /// ordinary publish/consume traffic remains O(1).  Avoid toggling
    /// this flag on queues with millions of messages at peak load.
    pub fn configure_queue(&self, queue_name: &str, config: QueueConfig) {
        let old_config = self.get_queue_config(queue_name);
        let existed = self.queue_exists(queue_name).unwrap_or(false);

        let mut effective_config = config;
        if existed {
            effective_config.ordering = old_config.ordering;
        }

        // Store in memory
        self.queue_configs
            .write()
            .unwrap()
            .insert(queue_name.to_string(), effective_config.clone());

        // Ensure queue_counters has an entry so /stats and /queues see it
        self.queue_counters
            .lock()
            .unwrap()
            .entry(queue_name.to_string())
            .or_default();

        // Persist to database using a special key prefix for queue configs
        let config_key = format!("_queue_config/{}", queue_name);
        if let Ok(config_json) = serde_json::to_vec(&effective_config) {
            let _ = self.messages.put(config_key.as_bytes(), &config_json);
        }

        // If we just disabled duplicates, de-dupe existing unlocked messages and
        // build the payload set from the resulting queue state (PER-0006, PER-0008).
        if old_config.allow_duplicates && !effective_config.allow_duplicates {
            match self.dedupe_unlocked_messages_by_payload(queue_name) {
                Ok(removed) => {
                    if removed > 0 {
                        tracing::info!(
                            "De-duplicated {} unlocked message(s) in queue '{}' after disabling duplicates",
                            removed,
                            queue_name
                        );
                    }
                }
                Err(e) => {
                    tracing::error!(
                        "Failed to de-duplicate unlocked messages for queue '{}': {}",
                        queue_name,
                        e
                    );
                }
            }

            // Rebuild the payload set from the post-dedupe queue state.
            let prefix = format!("{}/", queue_name);
            let mut set: HashSet<[u8; 16]> = HashSet::new();
            let iter = self.messages.prefix_iterator(prefix.as_bytes());
            for (key, value) in iter.flatten() {
                let key_str = String::from_utf8_lossy(&key);
                if !key_str.starts_with(&prefix) {
                    break;
                }
                if let Ok(msg) = serde_json::from_slice::<Message>(&value) {
                    set.insert(Self::get_payload_hash(&msg));
                }
            }
            self.payload_sets
                .write()
                .unwrap()
                .insert(queue_name.to_string(), set);
        }

        // If we just enabled duplicates, the payload set is no longer needed (PER-0007).
        if !old_config.allow_duplicates && effective_config.allow_duplicates {
            self.payload_sets.write().unwrap().remove(queue_name);
        }

        // Rebuild hot tier after reconfiguration (dedup may have removed
        // messages, or ordering may have changed).
        self.refill_hot_tier(queue_name);
    }

    /// Creates a new queue with the specified configuration.
    ///
    /// This is a convenience method that sets up a queue with explicit configuration.
    /// You can also configure queues by calling `configure_queue` directly.
    ///
    /// # Arguments
    ///
    /// * `queue_name` - Name of the queue to create
    /// * `config` - Configuration for the queue's behavior
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::{storage::Storage, message::{QueueConfig, PriorityOrdering}};
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// // Create a queue with min-first ordering
    /// let config = QueueConfig {
    ///     ordering: PriorityOrdering::MinFirst,
    ///     ..Default::default()
    /// };
    /// storage.create_queue("low_priority_first", config);
    ///
    /// // Create a queue with max-first ordering
    /// let config = QueueConfig {
    ///     ordering: PriorityOrdering::MaxFirst,
    ///     ..Default::default()
    /// };
    /// storage.create_queue("high_priority_first", config);
    /// # Ok(())
    /// # }
    /// ```
    pub fn create_queue(&self, queue_name: &str, config: QueueConfig) {
        self.configure_queue(queue_name, config);
    }

    /// Gets the configuration for a queue, returning default if not configured.
    ///
    /// If a queue has not been explicitly configured, this returns the default
    /// configuration with `MaxFirst` ordering for backward compatibility.
    ///
    /// # Arguments
    ///
    /// * `queue_name` - Name of the queue to get configuration for
    ///
    /// # Returns
    ///
    /// The queue configuration, or default configuration if not set.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::{storage::Storage, message::PriorityOrdering};
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// // Get config for unconfigured queue (returns default)
    /// let config = storage.get_queue_config("new_queue");
    /// assert_eq!(config.ordering, PriorityOrdering::MaxFirst);
    /// # Ok(())
    /// # }
    /// ```
    pub fn get_queue_config(&self, queue_name: &str) -> QueueConfig {
        self.queue_configs
            .read()
            .unwrap()
            .get(queue_name)
            .cloned()
            .unwrap_or_default()
    }

    /// Checks if a duplicate payload exists in the queue among unlocked messages.
    ///
    /// Scans the queue for any unlocked message with an identical payload.
    /// This is used when `allow_duplicates` is false to prevent duplicate payloads.
    ///
    /// # Arguments
    ///
    /// * `queue` - Name of the queue to check
    /// * `payload` - The payload to check for duplicates
    /// * `exclude_id` - Optional message ID to exclude from check (used during unlock)
    ///
    /// # Returns
    ///
    /// `Ok(true)` if a duplicate payload exists among unlocked messages.
    ///
    /// This is retained as a diagnostic / test-only helper.  Production code
    /// uses `payload_sets` for O(1) duplicate detection (PER-0008).
    #[allow(dead_code)]
    pub fn has_duplicate_payload(
        &self,
        queue: &str,
        payload: &str,
        exclude_id: Option<&str>,
    ) -> Result<bool> {
        let prefix = format!("{}/", queue);
        let now = Utc::now();
        let iter = self.messages.prefix_iterator(prefix.as_bytes());

        for item in iter {
            let (key, value) = item?;
            let key_str = String::from_utf8_lossy(&key);

            if !key_str.starts_with(&prefix) {
                break;
            }

            let mut msg: Message = serde_json::from_slice(&value)?;

            // Skip if this is the message we're excluding
            if let Some(exclude) = exclude_id {
                if msg.id == exclude {
                    continue;
                }
            }

            // Skip locked messages (they're not "available")
            if let Some(locked_until) = msg.locked_until {
                if locked_until > now {
                    continue;
                }
            }

            // Resolve externalized payload before comparing.
            self.resolve_payload(&mut msg);

            // Check for matching payload
            if msg.payload == payload {
                return Ok(true);
            }
        }

        Ok(false)
    }

    /// Generates a storage key for a message based on queue configuration.
    ///
    /// The key format ensures correct priority ordering based on the queue's
    /// configuration:
    /// - MaxFirst: `queue/priority_inverted/timestamp/uuid` (higher priority sorts first)
    /// - MinFirst: `queue/priority_normal/timestamp/uuid` (lower priority sorts first)
    ///
    /// # Arguments
    ///
    /// * `msg` - The message to generate a key for
    /// * `config` - The queue configuration determining ordering
    ///
    /// # Returns
    ///
    /// A storage key that ensures correct priority ordering.
    fn generate_message_key(&self, msg: &Message, config: &QueueConfig) -> String {
        let ts = msg.created_at.timestamp_nanos_opt().unwrap_or(0);
        match &msg.priority {
            Priority::Numeric(n) => {
                let priority_key = match config.ordering {
                    PriorityOrdering::MaxFirst => u64::MAX - n,
                    PriorityOrdering::MinFirst => *n,
                    PriorityOrdering::Fifo => 0,
                };
                format!("{}/{:020}/{:016}/{}", msg.queue, priority_key, ts, msg.id)
            }
            Priority::Text(s) => {
                let key_segment = match config.ordering {
                    PriorityOrdering::MaxFirst => format!("s:{}", byte_complement_hex(s)),
                    PriorityOrdering::MinFirst => format!("s:{}", encode_text_priority(s)),
                    PriorityOrdering::Fifo => "s:".to_string(),
                };
                format!("{}/{}/{:016}/{}", msg.queue, key_segment, ts, msg.id)
            }
        }
    }

    /// Adds a new message to the queue with priority ordering.
    ///
    /// Messages are stored using a composite key that ensures priority-based
    /// retrieval based on the queue's configuration:
    /// - MaxFirst queues: `queue_name/priority_inverted/timestamp/uuid`
    /// - MinFirst queues: `queue_name/priority_normal/timestamp/uuid`
    ///
    /// The ordering ensures that messages are naturally sorted by queue and priority
    /// according to the queue's configured ordering preference.
    ///
    /// # Arguments
    ///
    /// * `msg` - The message to store
    ///
    /// # Returns
    ///
    /// Returns the message ID on successful storage.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use chrono::Utc;
    /// use qrusty::{storage::Storage, message::{Message, Priority, QueueConfig, PriorityOrdering}};
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// // Configure queue for min-first ordering
    /// let config = QueueConfig {
    ///     ordering: PriorityOrdering::MinFirst,
    ///     ..Default::default()
    /// };
    /// storage.configure_queue("orders", config);
    ///
    /// let msg = Message {
    ///     id: "msg-123".to_string(),
    ///     queue: "orders".to_string(),
    ///     priority: Priority::Numeric(100),
    ///     payload: r#"{"order_id": 456}"#.to_string(),
    ///     created_at: Utc::now(),
    ///     locked_until: None,
    ///     locked_by: None,
    ///     retry_count: 0,
    ///     max_retries: 3,
    ///     payload_ref: None,
    ///     payload_hash: None,
    /// };
    ///
    /// let id = storage.push(msg)?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Queue Configuration
    ///
    /// If the queue has not been explicitly configured, it will use the default
    /// `MaxFirst` ordering for backward compatibility.
    ///
    /// # Errors
    ///
    /// - Serialization failures
    /// - RocksDB write errors
    /// - Disk space issues
    /// - Duplicate payload when `allow_duplicates` is false
    ///
    /// Returns the dedup hash for a message — either from the cached
    /// `payload_hash` field (externalized messages) or by computing it
    /// from the inline payload.
    fn get_payload_hash(msg: &Message) -> [u8; 16] {
        msg.payload_hash
            .unwrap_or_else(|| hash_payload(&msg.payload))
    }

    /// Resolves a message's payload from the external PayloadStore if it
    /// has a `payload_ref` and the inline payload is empty.  Old messages
    /// with inline payloads are returned as-is (backward compatible).
    ///
    /// Returns `true` if the payload is available (either inline or
    /// successfully resolved), `false` if the payload is unrecoverable.
    // Implements: PER-0016, PER-0018, PER-0019
    fn resolve_payload(&self, msg: &mut Message) -> bool {
        let pref = match msg.payload_ref {
            Some(ref p) if msg.payload.is_empty() => p,
            _ => return true, // inline payload or already resolved
        };

        match self.payload_store {
            Some(ref store) => match store.read(pref) {
                Some(data) => {
                    msg.payload = String::from_utf8_lossy(&data).into_owned();
                    true
                }
                None => {
                    tracing::error!(
                        msg_id = msg.id.as_str(),
                        queue = msg.queue.as_str(),
                        file_id = pref.file_id,
                        offset = pref.offset,
                        length = pref.length,
                        "Failed to read payload from store — segment file may be missing or corrupted"
                    );
                    false
                }
            },
            None => {
                tracing::error!(
                    msg_id = msg.id.as_str(),
                    queue = msg.queue.as_str(),
                    "Message has payload_ref but payload store is disabled — payload unrecoverable"
                );
                false
            }
        }
    }

    /// Requirements: PER-0001, PER-0002, PER-0003, PER-0004, PER-0008, PER-0010, PER-0012, DLV-0010, SCH-0001, SCH-0002
    pub fn push(&self, mut msg: Message) -> Result<String> {
        // Get queue configuration to determine ordering
        let config = self.get_queue_config(&msg.queue);

        // Enforce priority kind matches queue configuration
        if msg.priority.kind() != config.priority_kind {
            return Err(anyhow::anyhow!(
                "Priority kind mismatch: queue expects {:?} but got {:?}",
                config.priority_kind,
                msg.priority.kind()
            ));
        }

        // Validate text priority constraints
        if let Priority::Text(ref s) = msg.priority {
            if s.is_empty() {
                return Err(anyhow::anyhow!("Text priority must not be empty"));
            }
        }

        // Compute dedup hash from the original payload (PER-0014).
        // This must happen before externalization clears the payload.
        let payload_hash = hash_payload(&msg.payload);

        if !config.allow_duplicates {
            // Optimistic reservation strategy (PER-0012):
            //   1. Reserve slot in payload set (under write lock).
            //   2. Release write lock.
            //   3. Externalize payload + perform disk I/O without holding
            //      any in-memory lock.
            //   4. On failure, compensate by removing the reservation.
            {
                let mut sets = self.payload_sets.write().unwrap();
                let set = sets.entry(msg.queue.clone()).or_default();
                if set.contains(&payload_hash) {
                    return Err(anyhow::anyhow!("Duplicate payload rejected"));
                }
                // Reserve before releasing lock.
                set.insert(payload_hash);
            }
            // Lock released — externalize payload, then write to RocksDB.
            self.externalize_payload(&mut msg, payload_hash)?;
            let key = self.generate_message_key(&msg, &config);
            let value = serde_json::to_vec(&msg)?;
            if let Err(e) = self.messages.put(&key, value) {
                // Compensate: remove reservation so future pushes are not blocked.
                let mut sets = self.payload_sets.write().unwrap();
                if let Some(set) = sets.get_mut(&msg.queue) {
                    set.remove(&payload_hash);
                }
                return Err(e.into());
            }
            // SYS-0018: available++
            self.queue_counters
                .lock()
                .unwrap()
                .entry(msg.queue.clone())
                .or_default()
                .available += 1;
            // Insert into hot tier if it qualifies (SYS-0020).
            self.try_insert_hot_tier(&msg.queue, &key, &msg);
            return Ok(msg.id);
        }

        // allow_duplicates queue — no set maintenance needed.
        self.externalize_payload(&mut msg, payload_hash)?;
        let key = self.generate_message_key(&msg, &config);
        let value = serde_json::to_vec(&msg)?;
        self.messages.put(&key, value)?;
        // SYS-0018: available++
        self.queue_counters
            .lock()
            .unwrap()
            .entry(msg.queue.clone())
            .or_default()
            .available += 1;
        // Insert into hot tier if it qualifies (SYS-0020).
        self.try_insert_hot_tier(&msg.queue, &key, &msg);
        Ok(msg.id)
    }

    /// Moves the message payload into the PayloadStore if available and the
    /// payload is large enough to justify externalization.  Small payloads
    /// stay inline in RocksDB to avoid segment-file overhead.  No-op if
    /// the payload store is disabled.
    // Implements: PER-0016, PER-0020
    fn externalize_payload(&self, msg: &mut Message, payload_hash: [u8; 16]) -> Result<()> {
        if let Some(ref store) = self.payload_store {
            if msg.payload.len() < self.externalize_min_bytes {
                return Ok(()); // keep inline
            }
            let pref = store.append(msg.payload.as_bytes())?;
            msg.payload_ref = Some(pref);
            msg.payload_hash = Some(payload_hash);
            msg.payload = String::new();
        }
        Ok(())
    }

    /// Inserts a newly pushed message into the hot tier if the tier is not
    /// full, or if the message sorts before the tier's last (lowest-priority)
    /// entry (in which case the last entry is evicted).
    // Implements: SYS-0020
    fn try_insert_hot_tier(&self, queue: &str, key: &str, msg: &Message) {
        let value = match serde_json::to_vec(msg) {
            Ok(v) => v,
            Err(_) => return,
        };
        let mut tiers = self.hot_tier.write().unwrap();
        let tier = tiers.entry(queue.to_string()).or_default();
        if tier.len() < self.hot_tier_capacity {
            tier.insert(key.to_string(), value);
        } else if let Some(last_key) = tier.keys().next_back().cloned() {
            if key < last_key.as_str() {
                // New message has higher priority — evict lowest.
                tier.remove(&last_key);
                tier.insert(key.to_string(), value);
            }
        }
    }

    /// Retrieves and locks the highest priority available message from a queue.
    ///
    /// This method implements the core queue pop operation:
    /// 1. Scans messages in priority order (highest first)
    /// 2. Skips locked messages or those with expired locks
    /// 3. Locks the first available message for the specified consumer
    /// 4. Updates retry count and lock metadata
    /// 5. Adds to lock index for timeout monitoring
    ///
    /// # Arguments
    ///
    /// * `queue` - Name of the queue to consume from
    /// * `consumer_id` - Unique identifier for the consuming client
    /// * `timeout_secs` - How long to lock the message (seconds)
    ///
    /// # Returns
    ///
    /// - `Ok(Some(Message))` - Successfully retrieved and locked a message
    /// - `Ok(None)` - No messages available in the queue
    /// - `Err(...)` - Database or serialization error
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::storage::Storage;
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// // Try to get a message with 30 second timeout
    /// if let Some(msg) = storage.pop("orders", "worker-1", 30)? {
    ///     println!("Got message: {}", msg.payload);
    ///     // Process the message...
    ///     // Then ack or nack it
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Lock Behavior
    ///
    /// - Messages are locked exclusively to the consumer
    /// - Lock expires after `timeout_secs` if not ack'd or nack'd
    /// - Retry count is incremented on each pop operation
    /// - Expired locks are automatically released by timeout monitor
    ///
    /// Requirements: DLV-0001, DLV-0002, DLV-0011, SCH-0001, SCH-0002, SCH-0003, SCH-0004, API-0003, PER-0019
    pub fn pop(
        &self,
        queue: &str,
        consumer_id: &str,
        timeout_secs: u64,
    ) -> Result<Option<Message>> {
        // Acquire the per-queue pop mutex before scanning (DLV-0011).
        //
        // RocksDB provides no read-modify-write atomicity at the application
        // level.  Without this lock two concurrent pop() calls can both observe
        // the same message as available and both lock it, causing double-delivery.
        // By serialising pop() calls per queue we prevent the race while keeping
        // concurrent pops on different queues fully parallel.
        let queue_lock = {
            // Fast path: lock already exists — just clone it.
            let map = self.pop_locks.read().unwrap();
            map.get(queue).cloned()
        };
        let queue_lock = match queue_lock {
            Some(l) => l,
            None => {
                // Slow path: create entry under write lock.
                let mut map = self.pop_locks.write().unwrap();
                map.entry(queue.to_string())
                    .or_insert_with(|| Arc::new(Mutex::new(())))
                    .clone()
            }
        };
        let _guard = queue_lock.lock().unwrap();

        let now = Utc::now();
        let lock_until = now + chrono::Duration::seconds(timeout_secs as i64);

        // Retry loop: if a popped message has a broken payload_ref (missing
        // segment file), delete it and try the next message (PER-0019).
        loop {
            // --- Fast path: try the hot tier first (SYS-0020) ---
            let mut popped: Option<(String, Message, DateTime<Utc>)> = None;
            let mut hot_tier_len: usize = 0;

            {
                let mut tiers = self.hot_tier.write().unwrap();
                if let Some(tier) = tiers.get_mut(queue) {
                    // BTreeMap is sorted by key — first entry is highest-priority
                    // available message.
                    if let Some(first_key) = tier.keys().next().cloned() {
                        if let Some(value) = tier.remove(&first_key) {
                            if let Ok(mut msg) = serde_json::from_slice::<Message>(&value) {
                                msg.locked_until = Some(lock_until);
                                msg.locked_by = Some(consumer_id.to_string());
                                msg.retry_count += 1;
                                self.messages
                                    .put(first_key.as_bytes(), serde_json::to_vec(&msg)?)?;
                                popped = Some((first_key, msg, lock_until));
                            }
                        }
                    }
                    hot_tier_len = tier.len();
                }
            }

            // --- Slow path: fall back to RocksDB prefix scan ---
            if popped.is_none() {
                let prefix = format!("{}/", queue);
                let iter = self.messages.prefix_iterator(prefix.as_bytes());

                for item in iter {
                    let (key, value) = item?;
                    let key_str = String::from_utf8_lossy(&key);

                    if !key_str.starts_with(&prefix) {
                        break;
                    }

                    let mut msg: Message = serde_json::from_slice(&value)?;

                    if let Some(locked_until) = msg.locked_until {
                        if locked_until > now {
                            continue;
                        }
                    }

                    msg.locked_until = Some(lock_until);
                    msg.locked_by = Some(consumer_id.to_string());
                    msg.retry_count += 1;

                    self.messages.put(&key, serde_json::to_vec(&msg)?)?;
                    popped = Some((key_str.to_string(), msg, lock_until));
                    break;
                }
            }

            if let Some((key_str, mut msg, lock_until)) = popped {
                // Resolve externalized payload before returning (PER-0016).
                if self.resolve_payload(&mut msg) {
                    // Payload OK — refill hot tier AFTER confirming the message
                    // is valid, so broken messages can't be reloaded (PER-0019).
                    //
                    // Trigger refill when:
                    // 1. The hot tier dropped below the refill threshold, OR
                    // 2. There are cold-storage messages that should be served
                    //    (available count exceeds hot tier size + the one just
                    //    popped).  Without this, continuous publishes can keep
                    //    the hot tier full with NEW messages while OLD messages
                    //    languish in cold storage indefinitely (SYS-0020).
                    let has_cold_messages = {
                        let counters = self.queue_counters.lock().unwrap();
                        counters
                            .get(queue)
                            .is_some_and(|c| c.available > (hot_tier_len + 1) as u64)
                    };
                    if hot_tier_len < self.hot_tier_refill_threshold || has_cold_messages {
                        self.refill_hot_tier(queue);
                    }
                    // Update locked indexes and counters, return to consumer.
                    // Keep the two indexes in lockstep so the
                    // untracked_locks_possible fallback covers both.
                    {
                        let mut idx = self.locked_index.write().unwrap();
                        if idx.len() < self.locked_index_cap {
                            idx.insert(key_str.clone(), lock_until);
                            // DLV-0014: populate secondary index for O(1)
                            // lookup by (queue, message_id).  Drop the
                            // primary index write lock first to keep the
                            // critical section short.
                            drop(idx);
                            self.locked_id_index
                                .write()
                                .unwrap()
                                .entry(queue.to_string())
                                .or_default()
                                .insert(msg.id.clone(), key_str);
                        } else {
                            // Index is at cap — this message's lock is
                            // tracked only in RocksDB.  Set the sticky
                            // flag so the next sweep does a full scan to
                            // catch untracked expirations (DLV-0013) and
                            // so the O(1) lookup paths (DLV-0014) fall
                            // back to the prefix scan.
                            self.untracked_locks_possible
                                .store(true, AtomicOrdering::Release);
                        }
                    }
                    // SYS-0018: available--, locked++
                    {
                        let mut counters = self.queue_counters.lock().unwrap();
                        let entry = counters.entry(queue.to_string()).or_default();
                        entry.available = entry.available.saturating_sub(1);
                        entry.locked += 1;
                    }
                    return Ok(Some(msg));
                }

                // PER-0019: payload unrecoverable — delete the broken message
                // and loop to try the next one.
                tracing::warn!(
                    msg_id = msg.id.as_str(),
                    queue = queue,
                    "Deleting message with unrecoverable payload — \
                     segment file missing or corrupted"
                );
                let _ = self.messages.delete(key_str.as_bytes());
                // Remove payload hash from dedup set.
                {
                    let mut sets = self.payload_sets.write().unwrap();
                    if let Some(set) = sets.get_mut(queue) {
                        set.remove(&Self::get_payload_hash(&msg));
                    }
                }
                // SYS-0018: available-- (message removed, not locked)
                {
                    let mut counters = self.queue_counters.lock().unwrap();
                    if let Some(entry) = counters.get_mut(queue) {
                        entry.available = entry.available.saturating_sub(1);
                    }
                }
                continue; // Try next message
            }

            // No more messages in the queue.  Self-healing: if the hot
            // tier was empty, trigger a refill so it's primed for the next
            // pop attempt.  This covers the case where messages were
            // unlocked between the slow-path scan and now, and prevents
            // the hot tier from staying empty indefinitely.
            if hot_tier_len == 0 {
                self.refill_hot_tier(queue);
            }
            return Ok(None);
        }
    }

    /// Acknowledges successful processing of a message and removes it permanently.
    ///
    /// When a consumer successfully processes a message, they should call ack()
    /// to remove it from the queue. This is a destructive operation - the message
    /// is permanently deleted from storage.
    ///
    /// # Arguments
    ///
    /// * `queue` - Name of the queue containing the message
    /// * `message_id` - Unique ID of the message to acknowledge
    /// * `consumer_id` - ID of the consumer that processed the message
    ///
    /// # Returns
    ///
    /// - `Ok(true)` - Message was found and successfully deleted
    /// - `Ok(false)` - Message not found or not locked by this consumer
    /// - `Err(...)` - Database error
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::storage::Storage;
    ///
    /// # fn process_order(payload: &str) -> Result<(), Box<dyn std::error::Error>> { Ok(()) }
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// // Pop a message
    /// if let Some(msg) = storage.pop("orders", "worker-1", 30)? {
    ///     // Process the message...
    ///     process_order(&msg.payload)?;
    ///
    ///     // Acknowledge successful processing
    ///     let success = storage.ack("orders", &msg.id, "worker-1")?;
    ///     assert!(success);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Security
    ///
    /// Only the consumer that currently holds the lock can acknowledge the message.
    /// This prevents race conditions and ensures message processing integrity.
    ///
    /// Requirements: DLV-0003, API-0004
    pub fn ack(&self, queue: &str, message_id: &str, consumer_id: &str) -> Result<bool> {
        // Fast path (DLV-0014): look up the storage key via the
        // secondary index.  Avoids the O(N) prefix scan that used to
        // dominate renew/ack cost on queues with millions of messages.
        if let Some(key_str) = self.locked_key_for(queue, message_id) {
            return self.finalize_ack_by_key(queue, message_id, consumer_id, &key_str);
        }

        // Index miss.  If no untracked locks can exist (index has been
        // consistent since startup and never hit cap), the message is
        // simply not locked.  Skip the scan.
        if !self.untracked_locks_possible.load(AtomicOrdering::Acquire) {
            return Ok(false);
        }

        // Fallback: legacy O(N) scan for messages locked before this
        // index was populated, or while the index was at cap.
        tracing::debug!(
            queue = queue,
            message_id = message_id,
            "ack: taking O(N) prefix-scan fallback (untracked_locks_possible=true)"
        );
        let prefix = format!("{}/", queue);
        let iter = self.messages.prefix_iterator(prefix.as_bytes());
        for item in iter {
            let (key, value) = item?;
            let msg: Message = serde_json::from_slice(&value)?;
            if msg.id == message_id && msg.locked_by.as_deref() == Some(consumer_id) {
                let key_str = String::from_utf8_lossy(&key).into_owned();
                self.messages.delete(&key)?;
                self.finalize_ack_cleanup(queue, message_id, &msg, &key_str);
                return Ok(true);
            }
        }
        Ok(false)
    }

    /// Completes an ack when the storage key has already been looked
    /// up via `locked_id_index`.  Reads the message to verify it is
    /// still locked by `consumer_id`, deletes it, and performs the
    /// counter and index bookkeeping.
    ///
    /// A stale index entry (message already acked or message never
    /// existed) is cleaned up and treated as `Ok(false)`.
    fn finalize_ack_by_key(
        &self,
        queue: &str,
        message_id: &str,
        consumer_id: &str,
        key_str: &str,
    ) -> Result<bool> {
        let value = match self.messages.get(key_str.as_bytes())? {
            Some(v) => v,
            None => {
                self.remove_from_locked_id_index(queue, message_id);
                return Ok(false);
            }
        };
        let msg: Message = serde_json::from_slice(&value)?;
        if msg.locked_by.as_deref() != Some(consumer_id) {
            return Ok(false);
        }
        self.messages.delete(key_str.as_bytes())?;
        self.finalize_ack_cleanup(queue, message_id, &msg, key_str);
        Ok(true)
    }

    /// Shared ack bookkeeping: drop the lock indexes, the payload hash,
    /// and decrement the locked counter.
    fn finalize_ack_cleanup(&self, queue: &str, message_id: &str, msg: &Message, key_str: &str) {
        self.locked_index.write().unwrap().remove(key_str);
        self.remove_from_locked_id_index(queue, message_id);

        // Remove payload hash from set (PER-0010).
        {
            let mut sets = self.payload_sets.write().unwrap();
            if let Some(set) = sets.get_mut(queue) {
                set.remove(&Self::get_payload_hash(msg));
            }
        }

        // SYS-0018: locked--
        {
            let mut counters = self.queue_counters.lock().unwrap();
            if let Some(entry) = counters.get_mut(queue) {
                entry.locked = entry.locked.saturating_sub(1);
            }
        }
    }

    /// O(1) lookup of the storage key for a (queue, message_id) pair
    /// that is currently locked.  Returns `None` if the message is not
    /// tracked in the secondary index — caller decides whether to fall
    /// back to a full prefix scan based on `untracked_locks_possible`.
    fn locked_key_for(&self, queue: &str, message_id: &str) -> Option<String> {
        let idx = self.locked_id_index.read().unwrap();
        idx.get(queue).and_then(|map| map.get(message_id).cloned())
    }

    /// Removes the (queue, message_id) entry from the secondary index
    /// and cleans up the queue's inner map if it becomes empty.
    fn remove_from_locked_id_index(&self, queue: &str, message_id: &str) {
        let mut idx = self.locked_id_index.write().unwrap();
        if let Some(map) = idx.get_mut(queue) {
            map.remove(message_id);
            if map.is_empty() {
                idx.remove(queue);
            }
        }
    }

    /// Test-only: inject a secondary-index entry that points at a
    /// storage key which may or may not exist in RocksDB.  Lets tests
    /// exercise the stale-entry cleanup branches of the fast paths
    /// without fighting through normal API flows.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub fn __test_insert_locked_id_index_entry(&self, queue: &str, message_id: &str, key: &str) {
        self.locked_id_index
            .write()
            .unwrap()
            .entry(queue.to_string())
            .or_default()
            .insert(message_id.to_string(), key.to_string());
    }

    /// Test-only: read whether the secondary index has an entry for
    /// a given `(queue, message_id)`.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub fn __test_locked_id_index_has(&self, queue: &str, message_id: &str) -> bool {
        self.locked_id_index
            .read()
            .unwrap()
            .get(queue)
            .is_some_and(|m| m.contains_key(message_id))
    }

    /// Negative acknowledgment - unlocks a message for retry or moves to dead letter queue.
    ///
    /// When a consumer cannot process a message (due to errors, invalid data, etc.),
    /// they should call nack() to either:
    /// 1. Unlock the message for immediate retry (if retries remain)
    /// 2. Move the message to dead letter queue (if max retries exceeded)
    ///
    /// # Arguments
    ///
    /// * `queue` - Name of the queue containing the message
    /// * `message_id` - Unique ID of the message to negative acknowledge
    /// * `consumer_id` - ID of the consumer that failed to process the message
    ///
    /// # Returns
    ///
    /// - `Ok(true)` - Message was found and successfully nack'd
    /// - `Ok(false)` - Message not found or not locked by this consumer
    /// - `Err(...)` - Database error
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::storage::Storage;
    ///
    /// # fn process_order(payload: &str) -> Result<(), Box<dyn std::error::Error>> { Err("error".into()) }
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// // Pop a message
    /// if let Some(msg) = storage.pop("orders", "worker-1", 30)? {
    ///     // Try to process the message...
    ///     match process_order(&msg.payload) {
    ///         Ok(_) => {
    ///             storage.ack("orders", &msg.id, "worker-1")?;
    ///         }
    ///         Err(e) => {
    ///             println!("Processing failed: {}", e);
    ///             storage.nack("orders", &msg.id, "worker-1")?;
    ///         }
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Dead Letter Queue Behavior
    ///
    /// When `retry_count >= max_retries`, the message is moved to a dead letter queue
    /// with the key format: `_dlq/{original_queue}/{message_id}`
    ///
    /// # Security
    ///
    /// Only the consumer that currently holds the lock can nack the message.
    ///
    /// Requirements: DLV-0004, DLV-0006, DLV-0007, DLV-0008, DLV-0009, PER-0005, API-0005
    pub fn nack(&self, queue: &str, message_id: &str, consumer_id: &str) -> Result<bool> {
        // Fast path (DLV-0014): look up via the secondary index.
        if let Some(key_str) = self.locked_key_for(queue, message_id) {
            return self.finalize_nack_by_key(queue, message_id, consumer_id, &key_str);
        }

        if !self.untracked_locks_possible.load(AtomicOrdering::Acquire) {
            return Ok(false);
        }

        // Fallback: legacy O(N) prefix scan.
        tracing::debug!(
            queue = queue,
            message_id = message_id,
            "nack: taking O(N) prefix-scan fallback (untracked_locks_possible=true)"
        );
        let prefix = format!("{}/", queue);
        let iter = self.messages.prefix_iterator(prefix.as_bytes());
        for item in iter {
            let (key, value) = item?;
            let msg: Message = serde_json::from_slice(&value)?;
            if msg.id == message_id && msg.locked_by.as_deref() == Some(consumer_id) {
                let key_str = String::from_utf8_lossy(&key).into_owned();
                return self.finalize_nack(queue, message_id, &key_str, msg, &value);
            }
        }
        Ok(false)
    }

    /// Completes a nack when the storage key is already known via
    /// `locked_id_index`.
    fn finalize_nack_by_key(
        &self,
        queue: &str,
        message_id: &str,
        consumer_id: &str,
        key_str: &str,
    ) -> Result<bool> {
        let value = match self.messages.get(key_str.as_bytes())? {
            Some(v) => v,
            None => {
                self.remove_from_locked_id_index(queue, message_id);
                return Ok(false);
            }
        };
        let msg: Message = serde_json::from_slice(&value)?;
        if msg.locked_by.as_deref() != Some(consumer_id) {
            return Ok(false);
        }
        self.finalize_nack(queue, message_id, key_str, msg, &value)
    }

    /// Shared nack body: moves to DLQ or unlocks for retry, then
    /// performs index, counter, and hot-tier bookkeeping.
    fn finalize_nack(
        &self,
        queue: &str,
        message_id: &str,
        key_str: &str,
        mut msg: Message,
        original_value: &[u8],
    ) -> Result<bool> {
        let moved_to_dlq = if msg.retry_count >= msg.max_retries {
            let dlq_key = format!("_dlq/{}/{}", queue, msg.id);
            self.messages.put(dlq_key.as_bytes(), original_value)?;
            self.messages.delete(key_str.as_bytes())?;
            true
        } else {
            msg.locked_until = None;
            msg.locked_by = None;
            self.messages
                .put(key_str.as_bytes(), serde_json::to_vec(&msg)?)?;
            false
        };

        // Lock ordering: locked_index first, then payload_sets
        // (matches ack, batch_ack, batch_nack to prevent ABBA deadlock).
        self.locked_index.write().unwrap().remove(key_str);
        self.remove_from_locked_id_index(queue, message_id);

        if moved_to_dlq {
            let mut sets = self.payload_sets.write().unwrap();
            if let Some(set) = sets.get_mut(queue) {
                set.remove(&Self::get_payload_hash(&msg));
            }
        }

        // SYS-0018: locked--; if retry, available++
        {
            let mut counters = self.queue_counters.lock().unwrap();
            if let Some(entry) = counters.get_mut(queue) {
                entry.locked = entry.locked.saturating_sub(1);
                if !moved_to_dlq {
                    entry.available += 1;
                }
            }
        }

        // If retried (not DLQ'd), the message is available again —
        // insert into hot tier if it qualifies (SYS-0020).
        if !moved_to_dlq {
            self.try_insert_hot_tier(queue, key_str, &msg);
        }

        Ok(true)
    }

    /// Extends the lock on a message that is currently held by `consumer_id`.
    ///
    /// Resets `locked_until` to `now + timeout_secs` and updates the
    /// `locked_index` so the background timeout monitor uses the new expiry.
    /// Returns `Ok(true)` if the lock was renewed, `Ok(false)` if the message
    /// was not found or is not currently locked by `consumer_id`.
    ///
    /// # Security
    ///
    /// Only the consumer that currently holds the lock can renew it.
    ///
    /// Implements: WS-0020
    pub fn renew(
        &self,
        queue: &str,
        message_id: &str,
        consumer_id: &str,
        timeout_secs: u64,
    ) -> Result<bool> {
        let now = chrono::Utc::now();
        let new_expiry = now + chrono::Duration::seconds(timeout_secs as i64);

        // Fast path (DLV-0014): O(1) key lookup via the secondary index.
        // This is the path that made large-queue renews unusable before
        // the index existed: a renew for an already-acked message walked
        // the full prefix, deserializing every entry.
        if let Some(key_str) = self.locked_key_for(queue, message_id) {
            let value = match self.messages.get(key_str.as_bytes())? {
                Some(v) => v,
                None => {
                    // Stale index entry — message was deleted.
                    self.remove_from_locked_id_index(queue, message_id);
                    return Ok(false);
                }
            };
            let mut msg: Message = serde_json::from_slice(&value)?;
            if msg.locked_by.as_deref() != Some(consumer_id) {
                return Ok(false);
            }
            msg.locked_until = Some(new_expiry);
            self.messages
                .put(key_str.as_bytes(), serde_json::to_vec(&msg)?)?;
            self.locked_index
                .write()
                .unwrap()
                .insert(key_str.clone(), new_expiry);
            return Ok(true);
        }

        if !self.untracked_locks_possible.load(AtomicOrdering::Acquire) {
            return Ok(false);
        }

        // Fallback: legacy O(N) prefix scan for untracked locks.
        tracing::debug!(
            queue = queue,
            message_id = message_id,
            "renew: taking O(N) prefix-scan fallback (untracked_locks_possible=true)"
        );
        let prefix = format!("{}/", queue);
        let iter = self.messages.prefix_iterator(prefix.as_bytes());
        for item in iter {
            let (key, value) = item?;
            let mut msg: Message = serde_json::from_slice(&value)?;
            if msg.id == message_id && msg.locked_by.as_deref() == Some(consumer_id) {
                msg.locked_until = Some(new_expiry);
                self.messages.put(&key, serde_json::to_vec(&msg)?)?;

                let key_str = String::from_utf8_lossy(&key).into_owned();
                self.locked_index
                    .write()
                    .unwrap()
                    .insert(key_str, new_expiry);
                return Ok(true);
            }
        }
        Ok(false)
    }

    /// Acknowledges multiple messages in a single operation.
    ///
    /// Builds a single `WriteBatch` for all deletes and acquires `locked_index`
    /// once for all removals — avoiding the per-message fsync and lock overhead
    /// of calling `ack()` in a loop.
    ///
    /// Only messages locked by `consumer_id` are acked; others are reported in
    /// `not_found`.
    ///
    /// **Fast path (DLV-0014):** when no untracked locks are possible (sticky
    /// flag clear), resolves each id to its storage key via `locked_id_index`
    /// in O(batch_size) instead of scanning the whole queue.  IDs that miss
    /// the index provably aren't locked and are returned as `not_found`.
    ///
    /// **Fallback:** when the sticky flag is set, falls back to the original
    /// single prefix scan so cap-overflow locks are still found.
    pub fn batch_ack(
        &self,
        queue: &str,
        consumer_id: &str,
        message_ids: &[String],
    ) -> Result<BatchAckResult> {
        if message_ids.is_empty() {
            return Ok(BatchAckResult::default());
        }

        // Fast path (DLV-0014): O(batch_size) index-driven lookups.
        if !self.untracked_locks_possible.load(AtomicOrdering::Acquire) {
            return self.batch_ack_fast(queue, consumer_id, message_ids);
        }

        // Fallback: legacy single prefix scan for cap-overflow locks.
        tracing::debug!(
            queue = queue,
            batch_size = message_ids.len(),
            "batch_ack: taking O(N) prefix-scan fallback (untracked_locks_possible=true)"
        );
        let ids_to_find: HashSet<&str> = message_ids.iter().map(String::as_str).collect();
        let prefix = format!("{}/", queue);
        let iter = self.messages.prefix_iterator(prefix.as_bytes());

        let mut batch = WriteBatch::default();
        let mut found: Vec<(String, String, [u8; 16])> = Vec::new(); // (id, key, payload_hash)

        for item in iter {
            let (key, value) = item?;
            let key_str = String::from_utf8_lossy(&key);
            if !key_str.starts_with(&prefix) {
                break;
            }

            let msg: Message = serde_json::from_slice(&value)?;
            if ids_to_find.contains(msg.id.as_str())
                && msg.locked_by.as_deref() == Some(consumer_id)
            {
                batch.delete(&key);
                found.push((
                    msg.id.clone(),
                    key_str.to_string(),
                    Self::get_payload_hash(&msg),
                ));
            }
        }

        self.finalize_batch_ack(queue, message_ids, batch, found)
    }

    /// Fast-path batch_ack using the secondary index.  Resolves each id to
    /// its storage key in O(1) and issues point lookups.  Safe to call only
    /// when the `untracked_locks_possible` sticky flag is clear — otherwise
    /// cap-overflow locks would be missed.
    fn batch_ack_fast(
        &self,
        queue: &str,
        consumer_id: &str,
        message_ids: &[String],
    ) -> Result<BatchAckResult> {
        // Snapshot the (id, key) pairs we need from the secondary index,
        // releasing the read lock before doing DB point lookups.
        let targets: Vec<(String, String)> = {
            let idx = self.locked_id_index.read().unwrap();
            match idx.get(queue) {
                Some(map) => message_ids
                    .iter()
                    .filter_map(|id| map.get(id).map(|key| (id.clone(), key.clone())))
                    .collect(),
                None => Vec::new(),
            }
        };

        let mut batch = WriteBatch::default();
        let mut found: Vec<(String, String, [u8; 16])> = Vec::new();
        let mut stale: Vec<String> = Vec::new();

        for (id, key_str) in targets {
            let value = match self.messages.get(key_str.as_bytes())? {
                Some(v) => v,
                None => {
                    // Stale index entry — message already gone.
                    stale.push(id);
                    continue;
                }
            };
            let msg: Message = serde_json::from_slice(&value)?;
            if msg.locked_by.as_deref() == Some(consumer_id) {
                batch.delete(key_str.as_bytes());
                found.push((id, key_str, Self::get_payload_hash(&msg)));
            }
            // Locked by a different consumer → treated as not_found by the caller.
        }

        // Clean up stale secondary-index entries discovered during lookup.
        for id in stale {
            self.remove_from_locked_id_index(queue, &id);
        }

        self.finalize_batch_ack(queue, message_ids, batch, found)
    }

    /// Shared tail of `batch_ack` and `batch_ack_fast`: commit the write
    /// batch, clean up the indexes/payload sets/counters, and assemble the
    /// result.
    fn finalize_batch_ack(
        &self,
        queue: &str,
        message_ids: &[String],
        batch: WriteBatch,
        found: Vec<(String, String, [u8; 16])>,
    ) -> Result<BatchAckResult> {
        if !batch.is_empty() {
            self.messages.write(batch)?;
            let mut locked_index = self.locked_index.write().unwrap();
            for (_, key, _) in &found {
                locked_index.remove(key);
            }
        }

        if !found.is_empty() {
            let mut sets = self.payload_sets.write().unwrap();
            if let Some(set) = sets.get_mut(queue) {
                for (_, _, hash) in &found {
                    set.remove(hash);
                }
            }
        }

        // DLV-0014: drop acked messages from the secondary index.
        if !found.is_empty() {
            let mut id_index = self.locked_id_index.write().unwrap();
            if let Some(map) = id_index.get_mut(queue) {
                for (id, _, _) in &found {
                    map.remove(id);
                }
                if map.is_empty() {
                    id_index.remove(queue);
                }
            }
        }

        // SYS-0018: locked -= acked_count
        if !found.is_empty() {
            let mut counters = self.queue_counters.lock().unwrap();
            if let Some(entry) = counters.get_mut(queue) {
                entry.locked = entry.locked.saturating_sub(found.len() as u64);
            }
        }

        let acked: Vec<String> = found.into_iter().map(|(id, _, _)| id).collect();
        let acked_set: HashSet<&str> = acked.iter().map(String::as_str).collect();
        let not_found = message_ids
            .iter()
            .filter(|id| !acked_set.contains(id.as_str()))
            .cloned()
            .collect();

        Ok(BatchAckResult { acked, not_found })
    }

    /// Negatively acknowledges multiple messages in a single operation.
    ///
    /// Performs one prefix scan to find target messages, then processes each —
    /// either unlocking for retry or moving to DLQ — using a single `WriteBatch`
    /// and a single `locked_index` write-lock acquisition.
    ///
    /// Under DLV-0010 the old "drop-on-nack-due-to-duplicate" code path is removed:
    /// DLQ'd payloads are removed from `payload_sets`; retried messages keep their
    /// payload in the set (they remain in the queue).
    pub fn batch_nack(
        &self,
        queue: &str,
        consumer_id: &str,
        message_ids: &[String],
    ) -> Result<BatchNackResult> {
        if message_ids.is_empty() {
            return Ok(BatchNackResult::default());
        }

        // Fast path (DLV-0014): O(batch_size) index-driven lookups.
        if !self.untracked_locks_possible.load(AtomicOrdering::Acquire) {
            return self.batch_nack_fast(queue, consumer_id, message_ids);
        }

        // Fallback: legacy single prefix scan for cap-overflow locks.
        tracing::debug!(
            queue = queue,
            batch_size = message_ids.len(),
            "batch_nack: taking O(N) prefix-scan fallback (untracked_locks_possible=true)"
        );
        let ids_to_find: HashSet<&str> = message_ids.iter().map(String::as_str).collect();
        let prefix = format!("{}/", queue);

        // Placeholder: reuse the fast-path struct for uniform processing below.
        let mut targets: Vec<BatchNackTarget> = Vec::new();

        let iter = self.messages.prefix_iterator(prefix.as_bytes());
        for item in iter {
            let (key, value) = item?;
            let key_str = String::from_utf8_lossy(&key);
            if !key_str.starts_with(&prefix) {
                break;
            }

            let msg: Message = serde_json::from_slice(&value)?;
            if ids_to_find.contains(msg.id.as_str())
                && msg.locked_by.as_deref() == Some(consumer_id)
            {
                targets.push(BatchNackTarget {
                    key: key_str.to_string(),
                    msg,
                });
            }
        }

        self.finalize_batch_nack(queue, message_ids, targets)
    }

    /// Fast-path batch_nack using the secondary index.  Resolves each id to
    /// its storage key in O(1) and issues point lookups.  Safe to call only
    /// when the `untracked_locks_possible` sticky flag is clear.
    fn batch_nack_fast(
        &self,
        queue: &str,
        consumer_id: &str,
        message_ids: &[String],
    ) -> Result<BatchNackResult> {
        let snapshot: Vec<(String, String)> = {
            let idx = self.locked_id_index.read().unwrap();
            match idx.get(queue) {
                Some(map) => message_ids
                    .iter()
                    .filter_map(|id| map.get(id).map(|key| (id.clone(), key.clone())))
                    .collect(),
                None => Vec::new(),
            }
        };

        let mut targets: Vec<BatchNackTarget> = Vec::new();
        let mut stale: Vec<String> = Vec::new();

        for (id, key_str) in snapshot {
            let value = match self.messages.get(key_str.as_bytes())? {
                Some(v) => v,
                None => {
                    stale.push(id);
                    continue;
                }
            };
            let msg: Message = serde_json::from_slice(&value)?;
            if msg.locked_by.as_deref() == Some(consumer_id) {
                targets.push(BatchNackTarget { key: key_str, msg });
            }
        }

        for id in stale {
            self.remove_from_locked_id_index(queue, &id);
        }

        self.finalize_batch_nack(queue, message_ids, targets)
    }

    /// Shared tail of `batch_nack` and `batch_nack_fast`: for each target,
    /// DLQ it or unlock for retry, commit the write batch, and do the
    /// index / payload_sets / counter / hot-tier bookkeeping.
    fn finalize_batch_nack(
        &self,
        queue: &str,
        message_ids: &[String],
        targets: Vec<BatchNackTarget>,
    ) -> Result<BatchNackResult> {
        let mut batch = WriteBatch::default();
        let mut result = BatchNackResult::default();
        let mut index_keys: Vec<String> = Vec::new();
        let mut dlq_payloads: Vec<[u8; 16]> = Vec::new();

        // Track messages unlocked for retry so we can insert them into the
        // hot tier after the batch write succeeds (SYS-0020).
        let mut retried_for_hot_tier: Vec<(String, Message)> = Vec::new();

        for BatchNackTarget { key, mut msg } in targets {
            if msg.retry_count >= msg.max_retries {
                // Move to dead-letter queue; payload will be removed from set.
                let dlq_key = format!("_dlq/{}/{}", queue, msg.id);
                batch.put(dlq_key.as_bytes(), serde_json::to_vec(&msg)?);
                batch.delete(key.as_bytes());
                result.dead_lettered.push(msg.id.clone());
                dlq_payloads.push(Self::get_payload_hash(&msg));
            } else {
                // Unlock for retry — payload stays in set (message stays in queue).
                msg.locked_until = None;
                msg.locked_by = None;
                batch.put(key.as_bytes(), serde_json::to_vec(&msg)?);
                result.unlocked.push(msg.id.clone());
                retried_for_hot_tier.push((key.clone(), msg));
            }

            index_keys.push(key);
        }

        // Record not_found: IDs we were asked to nack but didn't find
        let processed: HashSet<&str> = result
            .unlocked
            .iter()
            .chain(result.dead_lettered.iter())
            .chain(result.dropped.iter())
            .map(String::as_str)
            .collect();
        result.not_found = message_ids
            .iter()
            .filter(|id| !processed.contains(id.as_str()))
            .cloned()
            .collect();

        if !batch.is_empty() {
            self.messages.write(batch)?;
        }

        // Single lock acquisition for all index removals
        if !index_keys.is_empty() {
            let mut locked_index = self.locked_index.write().unwrap();
            for key in index_keys {
                locked_index.remove(&key);
            }
        }

        // DLV-0014: drop every processed (retried or dead-lettered)
        // message from the secondary index.  They are no longer locked
        // by this consumer, so the fast path must stop returning their
        // old storage keys.
        let processed_ids: Vec<&String> = result
            .unlocked
            .iter()
            .chain(result.dead_lettered.iter())
            .collect();
        if !processed_ids.is_empty() {
            let mut id_index = self.locked_id_index.write().unwrap();
            if let Some(map) = id_index.get_mut(queue) {
                for id in &processed_ids {
                    map.remove(id.as_str());
                }
                if map.is_empty() {
                    id_index.remove(queue);
                }
            }
        }

        // Remove DLQ'd payloads from set (PER-0010).
        if !dlq_payloads.is_empty() {
            let mut sets = self.payload_sets.write().unwrap();
            if let Some(set) = sets.get_mut(queue) {
                for payload in dlq_payloads {
                    set.remove(&payload);
                }
            }
        }

        // SYS-0018: locked -= (unlocked + dead_lettered); available += unlocked
        {
            let total_processed = result.unlocked.len() + result.dead_lettered.len();
            if total_processed > 0 {
                let mut counters = self.queue_counters.lock().unwrap();
                if let Some(entry) = counters.get_mut(queue) {
                    entry.locked = entry.locked.saturating_sub(total_processed as u64);
                    entry.available += result.unlocked.len() as u64;
                }
            }
        }

        // Insert retried messages into hot tier (SYS-0020).
        for (key_str, msg) in retried_for_hot_tier {
            self.try_insert_hot_tier(queue, &key_str, &msg);
        }

        Ok(result)
    }

    /// Unlocks messages that have exceeded their timeout duration.
    ///
    /// This method is called by the timeout monitor to scan for and unlock
    /// messages whose lock has expired. It efficiently processes the in-memory
    /// locked index to identify expired messages, then unlocks them in storage.
    ///
    /// # Returns
    ///
    /// Returns `Ok(usize)` with the number of messages unlocked, or an error
    /// if there were issues accessing storage.
    ///
    /// # Implementation Details
    ///
    /// The method works in two phases:
    /// 1. **Collection Phase**: Read the locked index to collect expired keys
    /// 2. **Unlock Phase**: For each expired key, unlock the message in storage
    ///
    /// This approach minimizes the time spent holding write locks on the index.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::storage::Storage;
    /// use std::sync::Arc;
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Arc::new(Storage::new("/tmp/test")?);
    ///
    /// // Called by timeout monitor
    /// let unlocked_count = storage.unlock_expired_messages()?;
    /// println!("Unlocked {} expired messages", unlocked_count);
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Error Handling
    ///
    /// If unlocking individual messages fails, the method continues processing
    /// other expired messages. Only storage-level errors cause the method to return
    /// an error, ensuring maximum availability of the timeout monitoring system.
    ///
    /// Requirements: DLV-0005, DLV-0013, PER-0005, SCH-0004
    pub fn unlock_expired_messages(&self) -> Result<usize> {
        let now = Utc::now();
        let mut expired_keys = Vec::new();
        let index_at_cap;
        // Capture the sticky flag once; clear it only after a successful
        // full scan completes below.  If insertions happen during the
        // scan, they will re-set the flag and the next sweep will scan
        // again.
        let needs_full_scan = self.untracked_locks_possible.load(AtomicOrdering::Acquire);

        // Phase 1: Collect expired keys (using read lock to minimize contention)
        {
            let locked_index = self.locked_index.read().unwrap();
            index_at_cap = locked_index.len() >= self.locked_index_cap;
            for (key, lock_until) in locked_index.iter() {
                if *lock_until <= now {
                    expired_keys.push(key.clone());
                }
            }
        }

        // Fallback: do a full scan if
        //   (a) the index is currently at cap (in-flight overflow), or
        //   (b) the sticky `untracked_locks_possible` flag is set,
        //       meaning an insert was skipped at some point since the
        //       last successful full scan, even if the index has since
        //       dropped below cap.  Without (b), locks skipped under
        //       pressure would remain stranded forever (DLV-0013).
        let do_full_scan = index_at_cap || needs_full_scan;
        if do_full_scan {
            let tracked: HashSet<String> = expired_keys.iter().cloned().collect();
            // Count locked-but-not-yet-expired messages that are NOT in
            // `locked_index`.  If any exist after this scan, we must
            // keep the `untracked_locks_possible` flag set so the next
            // sweep runs a full scan again and catches them once they
            // expire.  Without this, locks whose `locked_index` insert
            // was skipped (because the index was at cap at pop-time)
            // would remain stranded forever — they're not in the index,
            // and the flag that triggers the full scan has been cleared.
            let mut still_locked_untracked: u64 = 0;
            let locked_index_snapshot: HashSet<String> = {
                let li = self.locked_index.read().unwrap();
                li.keys().cloned().collect()
            };

            let iter = self.messages.iterator(rocksdb::IteratorMode::Start);
            for item in iter {
                let (key, value) = item?;
                let key_str = String::from_utf8_lossy(&key);
                if key_str.starts_with("_dlq/") || key_str.starts_with("_queue_config/") {
                    continue;
                }
                if tracked.contains(key_str.as_ref()) {
                    continue;
                }
                if let Ok(msg) = serde_json::from_slice::<Message>(&value) {
                    if msg.locked_until.is_some_and(|lu| lu <= now) {
                        expired_keys.push(key_str.into_owned());
                    } else if msg.locked_until.is_some_and(|lu| lu > now)
                        && !locked_index_snapshot.contains(key_str.as_ref())
                    {
                        // This message is locked but not yet expired,
                        // and it's NOT tracked in the locked_index.
                        // We must keep scanning on future sweeps so it
                        // gets caught when it expires.
                        still_locked_untracked += 1;
                    }
                }
            }
            // Only clear the flag if zero untracked locked messages
            // remain.  If any exist, the next sweep must do a full
            // scan again to catch them when they expire.
            if needs_full_scan {
                if still_locked_untracked == 0 {
                    self.untracked_locks_possible
                        .store(false, AtomicOrdering::Release);
                } else {
                    tracing::debug!(
                        "Full scan found {} locked-but-not-yet-expired messages \
                         not tracked in locked_index — keeping full-scan flag set",
                        still_locked_untracked,
                    );
                }
            }
        }

        let mut unlocked_count = 0;
        // SYS-0018: track per-queue unlock counts for counter updates.
        let mut unlock_by_queue: HashMap<String, u64> = HashMap::new();

        // Phase 2: Unlock each expired message
        for key in &expired_keys {
            match self.unlock_message_by_key(key) {
                Ok(true) => {
                    unlocked_count += 1;
                    // Extract queue name from key (format: queue/priority/ts/uuid)
                    if let Some(queue) = key.split('/').next() {
                        *unlock_by_queue.entry(queue.to_string()).or_default() += 1;
                    }
                }
                Ok(false) => {
                    // Message not found or already unlocked - remove from index anyway
                    tracing::debug!("Message not found for expired key: {}", key);
                }
                Err(e) => {
                    tracing::error!("Failed to unlock message {}: {}", key, e);
                    // Continue processing other messages
                }
            }
        }

        // Phase 3: Remove expired entries from the locked index
        if !expired_keys.is_empty() {
            let mut locked_index = self.locked_index.write().unwrap();
            for key in expired_keys {
                locked_index.remove(&key);
            }
        }

        // SYS-0018: locked -= n, available += n for each affected queue.
        // Collect queue names before updating counters so we can refill
        // their hot tiers afterwards.
        let affected_queues: Vec<String> = unlock_by_queue.keys().cloned().collect();
        if !unlock_by_queue.is_empty() {
            let mut counters = self.queue_counters.lock().unwrap();
            for (queue, count) in unlock_by_queue {
                if let Some(entry) = counters.get_mut(&queue) {
                    entry.locked = entry.locked.saturating_sub(count);
                    entry.available += count;
                }
            }
        }

        // Batch refill: rebuild hot tiers for all affected queues AFTER
        // the full unlock batch completes.  This produces a consistent
        // snapshot and eliminates the race where per-message
        // try_insert_hot_tier interleaves with a concurrent pop's
        // refill_hot_tier (which replaces the entire tier).
        for queue in &affected_queues {
            self.refill_hot_tier(queue);
        }

        Ok(unlocked_count)
    }

    /// Gets comprehensive statistics for all queues in the system.
    ///
    /// Returns statistics for each queue including available, locked, and total
    /// message counts. This provides operational visibility into queue health
    /// and performance.
    ///
    /// # Returns
    ///
    /// Returns `Ok(Vec<QueueStats>)` with statistics for all queues, or an error
    /// if there were issues scanning the storage.
    ///
    /// # Implementation Details
    ///
    /// The method scans all messages in storage to compute statistics:
    /// 1. Groups messages by queue name
    /// 2. Counts available vs locked messages using current timestamps
    /// 3. Handles dead letter queues separately (prefixed with `_dlq/`)
    ///
    /// # Performance
    ///
    /// This operation scans the entire database and should be used judiciously
    /// in high-throughput environments. Consider caching results or calling
    /// less frequently for performance-sensitive applications.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::storage::Storage;
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// let stats = storage.get_all_queue_stats()?;
    /// for stat in stats {
    ///     println!("Queue {}: {} available, {} locked, {} total",
    ///              stat.name, stat.available, stat.locked, stat.total);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    // Implements: SYS-0018 — reads from in-memory cache, no DB scan.
    pub fn get_all_queue_stats(&self) -> Result<Vec<crate::message::QueueStats>> {
        let counters = self.queue_counters.lock().unwrap();

        let mut stats: Vec<crate::message::QueueStats> = counters
            .iter()
            .map(|(queue_name, counts)| {
                let config = self.get_queue_config(queue_name);
                crate::message::QueueStats {
                    name: queue_name.clone(),
                    available: counts.available as usize,
                    locked: counts.locked as usize,
                    total: (counts.available + counts.locked) as usize,
                    config,
                }
            })
            .collect();

        // Sort by queue name for consistent output
        stats.sort_by(|a, b| a.name.cmp(&b.name));

        Ok(stats)
    }

    /// Gets statistics for a specific queue.
    ///
    /// Returns detailed statistics for a single queue, including available,
    /// locked, and total message counts.
    ///
    /// # Arguments
    ///
    /// * `queue_name` - Name of the queue to get statistics for
    ///
    /// # Returns
    ///
    /// Returns `Ok(QueueStats)` with statistics for the specified queue, or an error
    /// if there were issues scanning the storage.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::storage::Storage;
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// let stats = storage.get_queue_stats("orders")?;
    /// println!("Orders queue: {} available, {} locked",
    ///          stats.available, stats.locked);
    /// # Ok(())
    /// # }
    /// ```
    // Implements: SYS-0018 — reads from in-memory cache, no DB scan.
    pub fn get_queue_stats(&self, queue_name: &str) -> Result<crate::message::QueueStats> {
        let counters = self.queue_counters.lock().unwrap();
        let counts = counters.get(queue_name);

        let (available, locked) = match counts {
            Some(c) => (c.available as usize, c.locked as usize),
            None => (0, 0),
        };

        Ok(crate::message::QueueStats {
            name: queue_name.to_string(),
            available,
            locked,
            total: available + locked,
            config: self.get_queue_config(queue_name),
        })
    }

    /// Gets a list of all queue names that contain messages.
    ///
    /// Scans the storage to identify all unique queue names that have
    /// at least one message (either available or locked).
    ///
    /// # Returns
    ///
    /// Returns `Ok(Vec<String>)` with all queue names, or an error if there
    /// were issues scanning the storage.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use qrusty::storage::Storage;
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/test")?;
    ///
    /// let queues = storage.list_queues()?;
    /// println!("Active queues: {:?}", queues);
    /// # Ok(())
    /// # }
    /// ```
    // Implements: SYS-0018 — reads from in-memory cache, no DB scan.
    pub fn list_queues(&self) -> Result<Vec<String>> {
        let counters = self.queue_counters.lock().unwrap();
        let mut queues: Vec<String> = counters.keys().cloned().collect();
        queues.sort();
        Ok(queues)
    }

    /// Unlocks a specific message by its storage key.
    ///
    /// This helper method unlocks a message identified by its storage key,
    /// setting `locked_until` and `locked_by` to `None` to make it available
    /// for consumption again.
    ///
    /// # Arguments
    ///
    /// * `key` - The storage key of the message to unlock
    ///
    /// # Returns
    ///
    /// - `Ok(true)` - Message was found and successfully unlocked
    /// - `Ok(false)` - Message not found (may have been ack'd or nack'd)
    /// - `Err(...)` - Storage or serialization error
    ///
    /// # Implementation Notes
    ///
    /// This method does not remove entries from the locked_index - that's
    /// handled by the calling method to allow for batch operations.
    /// Force-unlocks ALL locked messages on `queue`, regardless of
    /// whether their locks have expired.  Returns the count unlocked.
    ///
    /// Implements: API-0014
    ///
    /// This is the operational escape hatch for queues whose locks
    /// became stranded after a crash — the normal lock-expiry
    /// scanner may not find them if `untracked_locks_possible` was
    /// cleared before the locks expired.
    pub fn force_unlock_queue_sync(&self, queue: &str) -> Result<usize> {
        let prefix = format!("{}/", queue);
        let iter = self.messages.prefix_iterator(prefix.as_bytes());
        let mut unlocked = 0usize;
        let mut keys_to_remove = Vec::new();

        for item in iter {
            let (key, value) = item?;
            let key_str = String::from_utf8_lossy(&key);
            if !key_str.starts_with(&prefix) {
                break;
            }
            if let Ok(msg) = serde_json::from_slice::<Message>(&value) {
                if msg.locked_until.is_some() {
                    // Reuse the existing unlock helper which clears
                    // locked_until/locked_by, writes to RocksDB, and
                    // inserts into the hot tier.
                    match self.unlock_message_by_key(&key_str) {
                        Ok(true) => {
                            unlocked += 1;
                            keys_to_remove.push(key_str.into_owned());
                        }
                        Ok(false) => {}
                        Err(e) => {
                            tracing::warn!(
                                "force_unlock_queue: failed to unlock {}: {}",
                                key_str,
                                e
                            );
                        }
                    }
                }
            }
        }

        // Remove unlocked entries from locked_index.
        if !keys_to_remove.is_empty() {
            let mut locked_index = self.locked_index.write().unwrap();
            for key in &keys_to_remove {
                locked_index.remove(key);
            }
        }

        // DLV-0014: defensively drop the queue's entire sub-map from
        // the secondary index.  `unlock_message_by_key` already removes
        // each message individually; this guarantees consistency if any
        // entry was missed (e.g., a message whose RocksDB record was
        // corrupt and failed to deserialize).
        {
            let mut idx = self.locked_id_index.write().unwrap();
            idx.remove(queue);
        }

        // Update counters: locked -= N, available += N.
        if unlocked > 0 {
            let mut counters = self.queue_counters.lock().unwrap();
            if let Some(entry) = counters.get_mut(queue) {
                entry.locked = entry.locked.saturating_sub(unlocked as u64);
                entry.available += unlocked as u64;
            }
            // Refill the hot tier now that messages are available.
            self.refill_hot_tier(queue);

            tracing::info!(
                "force_unlock_queue: unlocked {} messages on queue '{}'",
                unlocked,
                queue
            );
        }

        Ok(unlocked)
    }

    fn unlock_message_by_key(&self, key: &str) -> Result<bool> {
        // Try to find and unlock the message
        match self.messages.get(key.as_bytes())? {
            Some(value) => {
                let mut msg: Message = serde_json::from_slice(&value)?;

                // Check if the message is actually locked
                if msg.locked_until.is_some() {
                    // Unlock the message — payload stays in payload_sets (message
                    // stays in queue).  DLV-0010 guarantees no duplicate unlocked
                    // message can exist, so the old drop-on-duplicate path is removed.
                    msg.locked_until = None;
                    msg.locked_by = None;

                    // Update in storage
                    self.messages
                        .put(key.as_bytes(), serde_json::to_vec(&msg)?)?;

                    // DLV-0014: drop from the secondary index so future
                    // ack/nack/renew calls on this (queue, id) miss fast
                    // instead of taking the stale fast path.
                    self.remove_from_locked_id_index(&msg.queue, &msg.id);

                    // Message is available again — insert into hot tier (SYS-0020).
                    self.try_insert_hot_tier(&msg.queue, key, &msg);

                    #[cfg(not(coverage))]
                    tracing::debug!(
                        "Unlocked expired message: {} from queue: {}",
                        msg.id,
                        msg.queue
                    );
                    Ok(true)
                } else {
                    // Message is already unlocked
                    Ok(false)
                }
            }
            None => {
                // Message doesn't exist (probably ack'd or moved to DLQ)
                Ok(false)
            }
        }
    }

    /// Deletes a queue and all its messages.
    ///
    /// Implements: PER-0013
    ///
    /// This is a destructive operation that removes all messages from the specified queue
    /// and cleans up associated metadata. Messages in any state (pending, locked, processed)
    /// and dead-letter queue entries will be permanently deleted.
    ///
    /// # Arguments
    ///
    /// * `queue_name` - Name of the queue to delete
    ///
    /// # Returns
    ///
    /// * `Ok(deleted_count)` - Number of messages that were deleted
    /// * `Err(error)` - If deletion fails
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use qrusty::storage::Storage;
    /// # use qrusty::message::QueueConfig;
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/qrusty_db")?;
    ///
    /// // Delete a queue and all its messages
    /// let deleted_count = storage.delete_queue("old_queue")?;
    /// println!("Deleted {} messages from queue", deleted_count);
    /// # Ok(())
    /// # }
    /// ```
    pub fn delete_queue(&self, queue_name: &str) -> Result<usize> {
        let mut deleted_count = 0;
        let mut keys_to_delete = Vec::new();

        // Find all messages in the queue
        let iter = self.messages.iterator(rocksdb::IteratorMode::Start);
        for item in iter {
            let (key, _value) = item?;
            let key_str = String::from_utf8_lossy(&key);

            // Parse key to check if it belongs to this queue
            // Key format: "queue_name/priority/timestamp/id"
            let queue_from_key = key_str.split('/').next().unwrap_or("");
            if queue_from_key == queue_name {
                keys_to_delete.push(key.to_vec());
            }
        }

        // Delete all messages for this queue
        for key in &keys_to_delete {
            self.messages.delete(key)?;
            deleted_count += 1;

            // Remove from locked index if present
            let key_str = String::from_utf8_lossy(key);
            self.locked_index.write().unwrap().remove(key_str.as_ref());
        }

        // Delete dead letter queue entries for this queue
        let dlq_prefix = format!("_dlq/{}/", queue_name);
        let dlq_iter = self.messages.prefix_iterator(dlq_prefix.as_bytes());
        for item in dlq_iter {
            let (key, _value) = item?;
            let key_str = String::from_utf8_lossy(&key);
            if !key_str.starts_with(&dlq_prefix) {
                break;
            }
            self.messages.delete(&key)?;
        }

        // Remove queue configuration from memory
        self.queue_configs.write().unwrap().remove(queue_name);

        // Remove payload set (PER-0010).
        self.payload_sets.write().unwrap().remove(queue_name);

        // Remove hot tier (SYS-0020).
        self.hot_tier.write().unwrap().remove(queue_name);

        // DLV-0014: drop the queue's sub-map from the secondary index.
        self.locked_id_index.write().unwrap().remove(queue_name);

        // SYS-0018: remove counter entry
        self.queue_counters.lock().unwrap().remove(queue_name);

        // Remove queue configuration from persistent storage
        let config_key = format!("_queue_config/{}", queue_name);
        let config_delete_result = self.messages.delete(config_key.as_bytes());

        #[cfg(not(any(test, coverage)))]
        if let Err(e) = config_delete_result {
            tracing::warn!("Failed to delete queue config for '{}': {}", queue_name, e);
        }

        #[cfg(any(test, coverage))]
        let _ = config_delete_result;

        #[cfg(not(coverage))]
        tracing::info!(
            "Deleted queue '{}' with {} messages",
            queue_name,
            deleted_count
        );

        Ok(deleted_count)
    }

    /// Purges all messages from a queue without deleting the queue itself.
    ///
    /// This operation removes all messages from the specified queue but preserves
    /// the queue configuration. The queue will continue to exist and can accept
    /// new messages.
    ///
    /// # Arguments
    ///
    /// * `queue_name` - Name of the queue to purge
    ///
    /// # Returns
    ///
    /// * `Ok(purged_count)` - Number of messages that were purged
    /// * `Err(error)` - If purging fails
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use qrusty::storage::Storage;
    /// # use qrusty::message::QueueConfig;
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let storage = Storage::new("/tmp/qrusty_db")?;
    ///
    /// // Purge all messages from a queue
    /// let purged_count = storage.purge_queue("busy_queue")?;
    /// println!("Purged {} messages from queue", purged_count);
    /// # Ok(())
    /// # }
    /// ```
    pub fn purge_queue(&self, queue_name: &str) -> Result<usize> {
        let mut purged_count = 0;
        let mut keys_to_delete = Vec::new();

        // Find all messages in the queue
        let iter = self.messages.iterator(rocksdb::IteratorMode::Start);
        for item in iter {
            let (key, _value) = item?;
            let key_str = String::from_utf8_lossy(&key);

            // Parse key to check if it belongs to this queue
            // Key format: "queue_name/priority/timestamp/id"
            let queue_from_key = key_str.split('/').next().unwrap_or("");
            if queue_from_key == queue_name {
                keys_to_delete.push(key.to_vec());
            }
        }

        // Delete all messages for this queue
        for key in keys_to_delete {
            self.messages.delete(&key)?;
            purged_count += 1;

            // Remove from locked index if present
            let key_str = String::from_utf8_lossy(&key);
            self.locked_index.write().unwrap().remove(key_str.as_ref());
        }

        // Clear the payload set — queue still exists but is now empty (PER-0010).
        let mut sets = self.payload_sets.write().unwrap();
        if let Some(set) = sets.get_mut(queue_name) {
            set.clear();
        }

        // Clear hot tier (SYS-0020).
        if let Some(tier) = self.hot_tier.write().unwrap().get_mut(queue_name) {
            tier.clear();
        }

        // DLV-0014: drop the queue's sub-map from the secondary index.
        self.locked_id_index.write().unwrap().remove(queue_name);

        // SYS-0018: zero out counts (queue still exists)
        {
            let mut counters = self.queue_counters.lock().unwrap();
            if let Some(entry) = counters.get_mut(queue_name) {
                entry.available = 0;
                entry.locked = 0;
            }
        }

        #[cfg(not(coverage))]
        tracing::info!(
            "Purged {} messages from queue '{}'",
            purged_count,
            queue_name
        );

        Ok(purged_count)
    }
}

#[cfg(test)]
mod unit_tests {
    use super::*;
    use crate::message::{PriorityOrdering, QueueConfig};
    use chrono::Utc;
    use std::sync::Mutex;
    use tempfile::TempDir;

    static ENV_LOCK: Mutex<()> = Mutex::new(());

    fn init_tracing_for_tests() {
        let _ = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::DEBUG)
            .with_test_writer()
            .try_init();
    }

    fn make_message(queue: &str, id: &str, payload: &str) -> Message {
        Message {
            id: id.to_string(),
            queue: queue.to_string(),
            priority: Priority::Numeric(0),
            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_has_duplicate_payload_returns_true_for_unlocked_duplicate() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        let m1 = make_message("dupq", "id1", "p");
        let m2 = make_message("dupq", "id2", "p");
        storage.push(m1).unwrap();
        storage.push(m2).unwrap();

        let has = storage
            .has_duplicate_payload("dupq", "p", Some("nonexistent"))
            .unwrap();
        assert!(has);
    }

    #[test]
    fn test_unlock_expired_messages_handles_missing_and_corrupt_entries() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        let expired = Utc::now() - chrono::Duration::seconds(10);

        // Case 1: locked_index contains a key that doesn't exist in DB.
        {
            let mut li = storage.locked_index.write().unwrap();
            li.insert("missing_queue/0/0/missing".to_string(), expired);
        }

        // Case 2: key exists but value is corrupt JSON -> unlock_message_by_key errors.
        let corrupt_key = "corrupt_queue/0/0/corrupt";
        storage
            .messages
            .put(corrupt_key.as_bytes(), b"not_json")
            .unwrap();
        {
            let mut li = storage.locked_index.write().unwrap();
            li.insert(corrupt_key.to_string(), expired);
        }

        let unlocked = storage.unlock_expired_messages().unwrap();
        assert_eq!(unlocked, 0);

        // Both entries should be removed from the index in phase 3.
        let li = storage.locked_index.read().unwrap();
        assert!(li.is_empty());
    }

    #[test]
    fn test_unlock_expired_messages_unlocks_no_dup_queue_message() {
        // Under DLV-0010 the old "drop-on-unlock-due-to-duplicate" code path is
        // removed.  An expired-lock message is simply unlocked; its payload stays
        // in payload_sets because the message remains in the queue.
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

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

        // Push and immediately lock a message.
        let msg = make_message("nodup", "exp", "payload-x");
        storage.push(msg).unwrap();
        storage.pop("nodup", "consumer", 3600).unwrap();

        // Back-date the lock so it looks expired.
        let locked_key = {
            let li = storage.locked_index.read().unwrap();
            li.keys().next().unwrap().clone()
        };
        {
            let mut li = storage.locked_index.write().unwrap();
            li.insert(
                locked_key.clone(),
                Utc::now() - chrono::Duration::seconds(1),
            );
        }

        let unlocked = storage.unlock_expired_messages().unwrap();
        assert_eq!(unlocked, 1);

        // Message must still be in the DB (just unlocked, not dropped).
        assert!(storage
            .messages
            .get(locked_key.as_bytes())
            .unwrap()
            .is_some());

        // Payload hash must still be in the payload set.
        let sets = storage.payload_sets.read().unwrap();
        let set = sets.get("nodup").expect("payload set must exist");
        assert!(
            set.contains(&hash_payload("payload-x")),
            "payload hash must remain in set"
        );
    }

    #[test]
    fn test_storage_new_loads_queue_config_from_db() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let data_path = temp_dir.path().to_str().unwrap();

        // Pre-populate a queue config directly in RocksDB.
        let mut opts = rocksdb::Options::default();
        opts.create_if_missing(true);
        let db = rocksdb::DB::open(&opts, format!("{}/messages", data_path)).unwrap();
        let cfg = QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates: true,
            ..Default::default()
        };
        db.put(
            b"_queue_config/preloaded",
            serde_json::to_vec(&cfg).unwrap(),
        )
        .unwrap();
        drop(db);

        let storage = Storage::new(data_path).unwrap();
        let loaded = storage.get_queue_config("preloaded");
        assert_eq!(loaded.ordering, PriorityOrdering::MaxFirst);
        assert!(loaded.allow_duplicates);
    }

    #[test]
    fn test_storage_new_skips_invalid_queue_config_json() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let data_path = temp_dir.path().to_str().unwrap();

        // Pre-populate an invalid queue config entry.
        let mut opts = rocksdb::Options::default();
        opts.create_if_missing(true);
        let db = rocksdb::DB::open(&opts, format!("{}/messages", data_path)).unwrap();
        db.put(b"_queue_config/", b"{}").unwrap();
        db.put(b"_queue_config/bad", b"not_json").unwrap();
        drop(db);

        let storage = Storage::new(data_path).unwrap();
        let loaded = storage.get_queue_config("bad");
        // Invalid JSON should be ignored, so we get defaults.
        assert_eq!(loaded.ordering, PriorityOrdering::MaxFirst);
        assert!(loaded.allow_duplicates);
    }

    #[test]
    fn test_storage_new_warns_on_forced_queue_config_load_error() {
        init_tracing_for_tests();
        let _guard = ENV_LOCK.lock().unwrap();

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

        std::env::set_var("QRUSTY_TEST_FORCE_LOAD_QUEUE_CONFIGS_ERROR", "1");
        let res = Storage::new(data_path);
        std::env::remove_var("QRUSTY_TEST_FORCE_LOAD_QUEUE_CONFIGS_ERROR");

        assert!(res.is_ok());
    }

    #[test]
    fn test_has_duplicate_payload_mismatch_and_break_on_non_prefix_key() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        let past = Utc::now() - chrono::Duration::seconds(10);

        // Put an expired-locked message in queue "a" with a non-matching payload.
        let msg_a = Message {
            locked_until: Some(past),
            locked_by: Some("someone".to_string()),
            ..make_message("a", "id_a", "hay")
        };
        storage
            .messages
            .put(b"a/0/0/id_a", serde_json::to_vec(&msg_a).unwrap())
            .unwrap();

        // Put a message in another queue so the prefix iterator can run past the prefix.
        let msg_b = make_message("b", "id_b", "zzz");
        storage
            .messages
            .put(b"b/0/0/id_b", serde_json::to_vec(&msg_b).unwrap())
            .unwrap();

        let has = storage.has_duplicate_payload("a", "needle", None).unwrap();
        assert!(!has);
    }

    #[test]
    fn test_pop_treats_expired_lock_as_available() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        let past = Utc::now() - chrono::Duration::seconds(10);
        let expired_locked = Message {
            locked_until: Some(past),
            locked_by: Some("old".to_string()),
            ..make_message("expq", "id", "p")
        };

        let key = b"expq/0/0/id";
        storage
            .messages
            .put(key, serde_json::to_vec(&expired_locked).unwrap())
            .unwrap();

        let popped = storage.pop("expq", "new_consumer", 30).unwrap().unwrap();
        assert_eq!(popped.id, "id");
        assert_eq!(popped.locked_by.as_deref(), Some("new_consumer"));
        assert!(popped.locked_until.is_some());
        assert_eq!(popped.retry_count, 1);
    }

    #[test]
    fn test_unlock_message_by_key_returns_false_for_unlocked_message() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        let msg = make_message("u", "id", "p");
        let key = "u/0/0/id";
        storage
            .messages
            .put(key.as_bytes(), serde_json::to_vec(&msg).unwrap())
            .unwrap();

        let did_unlock = storage.unlock_message_by_key(key).unwrap();
        assert!(!did_unlock);
    }

    #[test]
    fn test_delete_and_purge_queue_emit_info_logs() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

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

        storage.push(make_message("to_delete", "m1", "p1")).unwrap();
        storage.push(make_message("to_delete", "m2", "p2")).unwrap();

        let purged = storage.purge_queue("to_delete").unwrap();
        assert_eq!(purged, 2);

        let deleted = storage.delete_queue("to_delete").unwrap();
        assert_eq!(deleted, 0);
    }

    #[test]
    fn test_tracing_macros_execute_under_scoped_subscriber() {
        let subscriber = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::DEBUG)
            .with_test_writer()
            .finish();
        let dispatch = tracing::Dispatch::new(subscriber);

        tracing::dispatcher::with_default(&dispatch, || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();

            rt.block_on(async {
                let temp_dir = TempDir::new().unwrap();
                let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

                // Covers debug log: "Unlocked expired message..."
                let past = Utc::now() - chrono::Duration::seconds(10);
                let locked = Message {
                    locked_until: Some(past),
                    locked_by: Some("c".to_string()),
                    ..make_message("uql", "id1", "p")
                };
                let key1 = "uql/0/0/id1";
                storage
                    .messages
                    .put(key1.as_bytes(), serde_json::to_vec(&locked).unwrap())
                    .unwrap();
                assert!(storage.unlock_message_by_key(key1).unwrap());

                // Covers debug log: "Removed duplicate message on unlock..."
                storage.create_queue(
                    "nodup2",
                    QueueConfig {
                        ordering: PriorityOrdering::MaxFirst,
                        allow_duplicates: false,
                        ..Default::default()
                    },
                );
                storage
                    .push(make_message("nodup2", "avail", "same"))
                    .unwrap();
                let locked_dup = Message {
                    locked_until: Some(Utc::now() - chrono::Duration::seconds(1)),
                    locked_by: Some("c".to_string()),
                    ..make_message("nodup2", "locked", "same")
                };
                let key2 = "nodup2/0/0/locked";
                storage
                    .messages
                    .put(key2.as_bytes(), serde_json::to_vec(&locked_dup).unwrap())
                    .unwrap();
                assert!(storage.unlock_message_by_key(key2).unwrap());

                // Covers info logs in purge_queue/delete_queue.
                storage
                    .push(make_message("to_delete2", "m1", "p1"))
                    .unwrap();
                storage
                    .push(make_message("to_delete2", "m2", "p2"))
                    .unwrap();

                assert_eq!(storage.purge_queue("to_delete2").unwrap(), 2);
                assert_eq!(storage.delete_queue("to_delete2").unwrap(), 0);
            });
        });
    }

    // =====================================================================
    // SYS-0018 — QueueCounters cache tests
    // =====================================================================

    // Verifies: SYS-0018 — push increments available counter
    #[test]
    fn queue_counters_push_increments_available() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("q1", "m1", "p1")).unwrap();
        storage.push(make_message("q1", "m2", "p2")).unwrap();

        let counters = storage.queue_counters.lock().unwrap();
        let c = &counters["q1"];
        assert_eq!(c.available, 2);
        assert_eq!(c.locked, 0);
    }

    // Verifies: SYS-0018 — pop decrements available, increments locked
    #[test]
    fn queue_counters_pop_moves_available_to_locked() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("q1", "m1", "p1")).unwrap();
        storage.push(make_message("q1", "m2", "p2")).unwrap();
        storage.pop("q1", "c1", 60).unwrap();

        let counters = storage.queue_counters.lock().unwrap();
        let c = &counters["q1"];
        assert_eq!(c.available, 1);
        assert_eq!(c.locked, 1);
    }

    // Verifies: SYS-0018 — ack decrements locked
    #[test]
    fn queue_counters_ack_decrements_locked() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("q1", "m1", "p1")).unwrap();
        let msg = storage.pop("q1", "c1", 60).unwrap().unwrap();
        storage.ack("q1", &msg.id, "c1").unwrap();

        let counters = storage.queue_counters.lock().unwrap();
        let c = &counters["q1"];
        assert_eq!(c.available, 0);
        assert_eq!(c.locked, 0);
    }

    // Verifies: SYS-0018 — nack retry: locked--, available++
    #[test]
    fn queue_counters_nack_retry_moves_locked_to_available() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("q1", "m1", "p1")).unwrap();
        let msg = storage.pop("q1", "c1", 60).unwrap().unwrap();
        storage.nack("q1", &msg.id, "c1").unwrap();

        let counters = storage.queue_counters.lock().unwrap();
        let c = &counters["q1"];
        assert_eq!(c.available, 1);
        assert_eq!(c.locked, 0);
    }

    // Verifies: SYS-0018 — nack DLQ: locked-- only
    #[test]
    fn queue_counters_nack_dlq_decrements_locked() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // Create message with max_retries=0 so first nack goes to DLQ
        let mut msg = make_message("q1", "m1", "p1");
        msg.max_retries = 0;
        storage.push(msg).unwrap();
        let popped = storage.pop("q1", "c1", 60).unwrap().unwrap();
        storage.nack("q1", &popped.id, "c1").unwrap();

        let counters = storage.queue_counters.lock().unwrap();
        let c = &counters["q1"];
        assert_eq!(c.available, 0);
        assert_eq!(c.locked, 0);
    }

    // Verifies: SYS-0018 — batch_ack decrements locked
    #[test]
    fn queue_counters_batch_ack() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("q1", "m1", "p1")).unwrap();
        storage.push(make_message("q1", "m2", "p2")).unwrap();
        let msg1 = storage.pop("q1", "c1", 60).unwrap().unwrap();
        let msg2 = storage.pop("q1", "c1", 60).unwrap().unwrap();

        storage.batch_ack("q1", "c1", &[msg1.id, msg2.id]).unwrap();

        let counters = storage.queue_counters.lock().unwrap();
        let c = &counters["q1"];
        assert_eq!(c.available, 0);
        assert_eq!(c.locked, 0);
    }

    // Verifies: SYS-0018 — batch_nack with mixed retry and DLQ
    #[test]
    fn queue_counters_batch_nack_mixed() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // m1: max_retries=3 => will retry (locked--, available++)
        storage.push(make_message("q1", "m1", "p1")).unwrap();
        // m2: max_retries=0 => will DLQ (locked--)
        let mut m2 = make_message("q1", "m2", "p2");
        m2.max_retries = 0;
        storage.push(m2).unwrap();

        let msg1 = storage.pop("q1", "c1", 60).unwrap().unwrap();
        let msg2 = storage.pop("q1", "c1", 60).unwrap().unwrap();

        storage.batch_nack("q1", "c1", &[msg1.id, msg2.id]).unwrap();

        let counters = storage.queue_counters.lock().unwrap();
        let c = &counters["q1"];
        // One retried (back to available), one DLQ'd (gone)
        assert_eq!(c.available, 1);
        assert_eq!(c.locked, 0);
    }

    // Verifies: SYS-0018 — delete_queue removes cache entry
    #[test]
    fn queue_counters_delete_queue_removes_entry() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("q1", "m1", "p1")).unwrap();
        storage.delete_queue("q1").unwrap();

        let counters = storage.queue_counters.lock().unwrap();
        assert!(!counters.contains_key("q1"));
    }

    // Verifies: SYS-0018 — purge_queue zeros counts
    #[test]
    fn queue_counters_purge_queue_zeros_counts() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("q1", "m1", "p1")).unwrap();
        storage.push(make_message("q1", "m2", "p2")).unwrap();
        storage.pop("q1", "c1", 60).unwrap();

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

        let counters = storage.queue_counters.lock().unwrap();
        let c = &counters["q1"];
        assert_eq!(c.available, 0);
        assert_eq!(c.locked, 0);
    }

    // Verifies: SYS-0018 — rename_queue moves counter entry
    #[test]
    fn queue_counters_rename_queue_moves_entry() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("old", "m1", "p1")).unwrap();
        storage.push(make_message("old", "m2", "p2")).unwrap();
        storage.pop("old", "c1", 60).unwrap();

        storage.rename_queue("old", "new").unwrap();

        let counters = storage.queue_counters.lock().unwrap();
        assert!(!counters.contains_key("old"));
        let c = &counters["new"];
        assert_eq!(c.available, 1);
        assert_eq!(c.locked, 1);
    }

    // Verifies: SYS-0018 — seeding from DB on restart
    #[test]
    fn queue_counters_seeded_on_restart() {
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().to_str().unwrap();

        // Phase 1: populate
        {
            let storage = Storage::new(path).unwrap();
            storage.push(make_message("q1", "m1", "p1")).unwrap();
            storage.push(make_message("q1", "m2", "p2")).unwrap();
            storage.push(make_message("q2", "m3", "p3")).unwrap();
            storage.pop("q1", "c1", 3600).unwrap(); // lock one in q1
        }

        // Phase 2: reopen — counters should be seeded from the DB
        {
            let storage = Storage::new(path).unwrap();
            let counters = storage.queue_counters.lock().unwrap();

            let c1 = &counters["q1"];
            assert_eq!(c1.available, 1);
            assert_eq!(c1.locked, 1);

            let c2 = &counters["q2"];
            assert_eq!(c2.available, 1);
            assert_eq!(c2.locked, 0);
        }
    }

    // Verifies: SYS-0018 — get_all_queue_stats uses cache
    #[test]
    fn queue_counters_get_all_queue_stats_from_cache() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("alpha", "m1", "p1")).unwrap();
        storage.push(make_message("beta", "m2", "p2")).unwrap();
        storage.pop("alpha", "c1", 60).unwrap();

        let stats = storage.get_all_queue_stats().unwrap();
        assert_eq!(stats.len(), 2);

        let alpha = stats.iter().find(|s| s.name == "alpha").unwrap();
        assert_eq!(alpha.available, 0);
        assert_eq!(alpha.locked, 1);
        assert_eq!(alpha.total, 1);

        let beta = stats.iter().find(|s| s.name == "beta").unwrap();
        assert_eq!(beta.available, 1);
        assert_eq!(beta.locked, 0);
        assert_eq!(beta.total, 1);
    }

    // Verifies: SYS-0018 — list_queues reads from cache
    #[test]
    fn queue_counters_list_queues_from_cache() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.push(make_message("zebra", "m1", "p1")).unwrap();
        storage.push(make_message("apple", "m2", "p2")).unwrap();

        let queues = storage.list_queues().unwrap();
        assert_eq!(queues, vec!["apple", "zebra"]);
    }

    // Verifies: SYS-0018 — multiple operations maintain consistency
    #[test]
    fn queue_counters_full_lifecycle() {
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // Push 3 messages
        storage.push(make_message("q", "m1", "p1")).unwrap();
        storage.push(make_message("q", "m2", "p2")).unwrap();
        storage.push(make_message("q", "m3", "p3")).unwrap();

        // Pop 2
        let msg1 = storage.pop("q", "c1", 60).unwrap().unwrap();
        let msg2 = storage.pop("q", "c1", 60).unwrap().unwrap();

        {
            let c = storage.queue_counters.lock().unwrap();
            assert_eq!(c["q"].available, 1);
            assert_eq!(c["q"].locked, 2);
        }

        // Ack one
        storage.ack("q", &msg1.id, "c1").unwrap();
        {
            let c = storage.queue_counters.lock().unwrap();
            assert_eq!(c["q"].available, 1);
            assert_eq!(c["q"].locked, 1);
        }

        // Nack the other (retry)
        storage.nack("q", &msg2.id, "c1").unwrap();
        {
            let c = storage.queue_counters.lock().unwrap();
            assert_eq!(c["q"].available, 2);
            assert_eq!(c["q"].locked, 0);
        }

        // Verify stats match
        let stats = storage.get_queue_stats("q").unwrap();
        assert_eq!(stats.available, 2);
        assert_eq!(stats.locked, 0);
        assert_eq!(stats.total, 2);
    }

    // ── SYS-0025: Aggressive memory reclamation for empty queues ──

    /// Empty queues with dedup enabled must have their payload_sets cleared
    /// and shrunk after release_memory_for_empty_queues().
    // Verifies: SYS-0025
    #[test]
    fn test_empty_queue_dedup_set_reclaimed() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // Configure queue with dedup enabled.
        storage.configure_queue(
            "reclaim_q",
            QueueConfig {
                allow_duplicates: false,
                ..Default::default()
            },
        );

        // Push and ack messages to build up HashSet capacity.
        for i in 0..100 {
            let msg = make_message("reclaim_q", &format!("id{}", i), &format!("payload-{}", i));
            storage.push(msg).unwrap();
        }
        for _ in 0..100 {
            let popped = storage.pop("reclaim_q", "c1", 30).unwrap().unwrap();
            storage.ack("reclaim_q", &popped.id, "c1").unwrap();
        }

        // Queue is empty, but HashSet still has allocated capacity.
        {
            let sets = storage.payload_sets.read().unwrap();
            let set = sets.get("reclaim_q").expect("set should exist");
            assert_eq!(set.len(), 0, "set should be empty");
            assert!(
                set.capacity() > 0,
                "set should still have allocated capacity"
            );
        }

        // Run reclamation.
        storage.release_memory_for_empty_queues();

        // HashSet should be shrunk (capacity reduced to minimum).
        {
            let sets = storage.payload_sets.read().unwrap();
            let set = sets
                .get("reclaim_q")
                .expect("set should still exist for dedup queues");
            assert_eq!(
                set.capacity(),
                0,
                "empty set capacity should be 0 after reclamation"
            );
        }
    }

    /// Empty queues must have their hot tier entry removed after reclamation.
    // Verifies: SYS-0025
    #[test]
    fn test_empty_queue_hot_tier_removed() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // Push and consume a message so hot tier entry is created.
        let msg = make_message("ht_reclaim_q", "id1", "payload");
        storage.push(msg).unwrap();
        let popped = storage.pop("ht_reclaim_q", "c1", 30).unwrap().unwrap();
        storage.ack("ht_reclaim_q", &popped.id, "c1").unwrap();

        // Hot tier entry exists (even if empty BTreeMap).
        {
            let tiers = storage.hot_tier.read().unwrap();
            assert!(
                tiers.contains_key("ht_reclaim_q"),
                "hot tier entry should exist"
            );
        }

        // Run reclamation.
        storage.release_memory_for_empty_queues();

        // Hot tier entry should be gone.
        {
            let tiers = storage.hot_tier.read().unwrap();
            assert!(
                !tiers.contains_key("ht_reclaim_q"),
                "hot tier entry should be removed for empty queue"
            );
        }
    }

    /// Empty queues must have locked_index entries cleared after reclamation.
    // Verifies: SYS-0025
    #[test]
    fn test_empty_queue_locked_index_cleaned() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // Manually inject a stale locked_index entry for an empty queue.
        {
            let mut li = storage.locked_index.write().unwrap();
            li.insert(
                "orphan_q/0/0/stale_id".to_string(),
                Utc::now() - chrono::Duration::seconds(100),
            );
        }

        // Ensure the queue has zero counters.
        {
            let mut counters = storage.queue_counters.lock().unwrap();
            counters.insert(
                "orphan_q".to_string(),
                QueueCounts {
                    available: 0,
                    locked: 0,
                },
            );
        }

        storage.release_memory_for_empty_queues();

        // Stale locked_index entry should be removed.
        {
            let li = storage.locked_index.read().unwrap();
            assert!(
                !li.contains_key("orphan_q/0/0/stale_id"),
                "locked_index entry should be removed for empty queue"
            );
        }
    }

    /// Non-empty dedup sets should be shrunk (shrink_to_fit) during
    /// release_memory_for_empty_queues, reducing excess bucket capacity.
    // Verifies: SYS-0025
    #[test]
    fn test_nonempty_dedup_set_shrunk() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.configure_queue(
            "shrink_q",
            QueueConfig {
                allow_duplicates: false,
                ..Default::default()
            },
        );

        // Push many messages to grow the HashSet, then ack most of them.
        for i in 0..200 {
            let msg = make_message("shrink_q", &format!("id{}", i), &format!("p-{}", i));
            storage.push(msg).unwrap();
        }
        // Pop and ack 190 of them, leaving 10.
        for _ in 0..190 {
            let popped = storage.pop("shrink_q", "c1", 30).unwrap().unwrap();
            storage.ack("shrink_q", &popped.id, "c1").unwrap();
        }

        let capacity_before = {
            let sets = storage.payload_sets.read().unwrap();
            let set = sets.get("shrink_q").unwrap();
            assert_eq!(set.len(), 10);
            set.capacity()
        };

        storage.release_memory_for_empty_queues();

        let capacity_after = {
            let sets = storage.payload_sets.read().unwrap();
            let set = sets.get("shrink_q").unwrap();
            assert_eq!(set.len(), 10, "entries must not be lost");
            set.capacity()
        };

        assert!(
            capacity_after < capacity_before,
            "shrink_to_fit should reduce capacity from {} to {}",
            capacity_before,
            capacity_after
        );
    }

    /// The global locked_index should be shrunk after reclamation.
    // Verifies: SYS-0025
    #[test]
    fn test_locked_index_shrunk() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // Inflate locked_index with many entries, then remove most.
        {
            let mut li = storage.locked_index.write().unwrap();
            for i in 0..500 {
                li.insert(format!("q/0/0/id{}", i), Utc::now());
            }
        }
        {
            let mut li = storage.locked_index.write().unwrap();
            for i in 0..490 {
                li.remove(&format!("q/0/0/id{}", i));
            }
        }

        let capacity_before = {
            let li = storage.locked_index.read().unwrap();
            assert_eq!(li.len(), 10);
            li.capacity()
        };

        storage.release_memory_for_empty_queues();

        let capacity_after = {
            let li = storage.locked_index.read().unwrap();
            assert_eq!(li.len(), 10, "entries must not be lost");
            li.capacity()
        };

        assert!(
            capacity_after < capacity_before,
            "locked_index should shrink from {} to {}",
            capacity_before,
            capacity_after
        );
    }

    /// Non-empty queues must NOT have their hot tier or dedup set removed.
    // Verifies: SYS-0025
    #[test]
    fn test_nonempty_queue_not_reclaimed() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.configure_queue(
            "active_q",
            QueueConfig {
                allow_duplicates: false,
                ..Default::default()
            },
        );

        let msg = make_message("active_q", "id1", "payload");
        storage.push(msg).unwrap();

        storage.release_memory_for_empty_queues();

        // Hot tier must still exist.
        {
            let tiers = storage.hot_tier.read().unwrap();
            assert!(
                tiers.contains_key("active_q"),
                "active queue hot tier must survive"
            );
        }

        // Dedup set must still have entries.
        {
            let sets = storage.payload_sets.read().unwrap();
            let set = sets.get("active_q").unwrap();
            assert_eq!(
                set.len(),
                1,
                "dedup set must retain entries for active queue"
            );
        }
    }

    // ── PER-0015: RocksDB default tuning ──

    /// Default block cache should be 128 MB (not the old 256 MB).
    // Verifies: PER-0015
    #[test]
    fn test_default_block_cache_128mb() {
        init_tracing_for_tests();
        let _guard = ENV_LOCK.lock().unwrap();
        std::env::remove_var("ROCKSDB_CACHE_MB");
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();
        assert_eq!(
            storage.rocksdb_cache_capacity,
            128 * 1024 * 1024,
            "default block cache should be 128 MB"
        );
    }

    /// Default max open files should be 128 (not the old 256).
    /// We verify indirectly: the storage opens successfully with the
    /// default, and we check the configured capacity field.
    // Verifies: PER-0015
    #[test]
    fn test_default_max_open_files_128() {
        init_tracing_for_tests();
        let _guard = ENV_LOCK.lock().unwrap();
        std::env::remove_var("ROCKSDB_MAX_OPEN_FILES");
        let temp_dir = TempDir::new().unwrap();
        // If the default changed correctly, open succeeds.
        // We can't inspect RocksDB options after open, so this is a
        // smoke test that the new default doesn't break anything.
        let storage = Storage::new(temp_dir.path().to_str().unwrap());
        assert!(
            storage.is_ok(),
            "storage should open with default max_open_files=128"
        );
    }

    // ── SYS-0022: Memory breakdown for pressure log ──

    /// memory_breakdown() should return all expected fields.
    // Verifies: SYS-0022
    #[test]
    fn test_memory_breakdown_returns_expected_fields() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // Push a message so there's something in the structures.
        storage
            .push(make_message("breakdown_q", "id1", "payload"))
            .unwrap();

        let breakdown = storage.memory_breakdown();
        assert!(breakdown.contains_key("block_cache_capacity_mb"));
        assert!(breakdown.contains_key("hot_tier_entries"));
        assert!(breakdown.contains_key("dedup_set_entries"));
        assert!(breakdown.contains_key("locked_index_entries"));
    }

    /// memory_breakdown() should reflect actual state after push/pop.
    // Verifies: SYS-0022
    #[test]
    fn test_memory_breakdown_reflects_state() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        storage.configure_queue(
            "bd_q",
            QueueConfig {
                allow_duplicates: false,
                ..Default::default()
            },
        );

        // Push 3 messages to a no-dup queue.
        for i in 0..3 {
            storage
                .push(make_message(
                    "bd_q",
                    &format!("id{}", i),
                    &format!("p{}", i),
                ))
                .unwrap();
        }

        let bd = storage.memory_breakdown();
        assert_eq!(bd["dedup_set_entries"], 3, "3 messages in dedup set");

        // Pop one (locks it).
        storage.pop("bd_q", "c1", 30).unwrap();
        let bd = storage.memory_breakdown();
        assert!(bd["locked_index_entries"] >= 1, "should have locked entry");
    }

    /// Regression: when all messages in a queue are locked and the hot tier
    /// is empty, pop() must still recover once locks expire.  This tests
    /// the full cycle: push → pop-all (lock) → locks expire → unlock →
    /// pop again.  The fix ensures pop() triggers a refill on the empty
    /// hot tier even when returning None, and unlock_expired_messages()
    /// triggers a batch refill for affected queues.
    // Verifies: SYS-0020
    #[test]
    fn test_hot_tier_starvation_recovery() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // Push 5 messages, pop all with short lock.
        for i in 0..5 {
            storage
                .push(make_message(
                    "starve_q",
                    &format!("m{}", i),
                    &format!("p{}", i),
                ))
                .unwrap();
        }
        for _ in 0..5 {
            storage.pop("starve_q", "w1", 1).unwrap().unwrap();
        }

        // Hot tier empty, all locked.
        assert!(storage.pop("starve_q", "w2", 30).unwrap().is_none());

        // Wait for locks to expire, then unlock.
        std::thread::sleep(std::time::Duration::from_secs(2));
        let unlocked = storage.unlock_expired_messages().unwrap();
        assert_eq!(unlocked, 5);

        // Hot tier must be repopulated after batch unlock.
        {
            let tiers = storage.hot_tier.read().unwrap();
            let tier_len = tiers.get("starve_q").map_or(0, |t| t.len());
            assert_eq!(
                tier_len, 5,
                "hot tier must have all 5 messages after batch unlock refill"
            );
        }

        // All 5 messages must be poppable.
        let mut count = 0;
        for _ in 0..5 {
            if storage.pop("starve_q", "w2", 30).unwrap().is_some() {
                count += 1;
            }
        }
        assert_eq!(count, 5, "all 5 unlocked messages must be poppable");
    }

    /// Pop returning None on a queue with available messages (per counters)
    /// must trigger a refill so the hot tier is primed for the next attempt.
    // Verifies: SYS-0020
    #[test]
    fn test_failed_pop_refills_hot_tier() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();

        // Push messages, then manually clear the hot tier to simulate
        // the state after a refill-during-all-locked replaced it.
        for i in 0..3 {
            storage
                .push(make_message(
                    "fpop_q",
                    &format!("m{}", i),
                    &format!("p{}", i),
                ))
                .unwrap();
        }
        // Pop all with short lock.
        for _ in 0..3 {
            storage.pop("fpop_q", "w1", 1).unwrap().unwrap();
        }
        assert!(storage.pop("fpop_q", "w2", 30).unwrap().is_none());

        // After the failed pop, wait and unlock.
        std::thread::sleep(std::time::Duration::from_secs(2));
        storage.unlock_expired_messages().unwrap();

        // The next pop must succeed (hot tier was refilled).
        assert!(
            storage.pop("fpop_q", "w2", 30).unwrap().is_some(),
            "pop must find unlocked messages after failed pop + unlock cycle"
        );
    }

    /// Old messages in cold storage must not be starved when new publishes
    /// keep the hot tier full.  Reproduces the scenario where:
    ///
    /// 1. Queue has more messages than hot_tier_capacity (some in cold storage).
    /// 2. Workers pop and ack from the hot tier.
    /// 3. New publishes continuously refill the hot tier via try_insert_hot_tier.
    /// 4. Old cold-storage messages must eventually be served.
    ///
    /// Without the cold-message refill check in pop(), the old messages
    /// would remain in RocksDB indefinitely because the hot tier never
    /// drops below refill_threshold.
    // Verifies: SYS-0020, SYS-0021
    #[test]
    fn test_cold_storage_starvation_with_continuous_publish() {
        init_tracing_for_tests();
        let temp_dir = TempDir::new().unwrap();
        // Small hot tier: capacity=5, refill_threshold=2
        std::env::set_var("QRUSTY_HOT_TIER_SIZE", "5");
        std::env::set_var("QRUSTY_REFILL_THRESHOLD", "2");
        let storage = Storage::new(temp_dir.path().to_str().unwrap()).unwrap();
        std::env::remove_var("QRUSTY_HOT_TIER_SIZE");
        std::env::remove_var("QRUSTY_REFILL_THRESHOLD");

        let queue = "cold_starve_q";

        // Push 8 messages (5 fill hot tier, 3 overflow to cold storage).
        // Use descending priority numbers so earlier messages have higher
        // priority (lower key) and should be served first.
        for i in 0..8u32 {
            let mut msg = make_message(queue, &format!("old-{i}"), &format!("payload-{i}"));
            msg.priority = Priority::Numeric(5); // same priority, FIFO by timestamp
            storage.push(msg).unwrap();
        }

        // Hot tier should have 5 messages; 3 are in cold storage.
        {
            let tiers = storage.hot_tier.read().unwrap();
            assert_eq!(tiers.get(queue).map_or(0, |t| t.len()), 5);
        }

        // Pop one message (ack it) and immediately publish a new one.
        // This simulates continuous traffic that keeps the hot tier full.
        let popped = storage.pop(queue, "w1", 30).unwrap().unwrap();
        storage.ack(queue, &popped.id, "w1").unwrap();

        // Publish a brand-new message (goes into hot tier via try_insert).
        let mut fresh = make_message(queue, "fresh-0", "fresh-payload");
        fresh.priority = Priority::Numeric(5);
        storage.push(fresh).unwrap();

        // After the pop+ack+publish cycle, the hot tier should have been
        // refilled with cold-storage messages (the refill rebuilds from
        // RocksDB, pulling oldest first).
        {
            let tiers = storage.hot_tier.read().unwrap();
            let tier = tiers.get(queue).unwrap();
            assert_eq!(tier.len(), 5, "hot tier must be at capacity after refill");
        }

        // Drain all remaining messages — we should get all 8 remaining
        // (7 old ones minus the 1 acked = 7, plus the 1 fresh = 8).
        let mut ids = Vec::new();
        for _ in 0..9 {
            if let Some(msg) = storage.pop(queue, "w1", 30).unwrap() {
                ids.push(msg.id.clone());
                storage.ack(queue, &msg.id, "w1").unwrap();
            }
        }

        // Must have recovered the cold-storage messages.
        assert_eq!(
            ids.len(),
            8,
            "all 8 remaining messages (including cold-storage) must be delivered"
        );
        // Verify old cold-storage messages were included.
        let old_count = ids.iter().filter(|id| id.starts_with("old-")).count();
        assert!(
            old_count >= 7,
            "all 7 remaining old messages must be delivered (got {old_count})"
        );
    }
}