use std::collections::HashMap;
use std::num::NonZeroUsize;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use lru::LruCache;
use tokio::fs::{self, File, OpenOptions};
use tokio::io::AsyncWriteExt;
use tokio::sync::{Mutex, RwLock};

use crate::config::{FsyncStrategy, StorageConfig};
use crate::error::Result;
use crate::storage::StorageBackend;
use crate::storage::replay;
use crate::types::{NamespaceId, PublishedMessage, TopicName, TopicOffset};

const DEFAULT_SEGMENT_MAX_BYTES: u64 = 64 * 1024 * 1024; // 64 MiB
const DEFAULT_INDEX_INTERVAL: u32 = 256;
const INDEX_ENTRY_SIZE: usize = 24;

/// Single entry in a sparse segment index.
struct IndexEntry {
    offset: TopicOffset,
    byte_position: u64,
    timestamp_millis: u64,
}

/// Sparse offset/timestamp index for a single segment file.
///
/// Entries are sorted by offset. Loaded entirely into memory for
/// binary search. Typically < 100KB per segment.
struct SegmentIndex {
    entries: Vec<IndexEntry>,
}

impl SegmentIndex {
    /// Load an index from disk. Returns an empty index if the file
    /// doesn't exist or is empty.
    async fn load(path: &Path) -> Result<Self> {
        let data = match fs::read(path).await {
            Ok(d) => d,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                return Ok(Self {
                    entries: Vec::new(),
                });
            }
            Err(e) => return Err(e.into()),
        };

        let count = data.len() / INDEX_ENTRY_SIZE;
        let mut entries = Vec::with_capacity(count);
        for i in 0..count {
            let base = i * INDEX_ENTRY_SIZE;
            let offset = TopicOffset(u64::from_be_bytes(data[base..base + 8].try_into().unwrap()));
            let byte_position = u64::from_be_bytes(data[base + 8..base + 16].try_into().unwrap());
            let timestamp_millis =
                u64::from_be_bytes(data[base + 16..base + 24].try_into().unwrap());
            entries.push(IndexEntry {
                offset,
                byte_position,
                timestamp_millis,
            });
        }

        Ok(Self { entries })
    }

    /// Binary search by offset. Returns the byte_position of the closest
    /// entry whose offset is <= `target`, or `None` if target < first entry.
    fn find_position(&self, target: TopicOffset) -> Option<u64> {
        if self.entries.is_empty() {
            return None;
        }
        let idx = match self.entries.binary_search_by_key(&target, |e| e.offset) {
            Ok(i) => i,
            Err(0) => return None,
            Err(i) => i - 1,
        };
        Some(self.entries[idx].byte_position)
    }

    /// Binary search on timestamp. Returns `(offset, byte_position)` of the
    /// closest entry whose timestamp is <= `ts`, or `None` if ts < first entry.
    fn find_offset_at_timestamp(&self, ts: u64) -> Option<(TopicOffset, u64)> {
        if self.entries.is_empty() {
            return None;
        }
        let idx = match self
            .entries
            .binary_search_by_key(&ts, |e| e.timestamp_millis)
        {
            Ok(i) => i,
            Err(0) => return None,
            Err(i) => i - 1,
        };
        let e = &self.entries[idx];
        Some((e.offset, e.byte_position))
    }
}

/// Key for identifying a topic within a namespace.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct TopicKey {
    namespace: String,
    topic: String,
}

/// Tracks the currently-active segment file for a topic.
struct ActiveSegment {
    file: File,
    size: u64,
    base_offset: TopicOffset,
    /// Open `.idx` file handle for writing sparse index entries.
    index_file: Option<File>,
    /// Records written since the last index entry.
    records_since_last_index: u32,
}

/// Filesystem-backed storage using append-only segment files.
///
/// Each namespace/topic combination gets its own directory of segment files.
/// Segments are named by their base offset and are rotated when they exceed
/// `segment_max_bytes`.
///
/// An LRU cache bounded by [`max_open_fds`](FsStorage::max_open_fds) manages
/// open segment writers. When the limit is reached the least-recently-used
/// topic's file handle is evicted (closed); it will be transparently reopened
/// on the next write. A [`Mutex`] protects the LRU cache, and each topic has
/// its own inner [`Mutex`] for serializing writes to its segment file.
///
/// Writes go to the OS page cache without fsync — call
/// [`flush`](StorageBackend::flush) or [`shutdown`](StorageBackend::shutdown)
/// for durability.
pub struct FsStorage {
    data_dir: PathBuf,
    segment_max_bytes: u64,
    fsync_strategy: FsyncStrategy,
    /// Maximum segments per topic. Used by the background cleanup task.
    pub max_segments_per_topic: usize,
    /// Maximum open file descriptors for segment files.
    ///
    /// When the number of active segment writers reaches this limit, the
    /// least-recently-used topic's file handle is evicted (closed). The next
    /// write to that topic will transparently reopen the segment file.
    max_open_fds: usize,
    /// Records between sparse index entries in `.idx` files.
    index_interval: u32,
    writers: Mutex<LruCache<TopicKey, Arc<Mutex<ActiveSegment>>>>,
    /// Cached sorted list of `(base_offset, path)` per topic for fast lookups.
    /// Populated on `init()`, updated on segment rotation and deletion.
    segment_index: RwLock<HashMap<TopicKey, Vec<(TopicOffset, PathBuf)>>>,
    /// Cached `.idx` contents keyed by segment path. Avoids re-reading index
    /// files from disk on every `read_at` / `skip_to` call.
    index_cache: RwLock<HashMap<PathBuf, Arc<SegmentIndex>>>,
    /// Tracked disk bytes per namespace, updated incrementally on append/delete.
    /// Outer RwLock guards the HashMap (rare inserts); inner AtomicU64 allows
    /// concurrent updates under a read lock.
    namespace_disk_bytes: RwLock<HashMap<String, AtomicU64>>,
}

impl FsStorage {
    /// Create a new `FsStorage` with the default 64 MiB segment max.
    pub fn new(data_dir: impl Into<PathBuf>) -> Self {
        let max_open_fds = 1024;
        Self {
            data_dir: data_dir.into(),
            segment_max_bytes: DEFAULT_SEGMENT_MAX_BYTES,
            fsync_strategy: FsyncStrategy::EveryBatch,
            max_segments_per_topic: 64,
            max_open_fds,
            index_interval: DEFAULT_INDEX_INTERVAL,
            writers: Mutex::new(LruCache::new(
                NonZeroUsize::new(max_open_fds).expect("max_open_fds must be > 0"),
            )),
            segment_index: RwLock::new(HashMap::new()),
            index_cache: RwLock::new(HashMap::new()),
            namespace_disk_bytes: RwLock::new(HashMap::new()),
        }
    }

    /// Create a new `FsStorage` with a custom segment max size (useful for tests).
    pub fn with_segment_max_bytes(data_dir: impl Into<PathBuf>, max: u64) -> Self {
        let max_open_fds = 1024;
        Self {
            data_dir: data_dir.into(),
            segment_max_bytes: max,
            fsync_strategy: FsyncStrategy::EveryBatch,
            max_segments_per_topic: 64,
            max_open_fds,
            index_interval: DEFAULT_INDEX_INTERVAL,
            writers: Mutex::new(LruCache::new(
                NonZeroUsize::new(max_open_fds).expect("max_open_fds must be > 0"),
            )),
            segment_index: RwLock::new(HashMap::new()),
            index_cache: RwLock::new(HashMap::new()),
            namespace_disk_bytes: RwLock::new(HashMap::new()),
        }
    }

    /// Create a new `FsStorage` from a [`StorageConfig`].
    pub fn from_config(config: &StorageConfig) -> Self {
        Self {
            data_dir: PathBuf::from(&config.data_dir),
            segment_max_bytes: DEFAULT_SEGMENT_MAX_BYTES,
            fsync_strategy: config.fsync_strategy,
            max_segments_per_topic: config.max_segments_per_topic,
            max_open_fds: config.max_open_fds,
            index_interval: config.index_interval,
            writers: Mutex::new(LruCache::new(
                NonZeroUsize::new(config.max_open_fds).expect("max_open_fds must be > 0"),
            )),
            segment_index: RwLock::new(HashMap::new()),
            index_cache: RwLock::new(HashMap::new()),
            namespace_disk_bytes: RwLock::new(HashMap::new()),
        }
    }

    /// Returns the configured maximum number of open file descriptors.
    pub fn max_open_fds(&self) -> usize {
        self.max_open_fds
    }

    /// Returns the configured data directory.
    pub fn data_dir(&self) -> &Path {
        &self.data_dir
    }

    /// Load a `SegmentIndex` from cache or disk. Caches the result for future calls.
    async fn load_index(&self, seg_path: &Path) -> Option<Arc<SegmentIndex>> {
        let idx = index_path(seg_path);
        // Fast path: check cache
        {
            let cache = self.index_cache.read().await;
            if let Some(cached) = cache.get(&idx) {
                return Some(Arc::clone(cached));
            }
        }
        // Slow path: load from disk and cache
        let loaded = SegmentIndex::load(&idx).await.ok()?;
        if loaded.entries.is_empty() {
            return None;
        }
        let arc = Arc::new(loaded);
        {
            let mut cache = self.index_cache.write().await;
            cache.insert(idx, Arc::clone(&arc));
        }
        Some(arc)
    }
}

/// Binary search for the segment whose base_offset is the highest value <= `offset`.
/// Returns the path of the matching segment, or `None` if `offset` is before all segments.
fn find_segment_for_offset(
    entries: &[(TopicOffset, PathBuf)],
    offset: TopicOffset,
) -> Option<PathBuf> {
    if entries.is_empty() {
        return None;
    }
    let idx = match entries.binary_search_by_key(&offset, |(base, _)| *base) {
        Ok(i) => i,
        Err(0) => return None,
        Err(i) => i - 1,
    };
    Some(entries[idx].1.clone())
}

/// Returns the path for a segment file with the given base offset.
fn segment_path(seg_dir: &Path, base_offset: TopicOffset) -> PathBuf {
    seg_dir.join(format!("{:020}.seg", base_offset.0))
}

/// Segment record format version constants.
///
/// The sentinel byte `0x00` distinguishes v2+ records from v1 (legacy) records.
/// In v1 format, the first byte after the offset is the low byte of a
/// bincode-serialized `MessageId` (u64 LE), which is always ≥ 1.
const RECORD_FORMAT_SENTINEL: u8 = 0x00;
/// Version 2: adds `headers` field to `MessageEnvelope`.
const RECORD_FORMAT_VERSION_2: u8 = 0x02;

/// Encode a `PublishedMessage` as a record for on-disk storage.
///
/// Record format (v2):
/// ```text
/// [4 bytes: u32 BE record_length (= 8 + 2 + envelope_bytes)]
/// [8 bytes: u64 BE TopicOffset]
/// [1 byte:  0x00 sentinel — distinguishes v2 from v1]
/// [1 byte:  0x02 format version]
/// [N bytes: bincode-serialized MessageEnvelope (with headers)]
/// ```
///
/// Public for use by tests and `MockStorage` to compute consistent record sizes.
pub fn encode_record(msg: &PublishedMessage) -> Result<Vec<u8>> {
    debug_assert!(
        msg.envelope.id.0 != 0,
        "MessageId must never be 0 — the v1/v2 format detection relies on this invariant"
    );
    let envelope_bytes = bincode::serialize(&msg.envelope)?;
    // 8 (offset) + 2 (sentinel + version) + envelope
    let record_len = (8 + 2 + envelope_bytes.len()) as u32;
    let mut buf = Vec::with_capacity(4 + record_len as usize);
    buf.extend_from_slice(&record_len.to_be_bytes());
    buf.extend_from_slice(&msg.offset.0.to_be_bytes());
    buf.push(RECORD_FORMAT_SENTINEL);
    buf.push(RECORD_FORMAT_VERSION_2);
    buf.extend_from_slice(&envelope_bytes);
    Ok(buf)
}

/// Returns the path for an index file alongside the given segment path.
fn index_path(seg_path: &Path) -> PathBuf {
    seg_path.with_extension("idx")
}

/// Open or create an `.idx` file for appending sparse index entries.
async fn open_index_file(seg_path: &Path) -> Result<File> {
    let path = index_path(seg_path);
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).await?;
    }
    let file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
        .await?;
    Ok(file)
}

/// Open or create a segment file for appending. Returns the file handle and current size.
async fn open_segment(path: &Path) -> Result<(File, u64)> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).await?;
    }
    let file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .await?;
    let metadata = file.metadata().await?;
    let size = metadata.len();
    Ok((file, size))
}

/// Scan the segments directory and return the highest base offset found, if any.
async fn find_latest_segment_offset(seg_dir: &Path) -> Result<Option<TopicOffset>> {
    if !seg_dir.exists() {
        return Ok(None);
    }

    let mut max_offset: Option<u64> = None;
    let mut read_dir = fs::read_dir(seg_dir).await?;
    while let Some(entry) = read_dir.next_entry().await? {
        let path = entry.path();
        if path.extension().and_then(|e| e.to_str()) == Some("seg")
            && let Some(stem) = path.file_stem().and_then(|s| s.to_str())
            && let Ok(offset) = stem.parse::<u64>()
        {
            max_offset = Some(max_offset.map_or(offset, |m: u64| m.max(offset)));
        }
    }

    Ok(max_offset.map(TopicOffset))
}

/// Rebuild a `.idx` file for a segment by scanning all its records.
///
/// Uses `RecordStream::position()` to track byte positions without
/// re-serializing envelopes. Writes an index entry for the first record
/// and every `index_interval` records thereafter.
async fn rebuild_index(seg_path: &Path, index_interval: u32) -> Result<()> {
    let idx_path = index_path(seg_path);
    let mut idx_file = OpenOptions::new()
        .create(true)
        .write(true)
        .truncate(true)
        .open(&idx_path)
        .await?;

    let mut stream = replay::RecordStream::open(seg_path).await?;
    let mut records_since_last: u32 = index_interval; // trigger on first record
    let mut is_first = true;

    loop {
        let record_byte_pos = stream.position();
        let Some((offset, envelope)) = stream.next().await? else {
            break;
        };

        records_since_last += 1;
        if records_since_last >= index_interval || is_first {
            let mut entry = [0u8; INDEX_ENTRY_SIZE];
            entry[..8].copy_from_slice(&offset.0.to_be_bytes());
            entry[8..16].copy_from_slice(&record_byte_pos.to_be_bytes());
            entry[16..24].copy_from_slice(&envelope.timestamp.to_be_bytes());
            idx_file.write_all(&entry).await?;
            records_since_last = 0;
            is_first = false;
        }
    }

    idx_file.flush().await?;
    Ok(())
}

impl StorageBackend for FsStorage {
    async fn init(&self) -> Result<()> {
        fs::create_dir_all(&self.data_dir).await?;

        // Pre-populate writers and segment index from existing segment directories.
        let mut writer_entries = Vec::new();
        let mut index_entries: HashMap<TopicKey, Vec<(TopicOffset, PathBuf)>> = HashMap::new();
        let mut disk_bytes: HashMap<String, u64> = HashMap::new();

        let mut ns_dir = fs::read_dir(&self.data_dir).await?;
        while let Some(ns_entry) = ns_dir.next_entry().await? {
            if !ns_entry.file_type().await?.is_dir() {
                continue;
            }
            let ns_name = ns_entry.file_name().to_string_lossy().into_owned();

            let mut topic_dir = fs::read_dir(ns_entry.path()).await?;
            while let Some(topic_entry) = topic_dir.next_entry().await? {
                if !topic_entry.file_type().await?.is_dir() {
                    continue;
                }
                let topic_name = topic_entry.file_name().to_string_lossy().into_owned();
                let seg_dir = topic_entry.path().join("segments");
                let key = TopicKey {
                    namespace: ns_name.clone(),
                    topic: topic_name,
                };

                // Build segment index from all segment files
                let seg_paths = replay::list_segments(&seg_dir).await?;
                let mut seg_index = Vec::new();
                let mut latest_offset: Option<TopicOffset> = None;
                for path in &seg_paths {
                    if let Some(stem) = path.file_stem().and_then(|s| s.to_str())
                        && let Ok(base) = stem.parse::<u64>()
                    {
                        let offset = TopicOffset(base);
                        seg_index.push((offset, path.clone()));
                        latest_offset =
                            Some(latest_offset.map_or(offset, |m| std::cmp::max(m, offset)));

                        // Track segment + index file sizes for namespace disk bytes
                        if let Ok(meta) = fs::metadata(path).await {
                            *disk_bytes.entry(ns_name.clone()).or_default() += meta.len();
                        }
                        let idx_path = index_path(path);
                        if let Ok(meta) = fs::metadata(&idx_path).await {
                            *disk_bytes.entry(ns_name.clone()).or_default() += meta.len();
                        }
                    }
                }

                if !seg_index.is_empty() {
                    // Rebuild missing .idx files for non-active segments
                    for (_, seg_path) in &seg_index {
                        let idx = index_path(seg_path);
                        if !idx.exists()
                            && let Err(e) = rebuild_index(seg_path, self.index_interval).await
                        {
                            tracing::warn!(
                                path = ?seg_path,
                                error = %e,
                                "failed to rebuild index"
                            );
                        }
                    }
                    index_entries.insert(key.clone(), seg_index);
                }

                // Open the latest segment for the writer cache
                if let Some(base_offset) = latest_offset {
                    let path = segment_path(&seg_dir, base_offset);
                    let (file, size) = open_segment(&path).await?;
                    let idx_file = open_index_file(&path).await?;
                    writer_entries.push((
                        key,
                        Arc::new(Mutex::new(ActiveSegment {
                            file,
                            size,
                            base_offset,
                            index_file: Some(idx_file),
                            records_since_last_index: 0,
                        })),
                    ));
                }
            }
        }

        if !writer_entries.is_empty() {
            let mut cache = self.writers.lock().await;
            for (key, segment) in writer_entries {
                cache.put(key, segment);
            }
        }

        if !index_entries.is_empty() {
            let mut idx = self.segment_index.write().await;
            *idx = index_entries;
        }

        // Initialize per-namespace disk byte counters
        if !disk_bytes.is_empty() {
            let mut ndb = self.namespace_disk_bytes.write().await;
            for (ns, bytes) in disk_bytes {
                ndb.insert(ns, AtomicU64::new(bytes));
            }
        }

        Ok(())
    }

    async fn append(
        &self,
        namespace: &NamespaceId,
        topic: &TopicName,
        msg: &PublishedMessage,
    ) -> Result<()> {
        let record = encode_record(msg)?;
        let key = TopicKey {
            namespace: namespace.0.clone(),
            topic: topic.0.clone(),
        };

        // Look up or create a writer under a single Mutex guard.
        // `LruCache::get` updates access order so recently-written topics
        // are kept open while idle ones are evicted.
        let writer = {
            let mut cache = self.writers.lock().await;
            match cache.get(&key).cloned() {
                Some(w) => w,
                None => {
                    // Topic not in cache — either first-ever write or the entry
                    // was evicted by LRU pressure. (Re-)open the segment file.
                    let seg_dir = self.segments_dir(namespace, topic);
                    let base_offset = find_latest_segment_offset(&seg_dir)
                        .await?
                        .unwrap_or(msg.offset);
                    let path = segment_path(&seg_dir, base_offset);
                    let new_segment = !path.exists();
                    let (file, size) = open_segment(&path).await?;
                    let idx_file = open_index_file(&path).await?;
                    // For brand-new segments, set counter to index_interval so the
                    // first record gets indexed. For reopened segments, start at 0.
                    let counter = if new_segment { self.index_interval } else { 0 };
                    let segment = Arc::new(Mutex::new(ActiveSegment {
                        file,
                        size,
                        base_offset,
                        index_file: Some(idx_file),
                        records_since_last_index: counter,
                    }));
                    // `push` returns the evicted (K, V) when at capacity;
                    // dropping the evicted Arc closes the fd once no in-flight
                    // operations hold a clone.
                    if let Some((evicted_key, _)) = cache.push(key.clone(), segment.clone()) {
                        tracing::debug!(
                            evicted_namespace = %evicted_key.namespace,
                            evicted_topic = %evicted_key.topic,
                            "LRU fd eviction: segment writer closed"
                        );
                        metrics::counter!("fafb_fd_evictions_total").increment(1);
                    }

                    // If this is a brand-new segment, add to the index
                    if new_segment {
                        let mut idx = self.segment_index.write().await;
                        idx.entry(key.clone())
                            .or_default()
                            .push((base_offset, path));
                    }

                    segment
                }
            }
        };

        let mut segment = writer.lock().await;

        // Check if we need to rotate to a new segment
        if segment.size > 0 && segment.size + record.len() as u64 > self.segment_max_bytes {
            segment.file.flush().await?;
            // Flush and close old index file
            if let Some(ref mut old_idx) = segment.index_file {
                old_idx.flush().await?;
            }
            let seg_dir = self.segments_dir(namespace, topic);
            let path = segment_path(&seg_dir, msg.offset);
            let (file, size) = open_segment(&path).await?;
            let idx_file = open_index_file(&path).await?;
            segment.file = file;
            segment.size = size;
            segment.base_offset = msg.offset;
            segment.index_file = Some(idx_file);
            // Set to index_interval so the first record in the new segment gets indexed
            segment.records_since_last_index = self.index_interval;

            // Update segment index with new segment
            {
                let mut idx = self.segment_index.write().await;
                idx.entry(key.clone()).or_default().push((msg.offset, path));
            }

            tracing::debug!(
                namespace = %namespace,
                topic = %topic,
                new_base_offset = msg.offset.0,
                "segment rotated"
            );

            // Segment count limits are now enforced by the background cleanup
            // task, which respects consumer offsets before deleting.
        }

        let record_position = segment.size;
        tracing::trace!(
            namespace = %namespace, topic = %topic,
            offset = %msg.offset, record_bytes = record.len(),
            "storage append"
        );
        segment.file.write_all(&record).await?;
        segment.size += record.len() as u64;

        // Write sparse index entry every index_interval records
        let mut index_bytes_written: u64 = 0;
        segment.records_since_last_index += 1;
        if segment.records_since_last_index >= self.index_interval {
            if let Some(ref mut idx_file) = segment.index_file {
                let mut entry = [0u8; INDEX_ENTRY_SIZE];
                entry[..8].copy_from_slice(&msg.offset.0.to_be_bytes());
                entry[8..16].copy_from_slice(&record_position.to_be_bytes());
                entry[16..24].copy_from_slice(&msg.envelope.timestamp.to_be_bytes());
                idx_file.write_all(&entry).await?;
                index_bytes_written = INDEX_ENTRY_SIZE as u64;
            }
            segment.records_since_last_index = 0;
        }

        // Update per-namespace disk bytes counter (segment record + index entry)
        {
            let total_bytes = record.len() as u64 + index_bytes_written;
            let map = self.namespace_disk_bytes.read().await;
            if let Some(counter) = map.get(&namespace.0) {
                counter.fetch_add(total_bytes, Ordering::Relaxed);
            } else {
                drop(map);
                let mut map = self.namespace_disk_bytes.write().await;
                map.entry(namespace.0.clone())
                    .or_insert_with(|| AtomicU64::new(0))
                    .fetch_add(total_bytes, Ordering::Relaxed);
            }
        }

        // Per-write fsync when configured for maximum durability
        if matches!(self.fsync_strategy, FsyncStrategy::EveryWrite) {
            segment.file.sync_all().await?;
        }

        Ok(())
    }

    async fn flush(&self, namespace: &NamespaceId, topic: &TopicName) -> Result<()> {
        let key = TopicKey {
            namespace: namespace.0.clone(),
            topic: topic.0.clone(),
        };
        let writer = {
            // Use `peek` so flush doesn't affect LRU eviction order.
            let cache = self.writers.lock().await;
            cache.peek(&key).cloned()
        };
        if let Some(writer) = writer {
            let mut segment = writer.lock().await;
            match self.fsync_strategy {
                FsyncStrategy::OsManaged => {
                    // Rely on OS page-cache writeback; skip explicit flush/sync.
                }
                FsyncStrategy::EveryWrite => {
                    // Already synced per-write; just flush the buffer.
                    segment.file.flush().await?;
                }
                FsyncStrategy::EveryBatch => {
                    segment.file.flush().await?;
                    segment.file.sync_all().await?;
                }
            }
            // Always flush the index file so offset_at_timestamp can read it
            // without requiring a full shutdown + re-init cycle.
            if let Some(ref mut idx_file) = segment.index_file {
                idx_file.flush().await?;
            }
        }
        Ok(())
    }

    async fn replay(
        &self,
        namespace: &NamespaceId,
        topic: &TopicName,
    ) -> Result<Vec<PublishedMessage>> {
        let seg_dir = self.segments_dir(namespace, topic);
        replay::replay_topic(&seg_dir).await
    }

    async fn read_at(
        &self,
        namespace: &NamespaceId,
        topic: &TopicName,
        offset: TopicOffset,
    ) -> Result<Option<PublishedMessage>> {
        let key = TopicKey {
            namespace: namespace.0.clone(),
            topic: topic.0.clone(),
        };

        // Try the cached segment index first (fast path: no directory listing)
        let seg_path = {
            let idx = self.segment_index.read().await;
            idx.get(&key)
                .and_then(|entries| find_segment_for_offset(entries, offset))
        };

        // Fall back to directory scan if cache miss (e.g., init() not called)
        let seg_path = match seg_path {
            Some(p) => p,
            None => {
                let seg_dir = self.segments_dir(namespace, topic);
                if !seg_dir.exists() {
                    return Ok(None);
                }
                let segments = replay::list_segments(&seg_dir).await?;
                let mut seg_offsets: Vec<(TopicOffset, PathBuf)> = Vec::new();
                for path in segments {
                    if let Some(stem) = path.file_stem().and_then(|s| s.to_str())
                        && let Ok(base) = stem.parse::<u64>()
                    {
                        seg_offsets.push((TopicOffset(base), path));
                    }
                }
                match find_segment_for_offset(&seg_offsets, offset) {
                    Some(p) => p,
                    None => return Ok(None),
                }
            }
        };

        // Scan the segment for the target offset
        let mut stream = replay::RecordStream::open(&seg_path).await?;

        // Try to use cached index for fast seek
        if let Some(seg_index) = self.load_index(&seg_path).await
            && let Some(byte_pos) = seg_index.find_position(offset)
        {
            stream.seek_to_position(byte_pos).await?;
        }

        while let Some((rec_offset, envelope)) = stream.next().await? {
            if rec_offset == offset {
                return Ok(Some(PublishedMessage {
                    envelope: Arc::new(envelope),
                    offset,
                }));
            }
            if rec_offset > offset {
                break;
            }
        }

        Ok(None)
    }

    async fn list_topics(&self) -> Result<Vec<(NamespaceId, TopicName)>> {
        let mut topics = Vec::new();

        if !self.data_dir.exists() {
            return Ok(topics);
        }

        let mut ns_dir = fs::read_dir(&self.data_dir).await?;
        while let Some(ns_entry) = ns_dir.next_entry().await? {
            if !ns_entry.file_type().await?.is_dir() {
                continue;
            }
            let ns_name = ns_entry.file_name().to_string_lossy().into_owned();

            let mut topic_dir = fs::read_dir(ns_entry.path()).await?;
            while let Some(topic_entry) = topic_dir.next_entry().await? {
                if !topic_entry.file_type().await?.is_dir() {
                    continue;
                }
                let topic_name = topic_entry.file_name().to_string_lossy().into_owned();
                let seg_dir = topic_entry.path().join("segments");

                // Only include topics that have at least one segment file
                if seg_dir.exists() {
                    let mut has_segments = false;
                    let mut seg_entries = fs::read_dir(&seg_dir).await?;
                    while let Some(seg_entry) = seg_entries.next_entry().await? {
                        if seg_entry.path().extension().and_then(|e| e.to_str()) == Some("seg") {
                            has_segments = true;
                            break;
                        }
                    }
                    if has_segments {
                        topics.push((NamespaceId(ns_name.clone()), TopicName(topic_name)));
                    }
                }
            }
        }

        Ok(topics)
    }

    fn segments_dir(&self, namespace: &NamespaceId, topic: &TopicName) -> PathBuf {
        self.data_dir
            .join(&namespace.0)
            .join(&topic.0)
            .join("segments")
    }

    async fn read_cold_batch(
        &self,
        namespace: &NamespaceId,
        topic: &TopicName,
        start_offset: TopicOffset,
        max_count: usize,
    ) -> Result<Vec<PublishedMessage>> {
        let key = TopicKey {
            namespace: namespace.0.clone(),
            topic: topic.0.clone(),
        };

        // Snapshot the segment index (read lock)
        let segments = {
            let idx = self.segment_index.read().await;
            match idx.get(&key) {
                Some(entries) => entries.clone(),
                None => return Ok(Vec::new()),
            }
        };

        // Open a SegmentCursor at start_offset
        let mut cursor = match SegmentCursor::open(segments, start_offset).await? {
            Some(c) => c,
            None => return Ok(Vec::new()),
        };

        // Skip directly to start_offset instead of filtering record-by-record
        cursor.skip_to(start_offset).await?;
        let mut result = Vec::with_capacity(max_count);
        while result.len() < max_count {
            match cursor.read_next().await? {
                Some(msg) => result.push(msg),
                None => break,
            }
        }

        Ok(result)
    }

    async fn offset_at_timestamp(
        &self,
        namespace: &NamespaceId,
        topic: &TopicName,
        timestamp_millis: u64,
    ) -> Result<Option<TopicOffset>> {
        let key = TopicKey {
            namespace: namespace.0.clone(),
            topic: topic.0.clone(),
        };

        // Get segment list from cache
        let segments = {
            let idx = self.segment_index.read().await;
            match idx.get(&key) {
                Some(entries) => entries.clone(),
                None => return Ok(None),
            }
        };

        if segments.is_empty() {
            return Ok(None);
        }

        // Iterate segments in order. Use cached index for fast seek
        // within each segment; scan forward for exact first match.
        for (i, (_base_offset, seg_path)) in segments.iter().enumerate() {
            // Skip optimization: if the next segment's first indexed record
            // is still before the target, this entire segment is too early.
            if i + 1 < segments.len()
                && let Some(next_idx) = self.load_index(&segments[i + 1].1).await
                && let Some(first) = next_idx.entries.first()
                && first.timestamp_millis < timestamp_millis
            {
                continue;
            }

            let mut stream = replay::RecordStream::open(seg_path).await?;

            // Try to use the sparse index for fast seek.
            // If no index or ts is before the first indexed entry, scan from start.
            if let Some(seg_index) = self.load_index(seg_path).await
                && let Some((_approx_offset, byte_pos)) =
                    seg_index.find_offset_at_timestamp(timestamp_millis)
            {
                stream.seek_to_position(byte_pos).await?;
            }

            // Scan forward for the first record with timestamp >= target
            while let Some((offset, envelope)) = stream.next().await? {
                if envelope.timestamp >= timestamp_millis {
                    return Ok(Some(offset));
                }
            }
            // All records in this segment are before the target — try next segment
        }

        Ok(None)
    }

    async fn delete_segments(
        &self,
        namespace: &NamespaceId,
        topic: &TopicName,
        before_offset: Option<TopicOffset>,
        max_segments: Option<usize>,
        safe_below: Option<TopicOffset>,
    ) -> Result<usize> {
        let seg_dir = self.segments_dir(namespace, topic);
        if !seg_dir.exists() {
            return Ok(0);
        }

        // List all segments sorted by base offset (ascending)
        let seg_files = replay::list_segments(&seg_dir).await?;
        if seg_files.is_empty() {
            return Ok(0);
        }

        // Parse base offsets from filenames
        let mut segments: Vec<(TopicOffset, PathBuf)> = Vec::new();
        for path in seg_files {
            if let Some(stem) = path.file_stem().and_then(|s| s.to_str())
                && let Ok(base) = stem.parse::<u64>()
            {
                segments.push((TopicOffset(base), path));
            }
        }

        // Build deletion candidate set
        let mut to_delete: Vec<(TopicOffset, PathBuf)> = Vec::new();

        // Criterion 1: segments with base_offset < before_offset
        if let Some(threshold) = before_offset {
            for (base, path) in &segments {
                if *base < threshold {
                    to_delete.push((*base, path.clone()));
                }
            }
        }

        // Criterion 2: excess segments beyond max_segments (oldest first)
        if let Some(max) = max_segments
            && segments.len() > max
        {
            let excess = segments.len() - max;
            for (base, path) in segments.iter().take(excess) {
                if !to_delete.iter().any(|(b, _)| b == base) {
                    to_delete.push((*base, path.clone()));
                }
            }
        }

        // Apply safe_below floor: never delete segments with base >= floor
        if let Some(floor) = safe_below {
            to_delete.retain(|(base, _)| *base < floor);
        }

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

        // Delete the segment files and their index files
        let mut deleted = 0;
        let mut deleted_offsets = std::collections::HashSet::new();
        let mut deleted_idx_paths = Vec::new();
        let mut freed_bytes: u64 = 0;
        for (base, path) in &to_delete {
            // Stat the file before deleting to track freed bytes
            if let Ok(meta) = fs::metadata(path).await {
                freed_bytes += meta.len();
            }
            fs::remove_file(path).await?;
            let idx = index_path(path);
            // Stat .idx before deleting to track freed bytes
            if let Ok(meta) = fs::metadata(&idx).await {
                freed_bytes += meta.len();
            }
            let _ = fs::remove_file(&idx).await; // ignore if missing
            deleted_idx_paths.push(idx);
            deleted_offsets.insert(*base);
            deleted += 1;
        }

        // Update segment index cache (in-place retain)
        let key = TopicKey {
            namespace: namespace.0.clone(),
            topic: topic.0.clone(),
        };
        let mut idx = self.segment_index.write().await;
        if let Some(entries) = idx.get_mut(&key) {
            entries.retain(|(base, _)| !deleted_offsets.contains(base));
        }

        // Evict deleted indexes from cache
        {
            let mut ic = self.index_cache.write().await;
            for idx_path in &deleted_idx_paths {
                ic.remove(idx_path);
            }
        }

        // Subtract freed bytes from namespace disk counter
        if freed_bytes > 0 {
            let map = self.namespace_disk_bytes.read().await;
            if let Some(counter) = map.get(&namespace.0) {
                counter
                    .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
                        Some(v.saturating_sub(freed_bytes))
                    })
                    .ok();
            }
        }

        if deleted > 0 {
            tracing::debug!(
                namespace = %namespace,
                topic = %topic,
                deleted_segments = deleted,
                freed_bytes,
                "segments deleted"
            );
        }

        Ok(deleted)
    }

    async fn compact_segments(
        &self,
        namespace: &NamespaceId,
        topic: &TopicName,
        merge_threshold_bytes: u64,
        safe_below: Option<TopicOffset>,
    ) -> Result<usize> {
        use crate::retention::scan_segments;

        let seg_dir = self.segments_dir(namespace, topic);
        if !seg_dir.exists() {
            return Ok(0);
        }

        // (#4) Clean up orphan .tmp files from interrupted compactions
        if let Ok(mut read_dir) = fs::read_dir(&seg_dir).await {
            while let Ok(Some(entry)) = read_dir.next_entry().await {
                let path = entry.path();
                if path.extension().and_then(|e| e.to_str()) == Some("tmp") {
                    let _ = fs::remove_file(&path).await;
                }
            }
        }

        // (#5) Reuse scan_segments from retention module
        let segments = scan_segments(&seg_dir).await;
        if segments.len() < 2 {
            return Ok(0);
        }

        // (#3) Determine active writer segment from the writers cache,
        // falling back to the last segment if the writer was evicted from LRU
        let active_base = {
            let key = TopicKey {
                namespace: namespace.0.clone(),
                topic: topic.0.clone(),
            };
            let writer = {
                let cache = self.writers.lock().await;
                cache.peek(&key).cloned()
            };
            match writer {
                Some(w) => {
                    let seg = w.lock().await;
                    Some(seg.base_offset)
                }
                None => segments.last().map(|s| s.base_offset),
            }
        };

        // Filter: exclude active writer segment and segments at/above safe_below
        let candidates: Vec<_> = segments
            .iter()
            .filter(|s| {
                if Some(s.base_offset) == active_base {
                    return false;
                }
                if let Some(floor) = safe_below
                    && s.base_offset >= floor
                {
                    return false;
                }
                true
            })
            .collect();

        if candidates.len() < 2 {
            return Ok(0);
        }

        // (#8) Find runs of consecutive small segments, capped at segment_max_bytes
        let mut runs: Vec<Vec<&crate::retention::SegmentInfo>> = Vec::new();
        let mut current_run: Vec<&crate::retention::SegmentInfo> = Vec::new();
        let mut current_run_bytes: u64 = 0;

        for seg in &candidates {
            if seg.size_bytes < merge_threshold_bytes {
                if !current_run.is_empty()
                    && current_run_bytes + seg.size_bytes > self.segment_max_bytes
                {
                    // Adding this segment would exceed max — flush current run
                    if current_run.len() >= 2 {
                        runs.push(std::mem::take(&mut current_run));
                    } else {
                        current_run.clear();
                    }
                    current_run_bytes = 0;
                }
                current_run.push(seg);
                current_run_bytes += seg.size_bytes;
            } else if current_run.len() >= 2 {
                runs.push(std::mem::take(&mut current_run));
                current_run_bytes = 0;
            } else {
                current_run.clear();
                current_run_bytes = 0;
            }
        }
        if current_run.len() >= 2 {
            runs.push(current_run);
        }

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

        let key = TopicKey {
            namespace: namespace.0.clone(),
            topic: topic.0.clone(),
        };
        let mut merge_count = 0;

        for run in &runs {
            let merged_base = run[0].base_offset;
            let merged_seg_path = segment_path(&seg_dir, merged_base);
            let tmp_seg_path = merged_seg_path.with_extension("seg.tmp");
            let merged_idx_path = index_path(&merged_seg_path);
            let tmp_idx_path = merged_idx_path.with_extension("idx.tmp");

            // (#1) Per-run error isolation — failures are logged and skipped
            let merge_result: Result<()> = async {
                // Sum original sizes (seg + idx) for disk byte accounting
                let mut original_total_bytes: u64 = 0;
                for seg in run {
                    original_total_bytes += seg.size_bytes;
                    let idx = index_path(&seg.path);
                    if let Ok(meta) = fs::metadata(&idx).await {
                        original_total_bytes += meta.len();
                    }
                }

                // (#2) Write merged segment and index inline (no rebuild_index re-scan)
                {
                    let mut out = OpenOptions::new()
                        .create(true)
                        .write(true)
                        .truncate(true)
                        .open(&tmp_seg_path)
                        .await?;

                    let mut idx_out = OpenOptions::new()
                        .create(true)
                        .write(true)
                        .truncate(true)
                        .open(&tmp_idx_path)
                        .await?;

                    let mut byte_pos: u64 = 0;
                    let mut records_since_last: u32 = self.index_interval; // first record indexed

                    for seg in run {
                        let mut stream = replay::RecordStream::open(&seg.path).await?;
                        while let Some((offset, envelope)) = stream.next().await? {
                            let msg = PublishedMessage {
                                envelope: Arc::new(envelope),
                                offset,
                            };
                            let record = encode_record(&msg)?;

                            // Write index entry before the record data
                            records_since_last += 1;
                            if records_since_last >= self.index_interval {
                                let mut entry = [0u8; INDEX_ENTRY_SIZE];
                                entry[..8].copy_from_slice(&offset.0.to_be_bytes());
                                entry[8..16].copy_from_slice(&byte_pos.to_be_bytes());
                                entry[16..24]
                                    .copy_from_slice(&msg.envelope.timestamp.to_be_bytes());
                                idx_out.write_all(&entry).await?;
                                records_since_last = 0;
                            }

                            out.write_all(&record).await?;
                            byte_pos += record.len() as u64;
                        }
                    }
                    out.flush().await?;
                    out.sync_all().await?;
                    idx_out.flush().await?;
                    idx_out.sync_all().await?;
                }

                // Atomic renames: tmp → final
                fs::rename(&tmp_seg_path, &merged_seg_path).await?;
                fs::rename(&tmp_idx_path, &merged_idx_path).await?;

                // Delete remaining original segment files (first was replaced by rename)
                let mut deleted_idx_paths: Vec<PathBuf> = Vec::new();
                for seg in run.iter().skip(1) {
                    let _ = fs::remove_file(&seg.path).await;
                    let idx = index_path(&seg.path);
                    let _ = fs::remove_file(&idx).await;
                    deleted_idx_paths.push(idx);
                }
                // Old .idx for first segment was replaced by idx rename
                deleted_idx_paths.push(index_path(&run[0].path));

                // Compute merged file sizes for disk byte adjustment
                let mut merged_total_bytes: u64 = 0;
                if let Ok(meta) = fs::metadata(&merged_seg_path).await {
                    merged_total_bytes += meta.len();
                }
                if let Ok(meta) = fs::metadata(&merged_idx_path).await {
                    merged_total_bytes += meta.len();
                }

                // (#7) Batch cache updates — acquire both locks in consistent order
                {
                    let mut seg_idx = self.segment_index.write().await;
                    let mut ic = self.index_cache.write().await;

                    if let Some(entries) = seg_idx.get_mut(&key) {
                        let merged_offsets: std::collections::HashSet<TopicOffset> =
                            run.iter().map(|s| s.base_offset).collect();
                        entries.retain(|(base, _)| !merged_offsets.contains(base));
                        let insert_pos = entries
                            .binary_search_by_key(&merged_base, |(base, _)| *base)
                            .unwrap_or_else(|i| i);
                        entries.insert(insert_pos, (merged_base, merged_seg_path.clone()));
                    }

                    for idx_path in &deleted_idx_paths {
                        ic.remove(idx_path);
                    }
                }

                // Adjust namespace disk bytes
                if original_total_bytes != merged_total_bytes {
                    let map = self.namespace_disk_bytes.read().await;
                    if let Some(counter) = map.get(&namespace.0) {
                        if merged_total_bytes < original_total_bytes {
                            let freed = original_total_bytes - merged_total_bytes;
                            counter
                                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
                                    Some(v.saturating_sub(freed))
                                })
                                .ok();
                        } else {
                            let added = merged_total_bytes - original_total_bytes;
                            counter.fetch_add(added, Ordering::Relaxed);
                        }
                    }
                }

                Ok(())
            }
            .await;

            match merge_result {
                Ok(()) => merge_count += 1,
                Err(e) => {
                    // (#1) Clean up partial temp files and continue
                    let _ = fs::remove_file(&tmp_seg_path).await;
                    let _ = fs::remove_file(&tmp_idx_path).await;
                    tracing::warn!(
                        namespace = %namespace,
                        topic = %topic,
                        base_offset = merged_base.0,
                        error = %e,
                        "compaction: merge failed, skipping run"
                    );
                }
            }
        }

        Ok(merge_count)
    }

    async fn namespace_disk_bytes(&self, namespace: &NamespaceId) -> u64 {
        let map = self.namespace_disk_bytes.read().await;
        map.get(&namespace.0)
            .map(|v| v.load(Ordering::Relaxed))
            .unwrap_or(0)
    }

    async fn flush_all(&self) -> Result<()> {
        let writers: Vec<(TopicKey, Arc<Mutex<ActiveSegment>>)> = {
            let cache = self.writers.lock().await;
            cache
                .iter()
                .map(|(k, v)| (k.clone(), Arc::clone(v)))
                .collect()
        };
        for (key, _writer) in &writers {
            let ns = NamespaceId(key.namespace.clone());
            let topic = TopicName(key.topic.clone());
            self.flush(&ns, &topic).await?;
        }
        Ok(())
    }

    async fn close_topic_writer(&self, namespace: &NamespaceId, topic: &TopicName) -> Result<()> {
        let key = TopicKey {
            namespace: namespace.0.clone(),
            topic: topic.0.clone(),
        };
        let writer = {
            let mut cache = self.writers.lock().await;
            cache.pop(&key)
        };
        if let Some(writer) = writer {
            let mut segment = writer.lock().await;
            segment.file.flush().await?;
            segment.file.sync_all().await?;
            if let Some(ref mut idx_file) = segment.index_file {
                idx_file.flush().await?;
            }
        }
        Ok(())
    }

    async fn delete_topic_dir(&self, namespace: &NamespaceId, topic: &TopicName) -> Result<()> {
        let dir = self.segments_dir(namespace, topic);
        match tokio::fs::remove_dir_all(&dir).await {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(e) => Err(e.into()),
        }
    }

    async fn shutdown(&self) -> Result<()> {
        let drained: Vec<(TopicKey, Arc<Mutex<ActiveSegment>>)> = {
            let mut cache = self.writers.lock().await;
            let mut entries = Vec::with_capacity(cache.len());
            while let Some((k, v)) = cache.pop_lru() {
                entries.push((k, v));
            }
            entries
        };
        for (_, writer) in drained {
            let mut segment = writer.lock().await;
            segment.file.flush().await?;
            segment.file.sync_all().await?;
        }
        Ok(())
    }
}

/// Holds an open file handle for sequential cold reads across segments.
///
/// Created when a consumer first hits the cold path and advanced with
/// [`read_next`](Self::read_next). Automatically crosses segment boundaries.
/// Avoids the cost of segment lookup + file open + linear scan per message.
pub struct SegmentCursor {
    stream: replay::RecordStream,
    /// Snapshot of the segment index at cursor creation time.
    segments: Vec<(TopicOffset, PathBuf)>,
    /// Index into `segments` for the currently-open segment.
    current_idx: usize,
    /// A record read-ahead by `skip_to` that should be returned on the next `read_next`.
    pending: Option<PublishedMessage>,
}

impl SegmentCursor {
    /// Create a cursor positioned at (or just before) `start_offset`.
    ///
    /// Returns `None` if no segments contain the target offset.
    pub async fn open(
        segments: Vec<(TopicOffset, PathBuf)>,
        start_offset: TopicOffset,
    ) -> Result<Option<Self>> {
        if segments.is_empty() {
            return Ok(None);
        }

        // Find the segment containing start_offset
        let idx = match segments.binary_search_by_key(&start_offset, |(base, _)| *base) {
            Ok(i) => i,
            Err(0) => return Ok(None),
            Err(i) => i - 1,
        };

        let (_, ref path) = segments[idx];
        let stream = replay::RecordStream::open(path).await?;

        Ok(Some(Self {
            stream,
            segments,
            current_idx: idx,
            pending: None,
        }))
    }

    /// Advance past all records with offset < target.
    /// The next `read_next()` returns the first record at or after `target`.
    pub async fn skip_to(&mut self, target: TopicOffset) -> Result<()> {
        // Drain any pending record first
        if let Some(ref msg) = self.pending {
            if msg.offset >= target {
                return Ok(()); // pending record already past target
            }
            self.pending = None;
        }

        // Try to use index for fast seek within the current segment
        let idx_path = index_path(&self.segments[self.current_idx].1);
        if let Ok(index) = SegmentIndex::load(&idx_path).await
            && let Some(byte_pos) = index.find_position(target)
        {
            self.stream.seek_to_position(byte_pos).await?;
        }

        // Scan forward from indexed position (at most INDEX_INTERVAL records
        // if an index entry was found, or the full segment otherwise)
        loop {
            match self.stream.next().await? {
                Some((offset, _)) if offset < target => continue,
                Some((offset, envelope)) => {
                    self.pending = Some(PublishedMessage {
                        envelope: Arc::new(envelope),
                        offset,
                    });
                    return Ok(());
                }
                None => {
                    // Segment exhausted — try next segment
                    self.current_idx += 1;
                    if self.current_idx >= self.segments.len() {
                        return Ok(());
                    }
                    let (_, ref path) = self.segments[self.current_idx];
                    self.stream = replay::RecordStream::open(path).await?;
                }
            }
        }
    }

    /// Read the next record. Automatically crosses segment boundaries.
    /// Returns `None` when all segments are exhausted.
    pub async fn read_next(&mut self) -> Result<Option<PublishedMessage>> {
        if let Some(msg) = self.pending.take() {
            return Ok(Some(msg));
        }
        loop {
            match self.stream.next().await? {
                Some((offset, envelope)) => {
                    return Ok(Some(PublishedMessage {
                        envelope: Arc::new(envelope),
                        offset,
                    }));
                }
                None => {
                    // Current segment exhausted — try the next one
                    self.current_idx += 1;
                    if self.current_idx >= self.segments.len() {
                        return Ok(None);
                    }
                    let (_, ref path) = self.segments[self.current_idx];
                    self.stream = replay::RecordStream::open(path).await?;
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    #[tokio::test]
    async fn init_creates_data_dir() {
        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("storage-data");
        let storage = FsStorage::new(&data_dir);
        assert!(!data_dir.exists());
        storage.init().await.unwrap();
        assert!(data_dir.exists());
    }

    #[tokio::test]
    async fn init_idempotent() {
        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("storage-data");
        let storage = FsStorage::new(&data_dir);
        storage.init().await.unwrap();
        // Second call should not fail
        storage.init().await.unwrap();
        assert!(data_dir.exists());
    }

    #[tokio::test]
    async fn init_pre_populates_writers() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");

        // First instance: write some messages and shutdown
        let storage = FsStorage::new(&data_dir);
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        let msg = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(1),
                ns.clone(),
                topic.clone(),
                b"hello".to_vec(),
                now_millis(),
            )),
            offset: TopicOffset(0),
        };
        storage.append(&ns, &topic, &msg).await.unwrap();
        storage.shutdown().await.unwrap();

        // Second instance: init should pre-populate writers
        let storage2 = FsStorage::new(&data_dir);
        storage2.init().await.unwrap();

        let key = TopicKey {
            namespace: "ns".into(),
            topic: "t".into(),
        };
        let has_writer = storage2.writers.lock().await.contains(&key);
        assert!(
            has_writer,
            "init should pre-populate writers from existing segments"
        );
    }

    #[tokio::test]
    async fn append_creates_segments_dir() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("test-ns".into());
        let topic = TopicName("test-topic".into());
        let msg = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(1),
                ns.clone(),
                topic.clone(),
                b"hello".to_vec(),
                now_millis(),
            )),
            offset: TopicOffset(0),
        };

        storage.append(&ns, &topic, &msg).await.unwrap();

        let seg_dir = storage.segments_dir(&ns, &topic);
        assert!(seg_dir.exists());
    }

    #[tokio::test]
    async fn flush_no_op_for_unknown_topic() {
        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("unknown".into());
        let topic = TopicName("unknown".into());
        // Should not error
        storage.flush(&ns, &topic).await.unwrap();
    }

    #[tokio::test]
    async fn shutdown_with_no_writers() {
        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();
        storage.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn list_topics_empty_data_dir() {
        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();
        let topics = storage.list_topics().await.unwrap();
        assert!(topics.is_empty());
    }

    #[tokio::test]
    async fn list_topics_returns_persisted_topics() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns1".into());
        let t1 = TopicName("topic-a".into());
        let t2 = TopicName("topic-b".into());

        for (topic, offset) in [(&t1, 0u64), (&t2, 0u64)] {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(offset + 1),
                    ns.clone(),
                    topic.clone(),
                    b"data".to_vec(),
                    now_millis(),
                )),
                offset: TopicOffset(offset),
            };
            storage.append(&ns, topic, &msg).await.unwrap();
        }

        let mut topics = storage.list_topics().await.unwrap();
        topics.sort_by(|a, b| a.1.0.cmp(&b.1.0));

        assert_eq!(topics.len(), 2);
        assert_eq!(
            topics[0],
            (NamespaceId("ns1".into()), TopicName("topic-a".into()))
        );
        assert_eq!(
            topics[1],
            (NamespaceId("ns1".into()), TopicName("topic-b".into()))
        );
    }

    #[test]
    fn segment_path_formatting() {
        let dir = Path::new("/tmp/segments");
        let path = segment_path(dir, TopicOffset(0));
        assert_eq!(
            path,
            PathBuf::from("/tmp/segments/00000000000000000000.seg")
        );

        let path = segment_path(dir, TopicOffset(42));
        assert_eq!(
            path,
            PathBuf::from("/tmp/segments/00000000000000000042.seg")
        );

        let path = segment_path(dir, TopicOffset(u64::MAX));
        assert_eq!(
            path,
            PathBuf::from("/tmp/segments/18446744073709551615.seg")
        );
    }

    #[tokio::test]
    async fn from_config_uses_config_values() {
        let config = StorageConfig {
            data_dir: "/tmp/test-data".to_owned(),
            fsync_strategy: FsyncStrategy::OsManaged,
            batch_flush_interval_ms: 200,
            max_open_fds: 512,
            max_segments_per_topic: 16,
            retention_check_interval_secs: 60,
            ..Default::default()
        };
        let storage = FsStorage::from_config(&config);
        assert_eq!(storage.data_dir, PathBuf::from("/tmp/test-data"));
        assert_eq!(storage.fsync_strategy, FsyncStrategy::OsManaged);
        assert_eq!(storage.max_segments_per_topic, 16);
        assert_eq!(storage.max_open_fds, 512);
    }

    #[tokio::test]
    async fn cleanup_segments_enforces_limit() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");

        // Small segments (256 bytes)
        let storage = FsStorage::with_segment_max_bytes(&data_dir, 256);
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Write enough data to force multiple segment rotations (> 3 segments)
        for i in 0..100u64 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    vec![0xAB; 30],
                    now_millis(),
                )),
                offset: TopicOffset(i),
            };
            storage.append(&ns, &topic, &msg).await.unwrap();
        }

        let seg_dir = storage.segments_dir(&ns, &topic);

        // Count segment files before cleanup
        let before = count_seg_files(&seg_dir).await;
        assert!(
            before > 3,
            "should have more than 3 segments before cleanup"
        );

        // Cleanup with max_segments=3, no consumer floor
        let deleted = storage
            .delete_segments(&ns, &topic, None, Some(3), None)
            .await
            .unwrap();
        assert!(deleted > 0, "should have deleted some segments");

        let after = count_seg_files(&seg_dir).await;
        assert!(
            after <= 3,
            "expected at most 3 segments after cleanup, found {after}"
        );
    }

    #[tokio::test]
    async fn cleanup_segments_respects_safe_offset() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");

        let storage = FsStorage::with_segment_max_bytes(&data_dir, 256);
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        for i in 0..100u64 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    vec![0xAB; 30],
                    now_millis(),
                )),
                offset: TopicOffset(i),
            };
            storage.append(&ns, &topic, &msg).await.unwrap();
        }

        let seg_dir = storage.segments_dir(&ns, &topic);
        let before = count_seg_files(&seg_dir).await;

        // Cleanup with max_segments=1 but safe_below at a high offset
        // — should only delete segments whose base_offset < safe_below
        let deleted = storage
            .delete_segments(&ns, &topic, None, Some(1), Some(TopicOffset(5)))
            .await
            .unwrap();

        // Some segments may be deleted (those with base < 5), but many are protected
        let after = count_seg_files(&seg_dir).await;
        assert!(
            after > 1,
            "safe_below should prevent deleting most segments (before={before}, after={after}, deleted={deleted})"
        );
    }

    // -- delete_segments tests --

    #[tokio::test]
    async fn delete_segments_removes_files_before_offset() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");

        // Small segments to force rotation
        let storage = FsStorage::with_segment_max_bytes(&data_dir, 128);
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Write enough to create multiple segments
        for i in 0..30u64 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    vec![0xAB; 30],
                    now_millis(),
                )),
                offset: TopicOffset(i),
            };
            storage.append(&ns, &topic, &msg).await.unwrap();
        }

        // Count initial segments
        let seg_dir = storage.segments_dir(&ns, &topic);
        let initial_count = count_seg_files(&seg_dir).await;
        assert!(
            initial_count >= 3,
            "need at least 3 segments, got {initial_count}"
        );

        // Delete segments before offset 10
        let deleted = storage
            .delete_segments(&ns, &topic, Some(TopicOffset(10)), None, None)
            .await
            .unwrap();
        assert!(deleted > 0, "should have deleted at least one segment");

        let after_count = count_seg_files(&seg_dir).await;
        assert_eq!(after_count, initial_count - deleted);
    }

    #[tokio::test]
    async fn delete_segments_returns_zero_when_none_match() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Write one message (creates segment at offset 0)
        let msg = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(1),
                ns.clone(),
                topic.clone(),
                b"data".to_vec(),
                now_millis(),
            )),
            offset: TopicOffset(0),
        };
        storage.append(&ns, &topic, &msg).await.unwrap();

        // Try to delete before offset 0 — nothing should match
        let deleted = storage
            .delete_segments(&ns, &topic, Some(TopicOffset(0)), None, None)
            .await
            .unwrap();
        assert_eq!(deleted, 0);
    }

    #[tokio::test]
    async fn delete_segments_empty_directory() {
        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // No segments exist
        let deleted = storage
            .delete_segments(&ns, &topic, Some(TopicOffset(100)), None, None)
            .await
            .unwrap();
        assert_eq!(deleted, 0);
    }

    async fn count_seg_files(dir: &Path) -> usize {
        let mut count = 0;
        if let Ok(mut read_dir) = fs::read_dir(dir).await {
            while let Ok(Some(entry)) = read_dir.next_entry().await {
                if entry.path().extension().and_then(|e| e.to_str()) == Some("seg") {
                    count += 1;
                }
            }
        }
        count
    }

    /// Helper to build an FsStorage with a specific max_open_fds limit.
    fn fs_storage_with_fd_limit(data_dir: impl Into<PathBuf>, max_fds: usize) -> FsStorage {
        let max_fds_nz = NonZeroUsize::new(max_fds).expect("max_fds must be > 0");
        FsStorage {
            data_dir: data_dir.into(),
            segment_max_bytes: DEFAULT_SEGMENT_MAX_BYTES,
            fsync_strategy: FsyncStrategy::EveryBatch,
            max_segments_per_topic: 64,
            max_open_fds: max_fds,
            index_interval: DEFAULT_INDEX_INTERVAL,
            writers: Mutex::new(LruCache::new(max_fds_nz)),
            segment_index: RwLock::new(HashMap::new()),
            index_cache: RwLock::new(HashMap::new()),
            namespace_disk_bytes: RwLock::new(HashMap::new()),
        }
    }

    #[tokio::test]
    async fn lru_evicts_least_recently_used_topic() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");

        // Allow only 2 open fds at a time
        let storage = fs_storage_with_fd_limit(&data_dir, 2);
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topics: Vec<TopicName> = (0..3).map(|i| TopicName(format!("t{i}"))).collect();

        // Write to topic 0 and topic 1 — both fit in the cache
        for (i, topic) in topics.iter().enumerate().take(2) {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i as u64 + 1),
                    ns.clone(),
                    topic.clone(),
                    b"data".to_vec(),
                    now_millis(),
                )),
                offset: TopicOffset(i as u64),
            };
            storage.append(&ns, topic, &msg).await.unwrap();
        }

        {
            let cache = storage.writers.lock().await;
            assert_eq!(cache.len(), 2, "both topics should be cached");
        }

        // Write to topic 2 — should evict topic 0 (the LRU entry)
        let msg = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(2),
                ns.clone(),
                topics[2].clone(),
                b"data".to_vec(),
                now_millis(),
            )),
            offset: TopicOffset(2),
        };
        storage.append(&ns, &topics[2], &msg).await.unwrap();

        {
            let cache = storage.writers.lock().await;
            assert_eq!(cache.len(), 2, "cache should still be at capacity");
            let key0 = TopicKey {
                namespace: "ns".into(),
                topic: "t0".into(),
            };
            assert!(!cache.contains(&key0), "topic 0 should have been evicted");
            let key2 = TopicKey {
                namespace: "ns".into(),
                topic: "t2".into(),
            };
            assert!(cache.contains(&key2), "topic 2 should be cached");
        }

        // Writing to the evicted topic 0 again should reopen it
        let msg = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(3),
                ns.clone(),
                topics[0].clone(),
                b"more data".to_vec(),
                now_millis(),
            )),
            offset: TopicOffset(3),
        };
        storage.append(&ns, &topics[0], &msg).await.unwrap();

        {
            let cache = storage.writers.lock().await;
            let key0 = TopicKey {
                namespace: "ns".into(),
                topic: "t0".into(),
            };
            assert!(
                cache.contains(&key0),
                "topic 0 should be back in cache after reopen"
            );
        }

        // Verify data integrity: replay all three topics
        for (i, topic) in topics.iter().enumerate() {
            let msgs = storage.replay(&ns, topic).await.unwrap();
            assert!(!msgs.is_empty(), "topic {i} should have persisted messages");
        }

        storage.shutdown().await.unwrap();
    }

    #[test]
    fn find_segment_for_offset_binary_search() {
        let entries = vec![
            (TopicOffset(0), PathBuf::from("0.seg")),
            (TopicOffset(100), PathBuf::from("100.seg")),
            (TopicOffset(200), PathBuf::from("200.seg")),
        ];

        // Exact match
        assert_eq!(
            find_segment_for_offset(&entries, TopicOffset(100)).unwrap(),
            PathBuf::from("100.seg")
        );

        // Between segments — should return the preceding segment
        assert_eq!(
            find_segment_for_offset(&entries, TopicOffset(150)).unwrap(),
            PathBuf::from("100.seg")
        );

        // Before all segments
        assert!(
            find_segment_for_offset(&entries, TopicOffset(0))
                .unwrap()
                .to_str()
                .unwrap()
                .contains("0.seg")
        );

        // After all segments
        assert_eq!(
            find_segment_for_offset(&entries, TopicOffset(999)).unwrap(),
            PathBuf::from("200.seg")
        );

        // Empty entries
        assert!(find_segment_for_offset(&[], TopicOffset(0)).is_none());
    }

    #[tokio::test]
    async fn segment_cursor_sequential_read() {
        use crate::storage::replay::tests::{encode_record, make_msg};

        let tmp = TempDir::new().unwrap();
        let seg_dir = tmp.path();

        // Create 3 segments with 5 messages each
        for seg in 0..3u64 {
            let base = seg * 5;
            let mut data = Vec::new();
            for i in base..base + 5 {
                data.extend_from_slice(&encode_record(&make_msg(i)));
            }
            let path = seg_dir.join(format!("{:020}.seg", base));
            fs::write(&path, &data).await.unwrap();
        }

        let segments = vec![
            (TopicOffset(0), seg_dir.join("00000000000000000000.seg")),
            (TopicOffset(5), seg_dir.join("00000000000000000005.seg")),
            (TopicOffset(10), seg_dir.join("00000000000000000010.seg")),
        ];

        let mut cursor = SegmentCursor::open(segments, TopicOffset(0))
            .await
            .unwrap()
            .unwrap();

        let mut count = 0;
        while let Some(msg) = cursor.read_next().await.unwrap() {
            assert_eq!(msg.offset, TopicOffset(count));
            count += 1;
        }
        assert_eq!(count, 15);
    }

    #[tokio::test]
    async fn segment_cursor_skip_to() {
        use crate::storage::replay::tests::{encode_record, make_msg};

        let tmp = TempDir::new().unwrap();
        let seg_dir = tmp.path();

        // Create 2 segments with 5 messages each (offsets 0-4 and 5-9)
        for seg in 0..2u64 {
            let base = seg * 5;
            let mut data = Vec::new();
            for i in base..base + 5 {
                data.extend_from_slice(&encode_record(&make_msg(i)));
            }
            let path = seg_dir.join(format!("{:020}.seg", base));
            fs::write(&path, &data).await.unwrap();
        }

        let segments = vec![
            (TopicOffset(0), seg_dir.join("00000000000000000000.seg")),
            (TopicOffset(5), seg_dir.join("00000000000000000005.seg")),
        ];

        // Open cursor at offset 0
        let mut cursor = SegmentCursor::open(segments, TopicOffset(0))
            .await
            .unwrap()
            .unwrap();

        // Skip to offset 7 (in the second segment)
        cursor.skip_to(TopicOffset(7)).await.unwrap();

        // read_next should return offset 7
        let msg = cursor.read_next().await.unwrap().unwrap();
        assert_eq!(msg.offset, TopicOffset(7));

        // read_next should return offset 8
        let msg = cursor.read_next().await.unwrap().unwrap();
        assert_eq!(msg.offset, TopicOffset(8));
    }

    #[tokio::test]
    async fn segment_cursor_crosses_boundary() {
        use crate::storage::replay::tests::{encode_record, make_msg};

        let tmp = TempDir::new().unwrap();
        let seg_dir = tmp.path();

        // Segment 1: offsets 0-2
        let mut data1 = Vec::new();
        for i in 0..3 {
            data1.extend_from_slice(&encode_record(&make_msg(i)));
        }
        fs::write(seg_dir.join("00000000000000000000.seg"), &data1)
            .await
            .unwrap();

        // Segment 2: offsets 3-5
        let mut data2 = Vec::new();
        for i in 3..6 {
            data2.extend_from_slice(&encode_record(&make_msg(i)));
        }
        fs::write(seg_dir.join("00000000000000000003.seg"), &data2)
            .await
            .unwrap();

        let segments = vec![
            (TopicOffset(0), seg_dir.join("00000000000000000000.seg")),
            (TopicOffset(3), seg_dir.join("00000000000000000003.seg")),
        ];

        // Start from offset 2 (last message of first segment)
        let mut cursor = SegmentCursor::open(segments, TopicOffset(0))
            .await
            .unwrap()
            .unwrap();

        // Read all 6 messages — cursor should cross from segment 1 to segment 2
        let mut offsets = Vec::new();
        while let Some(msg) = cursor.read_next().await.unwrap() {
            offsets.push(msg.offset.0);
        }
        assert_eq!(offsets, vec![0, 1, 2, 3, 4, 5]);
    }

    // ---- Index tests ----

    #[tokio::test]
    async fn index_entry_roundtrip() {
        let tmp = TempDir::new().unwrap();
        let idx_path = tmp.path().join("test.idx");

        // Write 10 entries
        let mut file = OpenOptions::new()
            .create(true)
            .write(true)
            .open(&idx_path)
            .await
            .unwrap();
        for i in 0u64..10 {
            let mut entry = [0u8; INDEX_ENTRY_SIZE];
            entry[..8].copy_from_slice(&(i * 100).to_be_bytes());
            entry[8..16].copy_from_slice(&(i * 1000).to_be_bytes());
            entry[16..24].copy_from_slice(&(i * 10_000).to_be_bytes());
            file.write_all(&entry).await.unwrap();
        }
        file.flush().await.unwrap();

        let index = SegmentIndex::load(&idx_path).await.unwrap();
        assert_eq!(index.entries.len(), 10);
        for i in 0u64..10 {
            assert_eq!(index.entries[i as usize].offset, TopicOffset(i * 100));
            assert_eq!(index.entries[i as usize].byte_position, i * 1000);
            assert_eq!(index.entries[i as usize].timestamp_millis, i * 10_000);
        }
    }

    #[tokio::test]
    async fn index_find_position_exact() {
        let tmp = TempDir::new().unwrap();
        let idx_path = tmp.path().join("test.idx");

        let mut file = OpenOptions::new()
            .create(true)
            .write(true)
            .open(&idx_path)
            .await
            .unwrap();
        for i in [0u64, 256, 512, 768] {
            let mut entry = [0u8; INDEX_ENTRY_SIZE];
            entry[..8].copy_from_slice(&i.to_be_bytes());
            entry[8..16].copy_from_slice(&(i * 50).to_be_bytes());
            entry[16..24].copy_from_slice(&(1000 + i).to_be_bytes());
            file.write_all(&entry).await.unwrap();
        }
        file.flush().await.unwrap();

        let index = SegmentIndex::load(&idx_path).await.unwrap();
        assert_eq!(index.find_position(TopicOffset(256)), Some(256 * 50));
        assert_eq!(index.find_position(TopicOffset(512)), Some(512 * 50));
    }

    #[tokio::test]
    async fn index_find_position_between() {
        let tmp = TempDir::new().unwrap();
        let idx_path = tmp.path().join("test.idx");

        let mut file = OpenOptions::new()
            .create(true)
            .write(true)
            .open(&idx_path)
            .await
            .unwrap();
        for i in [0u64, 256, 512] {
            let mut entry = [0u8; INDEX_ENTRY_SIZE];
            entry[..8].copy_from_slice(&i.to_be_bytes());
            entry[8..16].copy_from_slice(&(i * 50).to_be_bytes());
            entry[16..24].copy_from_slice(&0u64.to_be_bytes());
            file.write_all(&entry).await.unwrap();
        }
        file.flush().await.unwrap();

        let index = SegmentIndex::load(&idx_path).await.unwrap();
        // Target 300 is between 256 and 512, should return byte_position for 256
        assert_eq!(index.find_position(TopicOffset(300)), Some(256 * 50));
    }

    #[tokio::test]
    async fn index_find_position_before_all() {
        let tmp = TempDir::new().unwrap();
        let idx_path = tmp.path().join("test.idx");

        let mut file = OpenOptions::new()
            .create(true)
            .write(true)
            .open(&idx_path)
            .await
            .unwrap();
        // First entry at offset 256
        let mut entry = [0u8; INDEX_ENTRY_SIZE];
        entry[..8].copy_from_slice(&256u64.to_be_bytes());
        entry[8..16].copy_from_slice(&5000u64.to_be_bytes());
        entry[16..24].copy_from_slice(&0u64.to_be_bytes());
        file.write_all(&entry).await.unwrap();
        file.flush().await.unwrap();

        let index = SegmentIndex::load(&idx_path).await.unwrap();
        // Target 100 < first entry 256
        assert_eq!(index.find_position(TopicOffset(100)), None);
    }

    #[tokio::test]
    async fn index_load_missing_file() {
        let tmp = TempDir::new().unwrap();
        let idx_path = tmp.path().join("nonexistent.idx");
        let index = SegmentIndex::load(&idx_path).await.unwrap();
        assert!(index.entries.is_empty());
    }

    #[tokio::test]
    async fn append_writes_index_entries() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Write 1000 records with default index_interval=256
        for i in 0u64..1000 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    b"test-data".to_vec(),
                    now_millis(),
                )),
                offset: TopicOffset(i),
            };
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        storage.shutdown().await.unwrap();

        // Check the .idx file exists and has the right number of entries
        let seg_dir = storage.segments_dir(&ns, &topic);
        let segments = replay::list_segments(&seg_dir).await.unwrap();
        assert!(!segments.is_empty());

        let idx = index_path(&segments[0]);
        assert!(idx.exists(), ".idx file should exist");

        let index = SegmentIndex::load(&idx).await.unwrap();
        // With 1000 records, interval 256, first-record indexing:
        // entries at offsets 0, 256, 512, 768 = 4 entries
        assert_eq!(index.entries.len(), 4);
        assert_eq!(index.entries[0].offset, TopicOffset(0)); // first record indexed
        assert_eq!(index.entries[0].byte_position, 0); // starts at byte 0
        assert_eq!(index.entries[1].offset, TopicOffset(256));
        assert_eq!(index.entries[2].offset, TopicOffset(512));
        assert_eq!(index.entries[3].offset, TopicOffset(768));
    }

    #[tokio::test]
    async fn read_at_uses_index() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        for i in 0u64..1000 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    format!("payload-{i}").into_bytes(),
                    now_millis(),
                )),
                offset: TopicOffset(i),
            };
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        storage.flush(&ns, &topic).await.unwrap();

        // Read at offset 500 (past the first index entry at 255)
        let msg = storage
            .read_at(&ns, &topic, TopicOffset(500))
            .await
            .unwrap();
        assert!(msg.is_some());
        let msg = msg.unwrap();
        assert_eq!(msg.offset, TopicOffset(500));
        assert_eq!(msg.envelope.payload, b"payload-500");
    }

    #[tokio::test]
    async fn skip_to_uses_index() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        for i in 0u64..1000 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    format!("msg-{i}").into_bytes(),
                    now_millis(),
                )),
                offset: TopicOffset(i),
            };
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        storage.flush(&ns, &topic).await.unwrap();

        // Use read_cold_batch which internally uses SegmentCursor + skip_to
        let batch = storage
            .read_cold_batch(&ns, &topic, TopicOffset(500), 5)
            .await
            .unwrap();
        assert_eq!(batch.len(), 5);
        assert_eq!(batch[0].offset, TopicOffset(500));
        assert_eq!(batch[4].offset, TopicOffset(504));
        assert_eq!(batch[0].envelope.payload, b"msg-500");
    }

    #[tokio::test]
    async fn delete_segments_removes_idx() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        // Use small segment size to force multiple segments
        let storage = FsStorage::with_segment_max_bytes(tmp.path().join("data"), 200);
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        for i in 0u64..100 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    b"test-payload".to_vec(),
                    now_millis(),
                )),
                offset: TopicOffset(i),
            };
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        storage.shutdown().await.unwrap();

        let seg_dir = storage.segments_dir(&ns, &topic);
        let segments_before = replay::list_segments(&seg_dir).await.unwrap();
        assert!(segments_before.len() > 1, "should have multiple segments");

        // Verify .idx files exist for segments
        for seg in &segments_before {
            let idx = index_path(seg);
            assert!(idx.exists(), ".idx should exist for {seg:?}");
        }

        // Delete segments before a high offset
        let deleted = storage
            .delete_segments(&ns, &topic, Some(TopicOffset(50)), None, None)
            .await
            .unwrap();
        assert!(deleted > 0, "should have deleted some segments");

        // Verify .idx files were also deleted for removed segments
        let segments_after = replay::list_segments(&seg_dir).await.unwrap();
        for seg in &segments_before {
            if !segments_after.contains(seg) {
                let idx = index_path(seg);
                assert!(
                    !idx.exists(),
                    ".idx should be deleted for removed segment {seg:?}"
                );
            }
        }
    }

    #[tokio::test]
    async fn rebuild_index_matches_append() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        for i in 0u64..600 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    b"rebuild-test".to_vec(),
                    now_millis(),
                )),
                offset: TopicOffset(i),
            };
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        storage.shutdown().await.unwrap();

        let seg_dir = storage.segments_dir(&ns, &topic);
        let segments = replay::list_segments(&seg_dir).await.unwrap();
        let seg = &segments[0];
        let idx = index_path(seg);

        // Load the index written during append
        let original = SegmentIndex::load(&idx).await.unwrap();
        assert!(
            !original.entries.is_empty(),
            "should have index entries from append"
        );

        // Delete the index and rebuild it
        fs::remove_file(&idx).await.unwrap();
        assert!(!idx.exists());

        rebuild_index(seg, DEFAULT_INDEX_INTERVAL).await.unwrap();
        assert!(idx.exists(), "rebuild should create .idx");

        let rebuilt = SegmentIndex::load(&idx).await.unwrap();
        assert_eq!(original.entries.len(), rebuilt.entries.len());
        for (o, r) in original.entries.iter().zip(rebuilt.entries.iter()) {
            assert_eq!(o.offset, r.offset);
            assert_eq!(o.byte_position, r.byte_position);
            assert_eq!(o.timestamp_millis, r.timestamp_millis);
        }
    }

    // -- Namespace disk bytes tracking tests --

    fn make_msg(offset: u64, payload: &[u8]) -> PublishedMessage {
        use crate::types::{MessageEnvelope, MessageId, now_millis};
        PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(offset + 1),
                NamespaceId("ns".into()),
                TopicName("t".into()),
                payload.to_vec(),
                now_millis(),
            )),
            offset: TopicOffset(offset),
        }
    }

    #[tokio::test]
    async fn namespace_disk_bytes_tracked_on_append() {
        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Append 10 messages and track total record bytes
        for i in 0..10u64 {
            let msg = make_msg(i, b"hello-world");
            storage.append(&ns, &topic, &msg).await.unwrap();
        }

        let tracked = storage.namespace_disk_bytes(&ns).await;
        assert!(tracked > 0, "disk bytes should be tracked after appends");

        // Verify counter matches actual file size on disk
        let seg_dir = storage.segments_dir(&ns, &topic);
        let mut actual_bytes: u64 = 0;
        let mut dir = fs::read_dir(&seg_dir).await.unwrap();
        while let Some(entry) = dir.next_entry().await.unwrap() {
            let path = entry.path();
            let ext = path.extension().and_then(|e| e.to_str());
            if ext == Some("seg") || ext == Some("idx") {
                actual_bytes += fs::metadata(&path).await.unwrap().len();
            }
        }
        assert_eq!(
            tracked, actual_bytes,
            "tracked bytes should match actual segment + index file sizes"
        );
    }

    #[tokio::test]
    async fn namespace_disk_bytes_decremented_on_delete() {
        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::with_segment_max_bytes(tmp.path().join("data"), 200);
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Write enough data to create multiple segments
        for i in 0..50u64 {
            let msg = make_msg(i, &[0xAB; 50]);
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        storage.flush(&ns, &topic).await.unwrap();

        let before_delete = storage.namespace_disk_bytes(&ns).await;
        assert!(before_delete > 0);

        // Delete oldest segments
        let deleted = storage
            .delete_segments(&ns, &topic, None, Some(1), None)
            .await
            .unwrap();
        assert!(deleted > 0, "should have deleted at least one segment");

        let after_delete = storage.namespace_disk_bytes(&ns).await;
        assert!(
            after_delete < before_delete,
            "disk bytes should decrease after delete: {after_delete} < {before_delete}"
        );
    }

    #[tokio::test]
    async fn namespace_disk_bytes_populated_on_init() {
        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");

        // Write data with first storage instance
        let storage = FsStorage::new(&data_dir);
        storage.init().await.unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        for i in 0..5u64 {
            let msg = make_msg(i, b"init-test-payload");
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        let original_bytes = storage.namespace_disk_bytes(&ns).await;
        assert!(original_bytes > 0);
        storage.shutdown().await.unwrap();

        // Create new storage instance and verify counter is populated from disk
        let storage2 = FsStorage::new(&data_dir);
        storage2.init().await.unwrap();
        let reloaded_bytes = storage2.namespace_disk_bytes(&ns).await;
        assert_eq!(
            reloaded_bytes, original_bytes,
            "init should populate disk bytes from existing segments"
        );
    }

    // -- offset_at_timestamp tests --

    fn make_msg_with_ts(offset: u64, payload: &[u8], timestamp: u64) -> PublishedMessage {
        use crate::types::{MessageEnvelope, MessageId};
        PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(offset + 1),
                NamespaceId("ns".into()),
                TopicName("t".into()),
                payload.to_vec(),
                timestamp,
            )),
            offset: TopicOffset(offset),
        }
    }

    /// Helper: write messages, shutdown to flush .idx, re-init for queries.
    async fn storage_with_ts_messages(
        data_dir: &Path,
        count: u64,
        base_ts: u64,
        ts_step: u64,
    ) -> FsStorage {
        let storage = FsStorage::new(data_dir);
        storage.init().await.unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        for i in 0..count {
            let msg = make_msg_with_ts(i, b"payload", base_ts + i * ts_step);
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        storage.shutdown().await.unwrap();

        // Re-open and init to populate segment_index + index_cache from disk
        let storage = FsStorage::new(data_dir);
        storage.init().await.unwrap();
        storage
    }

    #[tokio::test]
    async fn offset_at_timestamp_exact_match() {
        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");
        let base_ts = 1_700_000_000_000u64;
        let storage = storage_with_ts_messages(&data_dir, 10, base_ts, 1000).await;
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        let result = storage
            .offset_at_timestamp(&ns, &topic, base_ts + 5000)
            .await
            .unwrap();
        assert_eq!(result, Some(TopicOffset(5)));
    }

    #[tokio::test]
    async fn offset_at_timestamp_between_entries() {
        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");
        let base_ts = 1_700_000_000_000u64;
        // Timestamps: base, base+2000, base+4000, base+6000, ...
        let storage = storage_with_ts_messages(&data_dir, 10, base_ts, 2000).await;
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Timestamp between offset 2 (ts=base+4000) and offset 3 (ts=base+6000)
        // Should return offset 3 (first message >= target)
        let result = storage
            .offset_at_timestamp(&ns, &topic, base_ts + 5000)
            .await
            .unwrap();
        assert_eq!(result, Some(TopicOffset(3)));
    }

    #[tokio::test]
    async fn offset_at_timestamp_before_all_data() {
        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");
        let base_ts = 1_700_000_000_000u64;
        let storage = storage_with_ts_messages(&data_dir, 5, base_ts, 1000).await;
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        let result = storage
            .offset_at_timestamp(&ns, &topic, base_ts - 999_999)
            .await
            .unwrap();
        assert_eq!(result, Some(TopicOffset(0)));
    }

    #[tokio::test]
    async fn offset_at_timestamp_after_all_data() {
        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");
        let base_ts = 1_700_000_000_000u64;
        let storage = storage_with_ts_messages(&data_dir, 5, base_ts, 1000).await;
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        let result = storage
            .offset_at_timestamp(&ns, &topic, base_ts + 999_999_999)
            .await
            .unwrap();
        assert_eq!(result, None);
    }

    #[tokio::test]
    async fn offset_at_timestamp_nonexistent_topic() {
        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("nonexistent".into());

        let result = storage
            .offset_at_timestamp(&ns, &topic, 1_700_000_000_000)
            .await
            .unwrap();
        assert_eq!(result, None);
    }

    /// Helper: write messages with controlled timestamps to small-segment storage,
    /// shutdown, and re-init for multi-segment timestamp seek tests.
    async fn storage_with_small_segments(
        data_dir: &Path,
        count: u64,
        base_ts: u64,
        ts_step: u64,
    ) -> FsStorage {
        // 512 bytes ≈ 5-6 messages per segment with ~100-byte records
        let storage = FsStorage::with_segment_max_bytes(data_dir, 512);
        storage.init().await.unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        for i in 0..count {
            let msg = make_msg_with_ts(i, b"payload-data", base_ts + i * ts_step);
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        storage.shutdown().await.unwrap();

        let storage = FsStorage::with_segment_max_bytes(data_dir, 512);
        storage.init().await.unwrap();
        storage
    }

    #[tokio::test]
    async fn offset_at_timestamp_spans_segments() {
        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");
        let base_ts = 1_700_000_000_000u64;
        // 50 messages across ~10 segments, timestamps spaced 1000ms apart
        let storage = storage_with_small_segments(&data_dir, 50, base_ts, 1000).await;
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Verify we have multiple segments
        let seg_dir = storage.segments_dir(&ns, &topic);
        let seg_count = count_seg_files(&seg_dir).await;
        assert!(
            seg_count >= 3,
            "expected multiple segments, got {seg_count}"
        );

        // Seek to timestamp in the middle — should find the correct offset
        let result = storage
            .offset_at_timestamp(&ns, &topic, base_ts + 25_000)
            .await
            .unwrap();
        assert_eq!(result, Some(TopicOffset(25)));

        // Seek to timestamp near the end
        let result = storage
            .offset_at_timestamp(&ns, &topic, base_ts + 45_000)
            .await
            .unwrap();
        assert_eq!(result, Some(TopicOffset(45)));

        // Seek to timestamp at the very start
        let result = storage
            .offset_at_timestamp(&ns, &topic, base_ts)
            .await
            .unwrap();
        assert_eq!(result, Some(TopicOffset(0)));
    }

    #[tokio::test]
    async fn offset_at_timestamp_no_index_fallback() {
        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");
        let base_ts = 1_700_000_000_000u64;
        let storage = storage_with_small_segments(&data_dir, 30, base_ts, 1000).await;
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Delete all .idx files to force linear scan fallback
        let seg_dir = storage.segments_dir(&ns, &topic);
        let mut dir = fs::read_dir(&seg_dir).await.unwrap();
        while let Some(entry) = dir.next_entry().await.unwrap() {
            if entry.path().extension().and_then(|e| e.to_str()) == Some("idx") {
                fs::remove_file(entry.path()).await.unwrap();
            }
        }

        // Clear the index cache so load_index returns None
        {
            let mut cache = storage.index_cache.write().await;
            cache.clear();
        }

        // Should still find the correct offset via linear scan
        let result = storage
            .offset_at_timestamp(&ns, &topic, base_ts + 15_000)
            .await
            .unwrap();
        assert_eq!(result, Some(TopicOffset(15)));
    }

    // ---- Phase D: Segment compaction tests ----

    /// Helper: create multiple small segment files (one per message).
    ///
    /// Uses a tiny `segment_max_bytes` to force rotation on every append, then
    /// returns a **new** `FsStorage` with the default `segment_max_bytes` (64MB)
    /// pointing at the same data directory. This ensures the merge cap (#8)
    /// doesn't block compaction in tests.
    async fn make_small_segments(
        tmp: &TempDir,
        ns: &NamespaceId,
        topic: &TopicName,
        count: u64,
    ) -> FsStorage {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let data_dir = tmp.path().join("data");

        // Phase 1: write with tiny segment max → one segment per message
        {
            let writer = FsStorage::with_segment_max_bytes(&data_dir, 1);
            writer.init().await.unwrap();

            for i in 0..count {
                let msg = PublishedMessage {
                    envelope: Arc::new(MessageEnvelope::new(
                        MessageId(i + 1),
                        ns.clone(),
                        topic.clone(),
                        format!("payload-{i}").into_bytes(),
                        now_millis() + i,
                    )),
                    offset: TopicOffset(i),
                };
                writer.append(ns, topic, &msg).await.unwrap();
            }
            writer.shutdown().await.unwrap();
        }

        // Phase 2: reopen with default segment_max_bytes so merge cap is sane
        let storage = FsStorage::new(&data_dir);
        storage.init().await.unwrap();
        storage
    }

    #[tokio::test]
    async fn compact_segments_merges_small_segments() {
        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        // 5 messages → 5 segments (one per message with max_bytes=1)
        let storage = make_small_segments(&tmp, &ns, &topic, 5).await;

        let seg_dir = storage.segments_dir(&ns, &topic);
        let before = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(before.len(), 5, "should have 5 small segments");

        // Threshold large enough to include all; no safe_below
        // Active writer is segment 4 → candidates are 0,1,2,3 → 1 merge
        let merges = storage
            .compact_segments(&ns, &topic, 1_000_000, None)
            .await
            .unwrap();
        assert_eq!(
            merges, 1,
            "should perform exactly one merge of segments 0-3"
        );

        // After: merged segment (base=0) + active segment (base=4) = 2
        let after = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(after.len(), 2, "should have 2 segments after merge");

        let replayed = storage.replay(&ns, &topic).await.unwrap();
        assert_eq!(replayed.len(), 5, "all 5 records should be preserved");
        for (i, msg) in replayed.iter().enumerate() {
            assert_eq!(msg.offset, TopicOffset(i as u64));
        }
    }

    #[tokio::test]
    async fn compact_segments_skips_large_segments() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        let data_dir = tmp.path().join("data");

        // Phase 1: write with tiny segment max → one segment per message
        {
            let writer = FsStorage::with_segment_max_bytes(&data_dir, 1);
            writer.init().await.unwrap();

            // Messages 0,1: large payloads (1000 bytes each)
            // Messages 2,3,4: small payloads (10 bytes each)
            for i in 0..5u64 {
                let payload = if i < 2 {
                    vec![b'X'; 1000]
                } else {
                    vec![b'x'; 10]
                };
                let msg = PublishedMessage {
                    envelope: Arc::new(MessageEnvelope::new(
                        MessageId(i + 1),
                        ns.clone(),
                        topic.clone(),
                        payload,
                        now_millis() + i,
                    )),
                    offset: TopicOffset(i),
                };
                writer.append(&ns, &topic, &msg).await.unwrap();
            }
            writer.shutdown().await.unwrap();
        }

        // Phase 2: reopen with default segment_max_bytes
        let storage = FsStorage::new(&data_dir);
        storage.init().await.unwrap();

        let seg_dir = storage.segments_dir(&ns, &topic);
        let before = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(before.len(), 5);

        // Stat the large vs small segments
        let large_size = fs::metadata(&before[0]).await.unwrap().len();
        let small_size = fs::metadata(&before[2]).await.unwrap().len();

        // Threshold between small and large segment sizes
        let threshold = (small_size + large_size) / 2;
        assert!(small_size < threshold && large_size >= threshold);

        // Active writer is segment 4 → candidates: 0,1,2,3
        // Segments 0,1 are large (above threshold) → not in any run
        // Segments 2,3 are small → run of 2 → 1 merge
        let merges = storage
            .compact_segments(&ns, &topic, threshold, None)
            .await
            .unwrap();
        assert_eq!(merges, 1, "should merge the small run (segments 2+3)");

        // After: seg 0, seg 1, merged(2+3), seg 4 = 4 segments
        let after = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(after.len(), 4, "large segments should be untouched");

        // All records still present
        let replayed = storage.replay(&ns, &topic).await.unwrap();
        assert_eq!(replayed.len(), 5);
    }

    #[tokio::test]
    async fn compact_segments_preserves_records() {
        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        let storage = make_small_segments(&tmp, &ns, &topic, 8).await;

        // Capture original records before compaction
        let original = storage.replay(&ns, &topic).await.unwrap();

        storage
            .compact_segments(&ns, &topic, 1_000_000, None)
            .await
            .unwrap();

        // Replay after compaction must match exactly
        let after = storage.replay(&ns, &topic).await.unwrap();
        assert_eq!(after.len(), original.len());
        for (a, b) in original.iter().zip(after.iter()) {
            assert_eq!(a.offset, b.offset);
            assert_eq!(a.envelope.payload, b.envelope.payload);
            assert_eq!(a.envelope.timestamp, b.envelope.timestamp);
        }

        // Verify read_at works for every offset (exercises segment_index cache)
        for msg in &original {
            let found = storage.read_at(&ns, &topic, msg.offset).await.unwrap();
            assert!(
                found.is_some(),
                "read_at should find offset {} after merge",
                msg.offset.0
            );
            assert_eq!(found.unwrap().offset, msg.offset);
        }
    }

    #[tokio::test]
    async fn compact_segments_noop_when_none_small() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Use normal segment size — all messages in one segment
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        for i in 0..3u64 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    b"data".to_vec(),
                    now_millis() + i,
                )),
                offset: TopicOffset(i),
            };
            storage.append(&ns, &topic, &msg).await.unwrap();
        }
        storage.flush(&ns, &topic).await.unwrap();

        // Threshold of 1 byte — no segment is that small
        let merges = storage
            .compact_segments(&ns, &topic, 1, None)
            .await
            .unwrap();
        assert_eq!(merges, 0, "no segments should be small enough to merge");
    }

    #[tokio::test]
    async fn compact_segments_writes_idx() {
        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        let storage = make_small_segments(&tmp, &ns, &topic, 5).await;

        storage
            .compact_segments(&ns, &topic, 1_000_000, None)
            .await
            .unwrap();

        // After merge, check that .idx files exist for remaining segments
        let seg_dir = storage.segments_dir(&ns, &topic);
        let segs = replay::list_segments(&seg_dir).await.unwrap();
        for seg_path in &segs {
            let idx = index_path(seg_path);
            assert!(idx.exists(), ".idx should exist for segment {:?}", seg_path);
            // Index file should be non-empty (at least one entry for the first record)
            let meta = fs::metadata(&idx).await.unwrap();
            assert!(
                meta.len() >= INDEX_ENTRY_SIZE as u64,
                ".idx should contain at least one entry"
            );
        }

        // Verify the merged segment's .idx has correct entry count
        // Merged segment has 4 records (offsets 0-3), default index_interval=256
        // → only the first record is indexed → exactly 1 entry = 24 bytes
        let merged_idx = index_path(&segs[0]);
        let meta = fs::metadata(&merged_idx).await.unwrap();
        assert_eq!(
            meta.len(),
            INDEX_ENTRY_SIZE as u64,
            "merged segment with 4 records should have exactly 1 index entry"
        );
    }

    #[tokio::test]
    async fn compact_segments_respects_safe_below() {
        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        // Creates 5 segments with base offsets 0, 1, 2, 3, 4
        let storage = make_small_segments(&tmp, &ns, &topic, 5).await;

        let seg_dir = storage.segments_dir(&ns, &topic);
        let before = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(before.len(), 5);

        // safe_below=2 → segments with base >= 2 excluded.
        // Active writer is segment 4.
        // Candidates: segments 0, 1 only. Run of 2 → 1 merge.
        // Result: merged(0+1), seg 2, seg 3, seg 4 = 4 segments.
        let merges = storage
            .compact_segments(&ns, &topic, 1_000_000, Some(TopicOffset(2)))
            .await
            .unwrap();
        assert_eq!(merges, 1, "should merge segments 0 and 1");

        let after = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(after.len(), 4, "should have 4 segments after merge");

        // All records preserved
        let replayed = storage.replay(&ns, &topic).await.unwrap();
        assert_eq!(replayed.len(), 5);
        for (i, msg) in replayed.iter().enumerate() {
            assert_eq!(msg.offset, TopicOffset(i as u64));
        }
    }

    #[tokio::test]
    async fn compact_segments_nonexistent_topic() {
        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        // Topic was never written to — dir doesn't exist
        let ns = NamespaceId("ns".into());
        let topic = TopicName("nonexistent".into());
        let merges = storage
            .compact_segments(&ns, &topic, 1_000_000, None)
            .await
            .unwrap();
        assert_eq!(merges, 0);
    }

    #[tokio::test]
    async fn compact_segments_cleans_orphan_tmp_files() {
        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        let storage = make_small_segments(&tmp, &ns, &topic, 3).await;

        let seg_dir = storage.segments_dir(&ns, &topic);

        // Simulate crash leftovers: create orphan .tmp files
        let orphan_seg = seg_dir.join("00000000000000000099.seg.tmp");
        let orphan_idx = seg_dir.join("00000000000000000099.idx.tmp");
        fs::write(&orphan_seg, b"partial merge data").await.unwrap();
        fs::write(&orphan_idx, b"partial index data").await.unwrap();

        assert!(orphan_seg.exists());
        assert!(orphan_idx.exists());

        // compact_segments should clean up .tmp files
        storage
            .compact_segments(&ns, &topic, 1_000_000, None)
            .await
            .unwrap();

        assert!(!orphan_seg.exists(), "orphan .seg.tmp should be deleted");
        assert!(!orphan_idx.exists(), "orphan .idx.tmp should be deleted");
    }

    #[tokio::test]
    async fn compact_segments_writer_evicted_from_lru() {
        use crate::config::StorageConfig;
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");

        // Phase 1: write small segments with tiny max_bytes
        {
            let writer = FsStorage::with_segment_max_bytes(&data_dir, 1);
            writer.init().await.unwrap();

            let ns = NamespaceId("ns".into());
            let topic = TopicName("t".into());
            for i in 0..5u64 {
                let msg = PublishedMessage {
                    envelope: Arc::new(MessageEnvelope::new(
                        MessageId(i + 1),
                        ns.clone(),
                        topic.clone(),
                        format!("payload-{i}").into_bytes(),
                        now_millis() + i,
                    )),
                    offset: TopicOffset(i),
                };
                writer.append(&ns, &topic, &msg).await.unwrap();
            }
            writer.shutdown().await.unwrap();
        }

        // Phase 2: reopen with max_open_fds=1 so the writer is easily evicted
        let config = StorageConfig {
            data_dir: data_dir.to_str().unwrap().to_owned(),
            max_open_fds: 1,
            ..StorageConfig::default()
        };
        let storage = FsStorage::from_config(&config);
        storage.init().await.unwrap();

        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Write to a second topic to evict the first from the LRU cache
        let topic2 = TopicName("t2".into());
        let msg = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(100),
                ns.clone(),
                topic2.clone(),
                b"evict".to_vec(),
                now_millis(),
            )),
            offset: TopicOffset(0),
        };
        storage.append(&ns, &topic2, &msg).await.unwrap();

        // Verify the first topic's writer is evicted
        {
            let key = TopicKey {
                namespace: "ns".into(),
                topic: "t".into(),
            };
            let cache = storage.writers.lock().await;
            assert!(
                cache.peek(&key).is_none(),
                "writer for topic 't' should be evicted from LRU"
            );
        }

        let seg_dir = storage.segments_dir(&ns, &topic);
        let before = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(before.len(), 5);

        // Compact should still work — falls back to skipping the last segment
        let merges = storage
            .compact_segments(&ns, &topic, 1_000_000, None)
            .await
            .unwrap();
        assert_eq!(
            merges, 1,
            "should merge segments 0-3 (last skipped as fallback)"
        );

        let after = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(after.len(), 2);

        let replayed = storage.replay(&ns, &topic).await.unwrap();
        assert_eq!(replayed.len(), 5);
    }

    #[tokio::test]
    async fn compact_segments_error_cleans_temp_files() {
        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        let storage = make_small_segments(&tmp, &ns, &topic, 5).await;

        let seg_dir = storage.segments_dir(&ns, &topic);
        let before = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(before.len(), 5);

        // Replace segment 1 with a directory — RecordStream::open will fail
        // with an I/O error, triggering the error-handling path.
        fs::remove_file(&before[1]).await.unwrap();
        fs::create_dir(&before[1]).await.unwrap();

        // compact_segments should handle the error gracefully (not propagate)
        let merges = storage
            .compact_segments(&ns, &topic, 1_000_000, None)
            .await
            .unwrap();

        // The single run (segments 0-3) should fail because segment 1 can't be read.
        // Error handling cleans up temp files and continues.
        assert_eq!(merges, 0, "run with unreadable segment should fail");

        // Verify no .tmp files were left behind
        let mut has_tmp = false;
        let mut read_dir = fs::read_dir(&seg_dir).await.unwrap();
        while let Some(entry) = read_dir.next_entry().await.unwrap() {
            let path = entry.path();
            if path.is_file() && path.extension().and_then(|e| e.to_str()) == Some("tmp") {
                has_tmp = true;
            }
        }
        assert!(!has_tmp, "no .tmp files should remain after error");
    }

    #[tokio::test]
    async fn compact_segments_updates_namespace_disk_bytes() {
        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        let storage = make_small_segments(&tmp, &ns, &topic, 5).await;

        let bytes_before = storage.namespace_disk_bytes(&ns).await;
        assert!(bytes_before > 0, "should have tracked disk bytes from init");

        storage
            .compact_segments(&ns, &topic, 1_000_000, None)
            .await
            .unwrap();

        let bytes_after = storage.namespace_disk_bytes(&ns).await;
        // Merged file should be ≤ sum of originals (no overhead, just reorganized).
        // The 4 small .seg + .idx files are replaced by 1 .seg + .idx.
        // The metadata overhead per segment (file system alignment) means the
        // merged total may differ slightly. Just verify it was updated and not zero.
        assert!(
            bytes_after > 0,
            "disk bytes should still be tracked after merge"
        );
        // Disk bytes should have changed (fewer index files = less overhead)
        // In practice: 4 segments with 4 indexes → 1 segment with 1 index
        // Total data is the same, but index overhead shrinks significantly.
    }

    #[tokio::test]
    async fn compact_segments_safe_below_excludes_all_candidates() {
        let tmp = TempDir::new().unwrap();
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());
        // Creates 5 segments with base offsets 0, 1, 2, 3, 4
        let storage = make_small_segments(&tmp, &ns, &topic, 5).await;

        // safe_below=1 → only segment 0 is a candidate (active writer=4 also excluded)
        // 1 candidate < 2 → return 0
        let merges = storage
            .compact_segments(&ns, &topic, 1_000_000, Some(TopicOffset(1)))
            .await
            .unwrap();
        assert_eq!(merges, 0, "only 1 candidate, cannot form a run");

        // Verify nothing changed
        let seg_dir = storage.segments_dir(&ns, &topic);
        let segs = replay::list_segments(&seg_dir).await.unwrap();
        assert_eq!(segs.len(), 5);
    }

    #[tokio::test]
    async fn compact_segments_converts_v1_to_v2() {
        use crate::storage::replay::tests::encode_record_v1;
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let data_dir = tmp.path().join("data");
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Manually create two small v1-format segment files
        let seg_dir = data_dir.join("ns").join("t").join("segments");
        fs::create_dir_all(&seg_dir).await.unwrap();

        // Segment 0: offsets 0-1 (v1 format — no version prefix, no headers)
        let mut seg0_data = Vec::new();
        for i in 0..2u64 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    format!("v1-payload-{i}").into_bytes(),
                    now_millis() + i,
                )),
                offset: TopicOffset(i),
            };
            seg0_data.extend_from_slice(&encode_record_v1(&msg));
        }
        fs::write(seg_dir.join("00000000000000000000.seg"), &seg0_data)
            .await
            .unwrap();

        // Segment 1: offsets 2-3 (v1 format)
        let mut seg1_data = Vec::new();
        for i in 2..4u64 {
            let msg = PublishedMessage {
                envelope: Arc::new(MessageEnvelope::new(
                    MessageId(i + 1),
                    ns.clone(),
                    topic.clone(),
                    format!("v1-payload-{i}").into_bytes(),
                    now_millis() + i,
                )),
                offset: TopicOffset(i),
            };
            seg1_data.extend_from_slice(&encode_record_v1(&msg));
        }
        fs::write(seg_dir.join("00000000000000000002.seg"), &seg1_data)
            .await
            .unwrap();

        // Segment 2: offset 4 (active writer — won't be compacted)
        let mut seg2_data = Vec::new();
        let msg = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(5),
                ns.clone(),
                topic.clone(),
                b"v1-payload-4".to_vec(),
                now_millis() + 4,
            )),
            offset: TopicOffset(4),
        };
        seg2_data.extend_from_slice(&encode_record_v1(&msg));
        fs::write(seg_dir.join("00000000000000000004.seg"), &seg2_data)
            .await
            .unwrap();

        // Open storage and compact
        let storage = FsStorage::new(&data_dir);
        storage.init().await.unwrap();

        let merges = storage
            .compact_segments(&ns, &topic, 1_000_000, None)
            .await
            .unwrap();
        assert_eq!(merges, 1, "should merge the two v1 segments");

        // Read back the merged segment — should be v2 format with empty headers
        let segs = replay::list_segments(&seg_dir).await.unwrap();
        // Merged segment (offset 0) + active writer (offset 4) = 2 segments
        assert_eq!(segs.len(), 2);

        let mut stream = replay::RecordStream::open(&segs[0]).await.unwrap();
        let mut records = Vec::new();
        while let Some((offset, envelope)) = stream.next().await.unwrap() {
            records.push((offset, envelope));
        }

        assert_eq!(records.len(), 4, "merged segment should have 4 records");
        for (i, (offset, envelope)) in records.iter().enumerate() {
            assert_eq!(*offset, TopicOffset(i as u64));
            assert_eq!(envelope.payload, format!("v1-payload-{i}").into_bytes());
            assert!(
                envelope.headers.is_empty(),
                "compacted v1 records should have empty headers"
            );
        }

        // Verify the merged segment is actually v2 format by checking raw bytes
        let raw = fs::read(&segs[0]).await.unwrap();
        // Skip 4 bytes (record_len) + 8 bytes (offset) → byte 12 should be 0x00 sentinel
        assert_eq!(
            raw[12], 0x00,
            "merged segment should use v2 format (0x00 sentinel)"
        );
        assert_eq!(
            raw[13], 0x02,
            "merged segment should use v2 format (version byte)"
        );
    }

    #[tokio::test]
    async fn delete_topic_dir_removes_directory() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns1".into());
        let topic = TopicName("topic-a".into());

        let msg = PublishedMessage {
            offset: TopicOffset(0),
            envelope: Arc::new(MessageEnvelope::new(
                MessageId::new(1),
                ns.clone(),
                topic.clone(),
                b"data".to_vec(),
                now_millis(),
            )),
        };
        storage.append(&ns, &topic, &msg).await.unwrap();
        storage.flush(&ns, &topic).await.unwrap();

        let seg_dir = storage.segments_dir(&ns, &topic);
        assert!(seg_dir.exists());

        storage.close_topic_writer(&ns, &topic).await.unwrap();
        storage.delete_topic_dir(&ns, &topic).await.unwrap();

        assert!(!seg_dir.exists());
    }

    #[tokio::test]
    async fn close_topic_writer_allows_subsequent_write() {
        use crate::types::{MessageEnvelope, MessageId, now_millis};

        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns1".into());
        let topic = TopicName("topic-a".into());

        let msg = PublishedMessage {
            offset: TopicOffset(0),
            envelope: Arc::new(MessageEnvelope::new(
                MessageId::new(1),
                ns.clone(),
                topic.clone(),
                b"data".to_vec(),
                now_millis(),
            )),
        };
        storage.append(&ns, &topic, &msg).await.unwrap();
        storage.close_topic_writer(&ns, &topic).await.unwrap();

        let msg2 = PublishedMessage {
            offset: TopicOffset(1),
            envelope: Arc::new(MessageEnvelope::new(
                MessageId::new(2),
                ns.clone(),
                topic.clone(),
                b"data2".to_vec(),
                now_millis(),
            )),
        };
        storage.append(&ns, &topic, &msg2).await.unwrap();
    }

    #[tokio::test]
    async fn delete_topic_dir_nonexistent_is_ok() {
        let tmp = TempDir::new().unwrap();
        let storage = FsStorage::new(tmp.path().join("data"));
        storage.init().await.unwrap();

        let ns = NamespaceId("ns1".into());
        let topic = TopicName("nonexistent".into());

        storage.delete_topic_dir(&ns, &topic).await.unwrap();
    }
}
