ugnos 0.2.1

use crate::error::DbError;
use crate::types::{Row, TagSet, Timestamp, Value};

use crc32fast::Hasher as Crc32;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, HashMap};
use std::fs::{self, File, OpenOptions};
use std::io::{BufWriter, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::{mpsc, Arc, Mutex, RwLock};
use std::thread::{self, JoinHandle};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

const SEG_MAGIC: &[u8; 8] = b"UGNSEG01";
const SEG_FOOTER_MAGIC: &[u8; 8] = b"UGNSEGF1";
const SER_BLOCK_MAGIC: &[u8; 8] = b"UGNSER01";
const MANIFEST_MAGIC: &[u8; 8] = b"UGNMAN01";

const SEG_VERSION: u32 = 1;
const MANIFEST_VERSION: u32 = 1;

const FOOTER_LEN: u64 = 8 + 8 + 8 + 4; // magic + index_off + index_len + crc32

#[derive(Debug, Clone)]
pub struct SegmentStoreConfig {
    pub compaction_check_interval: Duration,
    pub l0_compaction_trigger_segment_count: usize,
}

impl Default for SegmentStoreConfig {
    fn default() -> Self {
        Self {
            compaction_check_interval: Duration::from_secs(1),
            l0_compaction_trigger_segment_count: 4,
        }
    }
}

#[derive(Debug, Clone, Copy)]
pub struct CompactionStats {
    pub input_segments: usize,
    pub output_segments: usize,
}

#[derive(Debug)]
pub struct SegmentStore {
    manifest_path: PathBuf,
    segments_dir: PathBuf,
    tmp_dir: PathBuf,

    state: Arc<RwLock<StoreState>>,

    compaction_tx: mpsc::Sender<CompactionCmd>,
    compaction_handle: Mutex<Option<JoinHandle<()>>>,
}

#[derive(Debug)]
struct StoreState {
    manifest: Manifest,
    // Active segments used for reads/compaction. Readers clone Arcs.
    active: Vec<Arc<Segment>>,
    // Removed from active; may be deleted by the reaper once no readers hold them.
    obsolete: Vec<Arc<Segment>>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
struct Manifest {
    version: u32,
    next_segment_id: u64,
    delete_before: Option<Timestamp>,
    // Persisted copy of segment metadata for faster startup.
    segments: Vec<SegmentRecord>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
struct SegmentRecord {
    id: u64,
    level: u8,
    created_at: Timestamp,
    max_seq: u64,
    min_ts: Timestamp,
    max_ts: Timestamp,
    file_name: String,
    series: BTreeMap<String, SeriesBlockMeta>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
pub(crate) struct SeriesBlockMeta {
    pub offset: u64,
    pub len: u64,
    pub row_count: u32,
    pub min_ts: Timestamp,
    pub max_ts: Timestamp,
    pub crc32: u32,
}

#[derive(Debug)]
struct Segment {
    rec: SegmentRecord,
    path: PathBuf,
}

#[derive(Debug)]
enum CompactionCmd {
    Maybe,
    Force {
        ack: mpsc::Sender<Result<CompactionStats, DbError>>,
    },
    Shutdown,
}

impl SegmentStore {
    pub fn open<P: AsRef<Path>>(dir: P, config: SegmentStoreConfig) -> Result<Self, DbError> {
        let dir = dir.as_ref().to_path_buf();
        let segments_dir = dir.join("segments");
        let tmp_dir = segments_dir.join("tmp");
        let manifest_path = segments_dir.join("MANIFEST.bin");

        fs::create_dir_all(&tmp_dir)?;

        let manifest = if manifest_path.exists() {
            read_manifest(&manifest_path)?
        } else {
            Manifest {
                version: MANIFEST_VERSION,
                next_segment_id: 1,
                delete_before: None,
                segments: Vec::new(),
            }
        };

        // Ensure we can load all referenced segments; if any is missing/corrupt, fail fast.
        let mut active = Vec::with_capacity(manifest.segments.len());
        for rec in &manifest.segments {
            let path = segments_dir.join(&rec.file_name);
            // Validate footer/index CRC early; also normalizes any future upgrades.
            let _ = load_segment_index(&path)?;
            active.push(Arc::new(Segment { rec: rec.clone(), path }));
        }

        // Persist manifest on first creation (so header/magic exists).
        if !manifest_path.exists() {
            write_manifest_atomic(&manifest_path, &tmp_dir, &manifest)?;
        }

        let state = Arc::new(RwLock::new(StoreState { manifest, active, obsolete: Vec::new() }));
        let (tx, rx) = mpsc::channel();
        let state_clone = Arc::clone(&state);
        let segments_dir_clone = segments_dir.clone();
        let tmp_dir_clone = tmp_dir.clone();
        let manifest_path_clone = manifest_path.clone();
        let cfg_clone = config.clone();

        let handle = thread::spawn(move || compaction_loop(
            rx,
            state_clone,
            &segments_dir_clone,
            &tmp_dir_clone,
            &manifest_path_clone,
            cfg_clone,
        ));

        Ok(Self {
            manifest_path,
            segments_dir,
            tmp_dir,
            state,
            compaction_tx: tx,
            compaction_handle: Mutex::new(Some(handle)),
        })
    }

    pub fn delete_before(&self) -> Option<Timestamp> {
        self.state.read().ok().and_then(|s| s.manifest.delete_before)
    }

    /// Advances the delete-before tombstone watermark (never decreases).
    ///
    /// This makes retention effective immediately for reads, and compaction will later
    /// physically drop the data.
    pub fn advance_delete_before(&self, delete_before: Timestamp) -> Result<(), DbError> {
        let mut st = self.state.write()?;
        let cur = st.manifest.delete_before.unwrap_or(0);
        if delete_before <= cur {
            return Ok(());
        }
        st.manifest.delete_before = Some(delete_before);
        write_manifest_atomic(&self.manifest_path, &self.tmp_dir, &st.manifest)?;
        Ok(())
    }

    pub fn max_persisted_seq(&self) -> u64 {
        self.state
            .read()
            .map(|s| s.active.iter().map(|seg| seg.rec.max_seq).max().unwrap_or(0))
            .unwrap_or(0)
    }

    pub(crate) fn ingest_l0(&self, mut rows_by_series: HashMap<String, Vec<Row>>) -> Result<(), DbError> {
        // Sort each series (timestamp, seq) for deterministic layout & query binary search.
        for rows in rows_by_series.values_mut() {
            rows.sort_unstable_by(|a, b| (a.timestamp, a.seq).cmp(&(b.timestamp, b.seq)));
        }

        let created_at = now_ns();

        // Assign id and create filename under manifest lock.
        let (id, delete_before) = {
            let mut st = self.state.write()?;
            let id = st.manifest.next_segment_id;
            st.manifest.next_segment_id = st.manifest.next_segment_id.saturating_add(1);
            let delete_before = st.manifest.delete_before;
            (id, delete_before)
        };

        // Apply retention to newly created segments too (prevents reintroducing expired data).
        let delete_before_ts = delete_before.unwrap_or(0);
        if delete_before_ts > 0 {
            for rows in rows_by_series.values_mut() {
                rows.retain(|r| r.timestamp >= delete_before_ts);
            }
            rows_by_series.retain(|_, rows| !rows.is_empty());
            if rows_by_series.is_empty() {
                return Err(DbError::Internal("Refusing to write an empty segment after retention filter".to_string()));
            }
        }

        let file_name = format!("seg_{:020}_l0.seg", id);
        let final_path = self.segments_dir.join(&file_name);
        let tmp_path = self.tmp_dir.join(format!("{}.tmp", &file_name));

        let rec = write_segment_file(
            &tmp_path,
            &final_path,
            id,
            0,
            created_at,
            delete_before,
            rows_by_series,
        )?;

        // Install into manifest + active set atomically.
        {
            let mut st = self.state.write()?;
            st.manifest.segments.push(rec.clone());
            write_manifest_atomic(&self.manifest_path, &self.tmp_dir, &st.manifest)?;
            st.active.push(Arc::new(Segment { rec: rec.clone(), path: final_path }));
        }

        // Trigger compaction opportunistically.
        let _ = self.compaction_tx.send(CompactionCmd::Maybe);
        Ok(())
    }

    pub fn query(
        &self,
        series: &str,
        time_range: std::ops::Range<Timestamp>,
        tag_filter: Option<&TagSet>,
    ) -> Result<Vec<(Timestamp, Value)>, DbError> {
        if time_range.start >= time_range.end {
            return Err(DbError::InvalidTimeRange { start: time_range.start, end: time_range.end });
        }

        let (segments, delete_before) = {
            let st = self.state.read()?;
            (st.active.clone(), st.manifest.delete_before)
        };

        let delete_before = delete_before.unwrap_or(0);
        let mut out = Vec::new();
        let mut seen_series = false;

        for seg in segments {
            // Segment-level min/max filter first.
            if seg.rec.max_ts < time_range.start || seg.rec.min_ts >= time_range.end {
                continue;
            }

            let Some(meta) = seg.rec.series.get(series) else { continue };
            seen_series = true;
            if meta.max_ts < time_range.start || meta.min_ts >= time_range.end {
                continue;
            }

            let mut results = read_series_range(
                &seg.path,
                meta,
                time_range.clone(),
                tag_filter,
                delete_before,
            )?;
            out.append(&mut results);
        }

        if !seen_series {
            return Err(DbError::SeriesNotFound(series.to_string()));
        }
        Ok(out)
    }

    /// Forces a full L0 compaction now and waits for completion.
    pub fn compact_blocking(&self) -> Result<CompactionStats, DbError> {
        let (tx, rx) = mpsc::channel();
        self.compaction_tx
            .send(CompactionCmd::Force { ack: tx })
            .map_err(|e| DbError::BackgroundTaskError(format!("Failed to request compaction: {}", e)))?;
        rx.recv().map_err(|e| DbError::BackgroundTaskError(format!("Failed to receive compaction ack: {}", e)))?
    }
}

impl Drop for SegmentStore {
    fn drop(&mut self) {
        let _ = self.compaction_tx.send(CompactionCmd::Shutdown);
        if let Ok(mut h) = self.compaction_handle.lock() {
            if let Some(handle) = h.take() {
                let _ = handle.join();
            }
        }
    }
}

fn now_ns() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos() as u64
}

fn compaction_loop(
    rx: mpsc::Receiver<CompactionCmd>,
    state: Arc<RwLock<StoreState>>,
    segments_dir: &Path,
    tmp_dir: &Path,
    manifest_path: &Path,
    cfg: SegmentStoreConfig,
) {
    let mut last_check = SystemTime::now();
    loop {
        let timeout = cfg.compaction_check_interval;
        match rx.recv_timeout(timeout) {
            Ok(CompactionCmd::Shutdown) | Err(mpsc::RecvTimeoutError::Disconnected) => break,
            Ok(CompactionCmd::Force { ack }) => {
                let res = compact_l0_once(&state, segments_dir, tmp_dir, manifest_path);
                let _ = ack.send(res);
            }
            Ok(CompactionCmd::Maybe) | Err(mpsc::RecvTimeoutError::Timeout) => {
                let _ = last_check; // reserved for future time-based triggers
                let _ = maybe_compact(&state, segments_dir, tmp_dir, manifest_path, &cfg);
                last_check = SystemTime::now();
                let _ = last_check;
            }
        }
    }
}

fn maybe_compact(
    state: &Arc<RwLock<StoreState>>,
    segments_dir: &Path,
    tmp_dir: &Path,
    manifest_path: &Path,
    cfg: &SegmentStoreConfig,
) -> Result<(), DbError> {
    let l0_count = {
        let st = state.read()?;
        st.active.iter().filter(|s| s.rec.level == 0).count()
    };
    if l0_count >= cfg.l0_compaction_trigger_segment_count {
        let _ = compact_l0_once(state, segments_dir, tmp_dir, manifest_path)?;
    }
    Ok(())
}

fn compact_l0_once(
    state: &Arc<RwLock<StoreState>>,
    segments_dir: &Path,
    tmp_dir: &Path,
    manifest_path: &Path,
) -> Result<CompactionStats, DbError> {
    // Select L0 segments to compact.
    let (to_compact, delete_before) = {
        let st = state.read()?;
        let l0: Vec<_> = st.active.iter().filter(|s| s.rec.level == 0).cloned().collect();
        (l0, st.manifest.delete_before)
    };

    if to_compact.len() < 2 {
        reap_obsolete(state)?;
        return Ok(CompactionStats { input_segments: 0, output_segments: 0 });
    }

    // Build merged rows by series.
    let mut merged: HashMap<String, Vec<Row>> = HashMap::new();
    let mut max_seq = 0u64;

    for seg in &to_compact {
        max_seq = max_seq.max(seg.rec.max_seq);

        for (series, meta) in &seg.rec.series {
            let rows = read_series_all_rows(&seg.path, meta)?;
            let entry = merged.entry(series.clone()).or_insert_with(Vec::new);
            entry.extend(rows);
        }
    }

    // Apply retention tombstone watermark.
    let delete_before = delete_before.unwrap_or(0);
    for rows in merged.values_mut() {
        rows.retain(|r| r.timestamp >= delete_before);
        rows.sort_unstable_by(|a, b| (a.timestamp, a.seq).cmp(&(b.timestamp, b.seq)));
    }
    merged.retain(|_, rows| !rows.is_empty());

    // Allocate new id and write output segment.
    let (new_id, created_at) = {
        let mut st = state.write()?;
        let id = st.manifest.next_segment_id;
        st.manifest.next_segment_id = st.manifest.next_segment_id.saturating_add(1);
        (id, now_ns())
    };

    let file_name = format!("seg_{:020}_l1.seg", new_id);
    let final_path = segments_dir.join(&file_name);
    let tmp_path = tmp_dir.join(format!("{}.tmp", &file_name));

    let new_rec = write_segment_file(
        &tmp_path,
        &final_path,
        new_id,
        1,
        created_at,
        Some(delete_before),
        merged,
    )?;

    // Install: remove old L0 from active list, add new L1, persist manifest.
    {
        let mut st = state.write()?;
        let old_ids: std::collections::HashSet<u64> = to_compact.iter().map(|s| s.rec.id).collect();

        // Update manifest records.
        st.manifest.segments.retain(|r| !old_ids.contains(&r.id));
        st.manifest.segments.push(new_rec.clone());
        write_manifest_atomic(manifest_path, tmp_dir, &st.manifest)?;

        // Update active list.
        let mut new_active = Vec::with_capacity(st.active.len() + 1);
        let mut new_obsolete = Vec::new();
        for seg in st.active.drain(..) {
            if old_ids.contains(&seg.rec.id) {
                new_obsolete.push(seg);
            } else {
                new_active.push(seg);
            }
        }
        st.obsolete.extend(new_obsolete);
        new_active.push(Arc::new(Segment { rec: new_rec.clone(), path: final_path }));
        st.active = new_active;
    }

    reap_obsolete(state)?;
    Ok(CompactionStats { input_segments: to_compact.len(), output_segments: 1 })
}

fn reap_obsolete(state: &Arc<RwLock<StoreState>>) -> Result<(), DbError> {
    let mut to_delete: Vec<PathBuf> = Vec::new();
    {
        let mut st = state.write()?;
        let mut keep = Vec::new();
        for seg in st.obsolete.drain(..) {
            if Arc::strong_count(&seg) == 1 {
                to_delete.push(seg.path.clone());
            } else {
                keep.push(seg);
            }
        }
        st.obsolete = keep;
    }

    for p in to_delete {
        let _ = fs::remove_file(&p);
    }
    Ok(())
}

fn write_segment_file(
    tmp_path: &Path,
    final_path: &Path,
    id: u64,
    level: u8,
    created_at: Timestamp,
    delete_before: Option<Timestamp>,
    rows_by_series: HashMap<String, Vec<Row>>,
) -> Result<SegmentRecord, DbError> {
    // Build series blocks in a deterministic order.
    let mut series_names: Vec<String> = rows_by_series.keys().cloned().collect();
    series_names.sort();

    // Track segment stats.
    let mut seg_min_ts = Timestamp::MAX;
    let mut seg_max_ts = 0u64;
    let mut seg_max_seq = 0u64;
    let mut series_meta: BTreeMap<String, SeriesBlockMeta> = BTreeMap::new();

    // Create tmp file
    let file = OpenOptions::new().create(true).write(true).truncate(true).open(tmp_path)?;
    let mut w = BufWriter::new(file);

    // Header
    w.write_all(SEG_MAGIC)?;
    w.write_all(&SEG_VERSION.to_le_bytes())?;
    w.write_all(&id.to_le_bytes())?;
    w.write_all(&[level])?;
    w.write_all(&created_at.to_le_bytes())?;

    // max_seq placeholder (we’ll fill it after computing)
    let max_seq_pos = w.stream_position()?;
    w.write_all(&0u64.to_le_bytes())?;

    // delete_before watermark persisted in the segment header for self-description (optional).
    let db = delete_before.unwrap_or(0);
    w.write_all(&db.to_le_bytes())?;

    // Series blocks
    for series in &series_names {
        let rows = rows_by_series.get(series).expect("series exists");
        if rows.is_empty() {
            continue;
        }

        let block_offset = w.stream_position()?;
        let block_bytes = encode_series_block(rows)?;
        let mut hasher = Crc32::new();
        hasher.update(&block_bytes);
        let crc32 = hasher.finalize();

        w.write_all(&block_bytes)?;
        let block_len = block_bytes.len() as u64;

        let row_count = rows.len() as u32;
        let min_ts = rows.first().unwrap().timestamp;
        let max_ts = rows.last().unwrap().timestamp;
        let max_seq = rows.iter().map(|r| r.seq).max().unwrap_or(0);

        seg_min_ts = seg_min_ts.min(min_ts);
        seg_max_ts = seg_max_ts.max(max_ts);
        seg_max_seq = seg_max_seq.max(max_seq);

        series_meta.insert(
            series.clone(),
            SeriesBlockMeta {
                offset: block_offset,
                len: block_len,
                row_count,
                min_ts,
                max_ts,
                crc32,
            },
        );
    }

    if series_meta.is_empty() {
        return Err(DbError::Internal("Refusing to write an empty segment".to_string()));
    }

    // Index
    let index_offset = w.stream_position()?;
    let mut index_buf = Vec::new();
    write_u32(&mut index_buf, series_meta.len() as u32);
    for (name, meta) in &series_meta {
        write_string(&mut index_buf, name);
        write_u64(&mut index_buf, meta.offset);
        write_u64(&mut index_buf, meta.len);
        write_u32(&mut index_buf, meta.row_count);
        write_u64(&mut index_buf, meta.min_ts);
        write_u64(&mut index_buf, meta.max_ts);
        write_u32(&mut index_buf, meta.crc32);
    }
    let index_len = index_buf.len() as u64;
    w.write_all(&index_buf)?;

    // Footer CRC covers (index_offset,index_len)
    let mut footer_hasher = Crc32::new();
    footer_hasher.update(&index_offset.to_le_bytes());
    footer_hasher.update(&index_len.to_le_bytes());
    let footer_crc = footer_hasher.finalize();

    w.write_all(SEG_FOOTER_MAGIC)?;
    w.write_all(&index_offset.to_le_bytes())?;
    w.write_all(&index_len.to_le_bytes())?;
    w.write_all(&footer_crc.to_le_bytes())?;

    // Backpatch max_seq.
    w.flush()?;
    let mut f = w.into_inner().map_err(|e| DbError::Io(e.into_error()))?;
    f.seek(SeekFrom::Start(max_seq_pos))?;
    f.write_all(&seg_max_seq.to_le_bytes())?;
    f.flush()?;
    f.sync_data()?;

    // Atomic install: rename tmp -> final.
    fs::rename(tmp_path, final_path)?;
    sync_parent_dir(final_path)?;

    let size = fs::metadata(final_path)?.len();
    let rec = SegmentRecord {
        id,
        level,
        created_at,
        max_seq: seg_max_seq,
        min_ts: seg_min_ts,
        max_ts: seg_max_ts,
        file_name: final_path
            .file_name()
            .ok_or_else(|| DbError::Internal("Invalid segment filename".to_string()))?
            .to_string_lossy()
            .into_owned(),
        series: series_meta,
    };

    // Sanity: index can be read back.
    let _ = load_segment_index(final_path)?;
    let _ = size;
    Ok(rec)
}

fn encode_series_block(rows: &[Row]) -> Result<Vec<u8>, DbError> {
    let row_count = rows.len();
    if row_count > (u32::MAX as usize) {
        return Err(DbError::Internal("Series block too large".to_string()));
    }

    let mut buf = Vec::new();
    buf.extend_from_slice(SER_BLOCK_MAGIC);
    write_u32(&mut buf, row_count as u32);

    // Columns: seq, ts, value
    for r in rows {
        write_u64(&mut buf, r.seq);
    }
    for r in rows {
        write_u64(&mut buf, r.timestamp);
    }
    for r in rows {
        write_f64(&mut buf, r.value);
    }

    // Tags column: offsets + blob.
    let mut offsets: Vec<u32> = Vec::with_capacity(row_count + 1);
    offsets.push(0);
    let mut tags_blob: Vec<u8> = Vec::new();
    for r in rows {
        let enc = bincode::serialize(&r.tags).map_err(|e| DbError::Serialization(e.to_string()))?;
        let next = offsets
            .last()
            .copied()
            .unwrap_or(0)
            .checked_add(enc.len() as u32)
            .ok_or_else(|| DbError::Internal("Tags blob overflow".to_string()))?;
        tags_blob.extend_from_slice(&enc);
        offsets.push(next);
    }

    // offsets
    for off in offsets {
        write_u32(&mut buf, off);
    }
    write_u32(&mut buf, tags_blob.len() as u32);
    buf.extend_from_slice(&tags_blob);
    Ok(buf)
}

fn read_series_range(
    path: &Path,
    meta: &SeriesBlockMeta,
    time_range: std::ops::Range<Timestamp>,
    tag_filter: Option<&TagSet>,
    delete_before: Timestamp,
) -> Result<Vec<(Timestamp, Value)>, DbError> {
    let mut f = File::open(path)?;
    f.seek(SeekFrom::Start(meta.offset))?;
    let mut block = vec![0u8; meta.len as usize];
    f.read_exact(&mut block)?;

    let mut hasher = Crc32::new();
    hasher.update(&block);
    let actual = hasher.finalize();
    if actual != meta.crc32 {
        return Err(DbError::Corruption {
            details: format!("Segment block CRC mismatch in {:?}", path),
            series: None,
            timestamp: None,
        });
    }

    let mut cur = std::io::Cursor::new(block);
    let mut magic = [0u8; 8];
    cur.read_exact(&mut magic)?;
    if &magic != SER_BLOCK_MAGIC {
        return Err(DbError::Corruption {
            details: format!("Bad series block magic in {:?}", path),
            series: None,
            timestamp: None,
        });
    }
    let row_count = read_u32(&mut cur)? as usize;

    // seq column (unused for query output, but still parsed to advance cursor)
    let mut _seqs = Vec::with_capacity(row_count);
    for _ in 0..row_count {
        _seqs.push(read_u64(&mut cur)?);
    }

    let mut timestamps = Vec::with_capacity(row_count);
    for _ in 0..row_count {
        timestamps.push(read_u64(&mut cur)?);
    }
    let mut values = Vec::with_capacity(row_count);
    for _ in 0..row_count {
        values.push(read_f64(&mut cur)?);
    }

    let mut offsets = Vec::with_capacity(row_count + 1);
    for _ in 0..(row_count + 1) {
        offsets.push(read_u32(&mut cur)?);
    }
    let tags_len = read_u32(&mut cur)? as usize;
    let mut tags_blob = vec![0u8; tags_len];
    cur.read_exact(&mut tags_blob)?;

    let start_idx = timestamps.partition_point(|&ts| ts < time_range.start.max(delete_before));
    let end_idx = timestamps.partition_point(|&ts| ts < time_range.end);
    if start_idx >= end_idx {
        return Ok(Vec::new());
    }

    let mut out = Vec::new();
    for i in start_idx..end_idx {
        if let Some(filter) = tag_filter {
            let s = offsets[i] as usize;
            let e = offsets[i + 1] as usize;
            let tags: TagSet = bincode::deserialize(&tags_blob[s..e])
                .map_err(|e| DbError::Serialization(e.to_string()))?;
            if !check_tags(&tags, filter) {
                continue;
            }
        }
        out.push((timestamps[i], values[i]));
    }
    Ok(out)
}

fn read_series_all_rows(path: &Path, meta: &SeriesBlockMeta) -> Result<Vec<Row>, DbError> {
    let mut f = File::open(path)?;
    f.seek(SeekFrom::Start(meta.offset))?;
    let mut block = vec![0u8; meta.len as usize];
    f.read_exact(&mut block)?;

    let mut hasher = Crc32::new();
    hasher.update(&block);
    let actual = hasher.finalize();
    if actual != meta.crc32 {
        return Err(DbError::Corruption {
            details: format!("Segment block CRC mismatch in {:?}", path),
            series: None,
            timestamp: None,
        });
    }

    let mut cur = std::io::Cursor::new(block);
    let mut magic = [0u8; 8];
    cur.read_exact(&mut magic)?;
    if &magic != SER_BLOCK_MAGIC {
        return Err(DbError::Corruption {
            details: format!("Bad series block magic in {:?}", path),
            series: None,
            timestamp: None,
        });
    }
    let row_count = read_u32(&mut cur)? as usize;

    let mut seqs = Vec::with_capacity(row_count);
    for _ in 0..row_count {
        seqs.push(read_u64(&mut cur)?);
    }
    let mut timestamps = Vec::with_capacity(row_count);
    for _ in 0..row_count {
        timestamps.push(read_u64(&mut cur)?);
    }
    let mut values = Vec::with_capacity(row_count);
    for _ in 0..row_count {
        values.push(read_f64(&mut cur)?);
    }

    let mut offsets = Vec::with_capacity(row_count + 1);
    for _ in 0..(row_count + 1) {
        offsets.push(read_u32(&mut cur)?);
    }
    let tags_len = read_u32(&mut cur)? as usize;
    let mut tags_blob = vec![0u8; tags_len];
    cur.read_exact(&mut tags_blob)?;

    let mut out = Vec::with_capacity(row_count);
    for i in 0..row_count {
        let s = offsets[i] as usize;
        let e = offsets[i + 1] as usize;
        let tags: TagSet = bincode::deserialize(&tags_blob[s..e])
            .map_err(|e| DbError::Serialization(e.to_string()))?;
        out.push(Row {
            seq: seqs[i],
            timestamp: timestamps[i],
            value: values[i],
            tags,
        });
    }
    Ok(out)
}

fn load_segment_index(path: &Path) -> Result<BTreeMap<String, SeriesBlockMeta>, DbError> {
    let mut f = File::open(path)?;
    // Validate header magic.
    let mut magic = [0u8; 8];
    f.read_exact(&mut magic)?;
    if &magic != SEG_MAGIC {
        return Err(DbError::Corruption {
            details: format!("Bad segment magic in {:?}", path),
            series: None,
            timestamp: None,
        });
    }
    // version
    let version = read_u32(&mut f)?;
    if version != SEG_VERSION {
        return Err(DbError::Corruption {
            details: format!("Unsupported segment version {} in {:?}", version, path),
            series: None,
            timestamp: None,
        });
    }

    // Footer
    let file_len = f.metadata()?.len();
    if file_len < FOOTER_LEN {
        return Err(DbError::Corruption {
            details: format!("Truncated segment file {:?}", path),
            series: None,
            timestamp: None,
        });
    }
    f.seek(SeekFrom::End(-(FOOTER_LEN as i64)))?;
    let mut footer_magic = [0u8; 8];
    f.read_exact(&mut footer_magic)?;
    if &footer_magic != SEG_FOOTER_MAGIC {
        return Err(DbError::Corruption {
            details: format!("Bad segment footer magic in {:?}", path),
            series: None,
            timestamp: None,
        });
    }
    let index_offset = read_u64(&mut f)?;
    let index_len = read_u64(&mut f)?;
    let footer_crc = read_u32(&mut f)?;

    let mut footer_hasher = Crc32::new();
    footer_hasher.update(&index_offset.to_le_bytes());
    footer_hasher.update(&index_len.to_le_bytes());
    if footer_hasher.finalize() != footer_crc {
        return Err(DbError::Corruption {
            details: format!("Segment footer CRC mismatch in {:?}", path),
            series: None,
            timestamp: None,
        });
    }

    // Index
    f.seek(SeekFrom::Start(index_offset))?;
    let mut index_bytes = vec![0u8; index_len as usize];
    f.read_exact(&mut index_bytes)?;
    let mut cur = std::io::Cursor::new(index_bytes);

    let series_count = read_u32(&mut cur)? as usize;
    let mut out = BTreeMap::new();
    for _ in 0..series_count {
        let name = read_string(&mut cur)?;
        let offset = read_u64(&mut cur)?;
        let len = read_u64(&mut cur)?;
        let row_count = read_u32(&mut cur)?;
        let min_ts = read_u64(&mut cur)?;
        let max_ts = read_u64(&mut cur)?;
        let crc32 = read_u32(&mut cur)?;
        out.insert(
            name,
            SeriesBlockMeta {
                offset,
                len,
                row_count,
                min_ts,
                max_ts,
                crc32,
            },
        );
    }
    Ok(out)
}

fn read_manifest(path: &Path) -> Result<Manifest, DbError> {
    let mut f = File::open(path)?;
    let mut magic = [0u8; 8];
    f.read_exact(&mut magic)?;
    if &magic != MANIFEST_MAGIC {
        return Err(DbError::Corruption {
            details: format!("Bad manifest magic in {:?}", path),
            series: None,
            timestamp: None,
        });
    }
    let version = read_u32(&mut f)?;
    if version != MANIFEST_VERSION {
        return Err(DbError::Corruption {
            details: format!("Unsupported manifest version {} in {:?}", version, path),
            series: None,
            timestamp: None,
        });
    }
    let len = read_u64(&mut f)? as usize;
    let crc = read_u32(&mut f)?;
    let mut buf = vec![0u8; len];
    f.read_exact(&mut buf)?;
    let mut hasher = Crc32::new();
    hasher.update(&buf);
    if hasher.finalize() != crc {
        return Err(DbError::Corruption {
            details: format!("Manifest CRC mismatch in {:?}", path),
            series: None,
            timestamp: None,
        });
    }
    let manifest: Manifest = bincode::deserialize(&buf).map_err(|e| DbError::Serialization(e.to_string()))?;
    Ok(manifest)
}

fn write_manifest_atomic(path: &Path, tmp_dir: &Path, manifest: &Manifest) -> Result<(), DbError> {
    let bytes = bincode::serialize(manifest).map_err(|e| DbError::Serialization(e.to_string()))?;
    let mut hasher = Crc32::new();
    hasher.update(&bytes);
    let crc = hasher.finalize();

    let tmp_path = tmp_dir.join("MANIFEST.bin.tmp");
    let mut w = BufWriter::new(OpenOptions::new().create(true).write(true).truncate(true).open(&tmp_path)?);
    w.write_all(MANIFEST_MAGIC)?;
    w.write_all(&MANIFEST_VERSION.to_le_bytes())?;
    w.write_all(&(bytes.len() as u64).to_le_bytes())?;
    w.write_all(&crc.to_le_bytes())?;
    w.write_all(&bytes)?;
    w.flush()?;
    w.get_ref().sync_data()?;
    drop(w);

    fs::rename(&tmp_path, path)?;
    sync_parent_dir(path)?;
    Ok(())
}

fn sync_parent_dir(path: &Path) -> Result<(), DbError> {
    let parent = path.parent().ok_or_else(|| DbError::Internal("Missing parent dir".to_string()))?;
    let dir = File::open(parent)?;
    dir.sync_data()?;
    Ok(())
}

#[inline]
fn check_tags(point_tags: &TagSet, filter_tags: &TagSet) -> bool {
    if point_tags.len() < filter_tags.len() {
        return false;
    }
    filter_tags
        .iter()
        .all(|(key, value)| point_tags.get(key) == Some(value))
}

// --- binary helpers ---

fn write_u32(buf: &mut Vec<u8>, v: u32) {
    buf.extend_from_slice(&v.to_le_bytes());
}
fn write_u64(buf: &mut Vec<u8>, v: u64) {
    buf.extend_from_slice(&v.to_le_bytes());
}
fn write_f64(buf: &mut Vec<u8>, v: f64) {
    buf.extend_from_slice(&v.to_le_bytes());
}
fn write_string(buf: &mut Vec<u8>, s: &str) {
    write_u32(buf, s.len() as u32);
    buf.extend_from_slice(s.as_bytes());
}

fn read_u32<R: Read>(r: &mut R) -> Result<u32, DbError> {
    let mut b = [0u8; 4];
    r.read_exact(&mut b)?;
    Ok(u32::from_le_bytes(b))
}
fn read_u64<R: Read>(r: &mut R) -> Result<u64, DbError> {
    let mut b = [0u8; 8];
    r.read_exact(&mut b)?;
    Ok(u64::from_le_bytes(b))
}
fn read_f64<R: Read>(r: &mut R) -> Result<f64, DbError> {
    let mut b = [0u8; 8];
    r.read_exact(&mut b)?;
    Ok(f64::from_le_bytes(b))
}
fn read_string<R: Read>(r: &mut R) -> Result<String, DbError> {
    let len = read_u32(r)? as usize;
    let mut b = vec![0u8; len];
    r.read_exact(&mut b)?;
    String::from_utf8(b).map_err(|e| DbError::Internal(format!("Invalid UTF-8: {}", e)))
}