use std::fs;
use std::hash::Hash;
use std::path::Path;

use zerocopy::FromBytes;

use crate::Key;

use crate::const_map::MapEntry;
use crate::disk_loc::DiskLoc;
use crate::entry::{EntryHeader, compute_crc32, entry_size};
use crate::error::{DbError, DbResult};
use crate::hint;
use crate::io::direct;
use crate::map_index::ShardIndex;
use crate::skiplist::SkipList;
#[cfg(feature = "var-collections")]
use crate::skiplist::node::VarNode;
use crate::skiplist::node::{ConstNode, SkipNode, random_height};
use crate::sync::{self, Mutex};

#[cfg(feature = "typed-tree")]
use crate::codec::Codec;
#[cfg(feature = "typed-tree")]
use crate::skiplist::node::{TypedData, TypedNode};

#[cfg(feature = "encryption")]
use std::sync::Arc;

#[cfg(feature = "encryption")]
use crate::crypto::PageCipher;

#[cfg(feature = "encryption")]
use crate::io::tags::{self, TagFile};

// ---------------------------------------------------------------------------
// Reader helper — dispatches to plain or encrypted pread_value
// ---------------------------------------------------------------------------

type ReadFn = dyn Fn(&std::fs::File, u64, usize) -> DbResult<Vec<u8>>;

fn plain_reader() -> Box<ReadFn> {
    Box::new(direct::pread_value)
}

#[cfg(feature = "encryption")]
fn encrypted_reader(
    cipher: &Arc<PageCipher>,
    tag_file: &Arc<TagFile>,
    file_id: u32,
) -> Box<ReadFn> {
    let cipher = cipher.clone();
    let tag_file = tag_file.clone();
    Box::new(move |file, offset, len| {
        direct::pread_value_encrypted(file, &tag_file, &cipher, file_id, offset, len)
    })
}

#[cfg(feature = "encryption")]
fn make_reader(
    cipher: &Option<Arc<PageCipher>>,
    dir: &Path,
    file_id: u32,
) -> DbResult<Box<ReadFn>> {
    if let Some(cipher) = cipher {
        let tp = tags::tags_path_for_data(&dir.join(format!("{file_id:06}.data")));
        if tp.exists() {
            let tag_file = Arc::new(TagFile::open_read(&tp)?);
            return Ok(encrypted_reader(cipher, &tag_file, file_id));
        }
        // Tag file missing for an encrypted database — data cannot be decrypted.
        // Return an error instead of silently reading encrypted bytes as plaintext.
        return Err(crate::error::DbError::EncryptionError(format!(
            "tag file missing for encrypted data file {file_id:06}.data"
        )));
    }
    Ok(plain_reader())
}

#[cfg(not(feature = "encryption"))]
fn make_reader(_dir: &Path, _file_id: u32) -> DbResult<Box<ReadFn>> {
    Ok(plain_reader())
}

// ---------------------------------------------------------------------------
// RecoveryOutcome types
// ---------------------------------------------------------------------------

/// Outcome of a recovery pass over all shards.
pub(crate) struct RecoveryOutcome {
    /// Maximum GSN seen across all shards.
    pub max_gsn: u64,
    /// One entry per shard whose active file needs tail adjustment.
    pub active_tails: Vec<ActiveTail>,
    /// Per-shard dead_bytes maps: (shard_idx, HashMap<file_id, dead_bytes>).
    pub shard_dead_bytes: Vec<(usize, DeadBytesMap)>,
}

/// Describes the tail of the active (last) data file in one shard.
pub(crate) struct ActiveTail {
    /// Index into the `shards` slice.
    pub shard_idx: usize,
    /// File ID of the active data file.
    pub file_id: u32,
    /// Last offset at which a valid (or zero-padding) boundary was found.
    pub last_valid_offset: u64,
}

// ---------------------------------------------------------------------------
// Per-key sequence tracker helpers
// ---------------------------------------------------------------------------

type RecoveryTracker = std::collections::HashMap<Vec<u8>, (u64, u32, u64)>;

/// Per-shard dead-bytes map: file_id -> dead bytes.
type DeadBytesMap = std::collections::HashMap<u32, u64>;

/// Return type for per-shard recovery threads.
type ShardRecoveryResult = DbResult<(u64, Option<ActiveTail>, usize, DeadBytesMap)>;

/// Update the per-key tracker and dead_bytes accounting.
///
/// Returns `true` if `entry` is newer and should be applied to the index.
/// Returns `false` if `entry` is stale (already superseded); the caller
/// should skip it.
fn tracker_update(
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
    key_bytes: &[u8],
    seq: u64,
    file_id: u32,
    entry_total_size: u64,
) -> bool {
    if let Some(val) = tracker.get_mut(key_bytes) {
        let (prev_seq, prev_file_id, prev_entry_size) = *val;
        if seq <= prev_seq {
            *dead_bytes.entry(file_id).or_insert(0) += entry_total_size;
            return false;
        }
        *dead_bytes.entry(prev_file_id).or_insert(0) += prev_entry_size;
        *val = (seq, file_id, entry_total_size);
        return true;
    }
    tracker.insert(key_bytes.to_vec(), (seq, file_id, entry_total_size));
    true
}

// ---------------------------------------------------------------------------
// Public entry points
// ---------------------------------------------------------------------------

/// Recover a ConstTree index by scanning all data files in each shard.
/// Uses hint files when available, falls back to full data scan otherwise.
/// Returns a [`RecoveryOutcome`] with the maximum GSN and per-shard active tails.
pub fn recover_const_tree<K: Key, const V: usize>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    index: &SkipList<ConstNode<K, V>>,
    hints: bool,
    #[cfg(feature = "encryption")] cipher: Option<Arc<PageCipher>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                #[cfg(feature = "encryption")]
                let cipher = cipher.clone();
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let mut local_max_gsn: u64 = 0;
                    let mut used_hints = false;
                    let mut active_tail: Option<ActiveTail> = None;
                    let guard = index.collector().enter();
                    let file_ids = scan_data_files(dir)?;
                    let hint_ids = if hints {
                        hint::scan_hint_files(dir)?
                    } else {
                        Vec::new()
                    };
                    let mut tracker: RecoveryTracker = std::collections::HashMap::new();
                    let mut dead_bytes: DeadBytesMap = std::collections::HashMap::new();

                    for file_id in &file_ids {
                        let data_path = dir.join(format!("{file_id:06}.data"));
                        let is_active = Some(file_id) == file_ids.last();

                        #[cfg(feature = "encryption")]
                        let read_fn = make_reader(&cipher, dir, *file_id)?;
                        #[cfg(not(feature = "encryption"))]
                        let read_fn = make_reader(dir, *file_id)?;

                        if hint_ids.contains(file_id) {
                            let hint_path = dir.join(format!("{file_id:06}.hint"));
                            if let Some(hint_data) = hint::read_hint_file(&hint_path)? {
                                if hint_data.len() % hint::hint_entry_size(size_of::<K>()) == 0 {
                                    let data_file = direct::open_bulk_read(&data_path)?;
                                    let file_len = data_file.metadata()?.len();
                                    recover_const_hint::<K, V>(
                                        shard_id,
                                        *file_id,
                                        &hint_data,
                                        &data_file,
                                        index,
                                        &guard,
                                        &mut local_max_gsn,
                                        &read_fn,
                                        &mut tracker,
                                        &mut dead_bytes,
                                    )?;
                                    direct::fadvise_dontneed(&data_file, 0, file_len);
                                    used_hints = true;
                                    continue;
                                }
                                tracing::warn!(
                                    shard_id,
                                    file_id,
                                    "hint file has unexpected size, falling back to full scan"
                                );
                            }
                        }

                        let last_valid = recover_const_full_scan::<K, V>(
                            shard_id,
                            *file_id,
                            &data_path,
                            index,
                            &guard,
                            &mut local_max_gsn,
                            &read_fn,
                            &mut tracker,
                            &mut dead_bytes,
                        )?;
                        if is_active {
                            active_tail = Some(ActiveTail {
                                shard_idx: i,
                                file_id: *file_id,
                                last_valid_offset: last_valid,
                            });
                        }
                    }

                    if let Some(&last_id) = file_ids.last() {
                        let _ = fs::remove_file(dir.join(format!("{last_id:06}.hint")));
                    }

                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((local_max_gsn, active_tail, i, dead_bytes))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    let mut max_gsn: u64 = 0;
    let mut active_tails: Vec<ActiveTail> = Vec::new();
    let mut shard_dead_bytes: Vec<(usize, DeadBytesMap)> = Vec::new();
    for result in results {
        let (shard_gsn, tail, shard_idx, dead) = result?;
        if shard_gsn > max_gsn {
            max_gsn = shard_gsn;
        }
        if let Some(t) = tail {
            active_tails.push(t);
        }
        if !dead.is_empty() {
            shard_dead_bytes.push((shard_idx, dead));
        }
    }

    Ok(RecoveryOutcome {
        max_gsn,
        active_tails,
        shard_dead_bytes,
    })
}

#[cfg(feature = "var-collections")]
/// Recover a VarTree index by scanning all data files in each shard.
pub fn recover_var_tree<K: Key>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    index: &SkipList<VarNode<K>>,
    hints: bool,
    #[cfg(feature = "encryption")] cipher: Option<Arc<PageCipher>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                #[cfg(feature = "encryption")]
                let cipher = cipher.clone();
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let mut local_max_gsn: u64 = 0;
                    let mut used_hints = false;
                    let mut active_tail: Option<ActiveTail> = None;
                    let guard = index.collector().enter();
                    let file_ids = scan_data_files(dir)?;
                    let hint_ids = if hints {
                        hint::scan_hint_files(dir)?
                    } else {
                        Vec::new()
                    };
                    let mut tracker: RecoveryTracker = std::collections::HashMap::new();
                    let mut dead_bytes: DeadBytesMap = std::collections::HashMap::new();

                    for file_id in &file_ids {
                        let data_path = dir.join(format!("{file_id:06}.data"));
                        let is_active = Some(file_id) == file_ids.last();

                        #[cfg(feature = "encryption")]
                        let read_fn = make_reader(&cipher, dir, *file_id)?;
                        #[cfg(not(feature = "encryption"))]
                        let read_fn = make_reader(dir, *file_id)?;

                        if hint_ids.contains(file_id) {
                            let hint_path = dir.join(format!("{file_id:06}.hint"));
                            if let Some(hint_data) = hint::read_hint_file(&hint_path)? {
                                if hint_data.len() % hint::hint_entry_size(size_of::<K>()) == 0 {
                                    recover_var_hint::<K>(
                                        shard_id,
                                        *file_id,
                                        &hint_data,
                                        index,
                                        &guard,
                                        &mut local_max_gsn,
                                        &mut tracker,
                                        &mut dead_bytes,
                                    )?;
                                    used_hints = true;
                                    continue;
                                }
                                tracing::warn!(
                                    shard_id,
                                    file_id,
                                    "hint file has unexpected size, falling back to full scan"
                                );
                            }
                        }

                        let last_valid = recover_var_full_scan::<K>(
                            shard_id,
                            *file_id,
                            &data_path,
                            index,
                            &guard,
                            &mut local_max_gsn,
                            &read_fn,
                            &mut tracker,
                            &mut dead_bytes,
                        )?;
                        if is_active {
                            active_tail = Some(ActiveTail {
                                shard_idx: i,
                                file_id: *file_id,
                                last_valid_offset: last_valid,
                            });
                        }
                    }

                    if let Some(&last_id) = file_ids.last() {
                        let _ = fs::remove_file(dir.join(format!("{last_id:06}.hint")));
                    }

                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((local_max_gsn, active_tail, i, dead_bytes))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    let mut max_gsn: u64 = 0;
    let mut active_tails: Vec<ActiveTail> = Vec::new();
    let mut shard_dead_bytes: Vec<(usize, DeadBytesMap)> = Vec::new();
    for result in results {
        let (shard_gsn, tail, shard_idx, dead) = result?;
        if shard_gsn > max_gsn {
            max_gsn = shard_gsn;
        }
        if let Some(t) = tail {
            active_tails.push(t);
        }
        if !dead.is_empty() {
            shard_dead_bytes.push((shard_idx, dead));
        }
    }

    Ok(RecoveryOutcome {
        max_gsn,
        active_tails,
        shard_dead_bytes,
    })
}

/// Recover a TypedTree index by scanning all data files in each shard.
/// Uses hint files when available, falls back to full data scan otherwise.
/// Returns a [`RecoveryOutcome`] with the maximum GSN and per-shard active tails.
#[cfg(feature = "typed-tree")]
pub fn recover_typed_tree<K: Key, T: Send + Sync, C: Codec<T> + Sync>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    index: &SkipList<TypedNode<K, T>>,
    codec: &C,
    hints: bool,
    #[cfg(feature = "encryption")] cipher: Option<Arc<PageCipher>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                #[cfg(feature = "encryption")]
                let cipher = cipher.clone();
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let mut local_max_gsn: u64 = 0;
                    let mut used_hints = false;
                    let mut active_tail: Option<ActiveTail> = None;
                    let guard = index.collector().enter();
                    let file_ids = scan_data_files(dir)?;
                    let hint_ids = if hints {
                        hint::scan_hint_files(dir)?
                    } else {
                        Vec::new()
                    };
                    let mut tracker: RecoveryTracker = std::collections::HashMap::new();
                    let mut dead_bytes: DeadBytesMap = std::collections::HashMap::new();

                    for file_id in &file_ids {
                        let data_path = dir.join(format!("{file_id:06}.data"));
                        let is_active = Some(file_id) == file_ids.last();

                        #[cfg(feature = "encryption")]
                        let read_fn = make_reader(&cipher, dir, *file_id)?;
                        #[cfg(not(feature = "encryption"))]
                        let read_fn = make_reader(dir, *file_id)?;

                        if hint_ids.contains(file_id) {
                            let hint_path = dir.join(format!("{file_id:06}.hint"));
                            if let Some(hint_data) = hint::read_hint_file(&hint_path)? {
                                if hint_data.len() % hint::hint_entry_size(size_of::<K>()) == 0 {
                                    let data_file = direct::open_bulk_read(&data_path)?;
                                    let file_len = data_file.metadata()?.len();
                                    recover_typed_hint::<K, T, C>(
                                        shard_id,
                                        *file_id,
                                        &hint_data,
                                        &data_file,
                                        index,
                                        codec,
                                        &guard,
                                        &mut local_max_gsn,
                                        &read_fn,
                                        &mut tracker,
                                        &mut dead_bytes,
                                    )?;
                                    direct::fadvise_dontneed(&data_file, 0, file_len);
                                    used_hints = true;
                                    continue;
                                }
                                tracing::warn!(
                                    shard_id,
                                    file_id,
                                    "hint file has unexpected size, falling back to full scan"
                                );
                            }
                        }

                        let last_valid = recover_typed_full_scan::<K, T, C>(
                            shard_id,
                            *file_id,
                            &data_path,
                            index,
                            codec,
                            &guard,
                            &mut local_max_gsn,
                            &read_fn,
                            &mut tracker,
                            &mut dead_bytes,
                        )?;
                        if is_active {
                            active_tail = Some(ActiveTail {
                                shard_idx: i,
                                file_id: *file_id,
                                last_valid_offset: last_valid,
                            });
                        }
                    }

                    if let Some(&last_id) = file_ids.last() {
                        let _ = fs::remove_file(dir.join(format!("{last_id:06}.hint")));
                    }

                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((local_max_gsn, active_tail, i, dead_bytes))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    let mut max_gsn: u64 = 0;
    let mut active_tails: Vec<ActiveTail> = Vec::new();
    let mut shard_dead_bytes: Vec<(usize, DeadBytesMap)> = Vec::new();
    for result in results {
        let (shard_gsn, tail, shard_idx, dead) = result?;
        if shard_gsn > max_gsn {
            max_gsn = shard_gsn;
        }
        if let Some(t) = tail {
            active_tails.push(t);
        }
        if !dead.is_empty() {
            shard_dead_bytes.push((shard_idx, dead));
        }
    }

    Ok(RecoveryOutcome {
        max_gsn,
        active_tails,
        shard_dead_bytes,
    })
}

// ---------------------------------------------------------------------------
// Hint-based recovery (fast path)
// ---------------------------------------------------------------------------
#[allow(clippy::too_many_arguments)]
fn recover_const_hint<K: Key, const V: usize>(
    _shard_id: u8,
    file_id: u32,
    hint_data: &[u8],
    data_file: &std::fs::File,
    index: &SkipList<ConstNode<K, V>>,
    guard: &seize::LocalGuard<'_>,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<()> {
    for entry in hint::parse_hint_entries::<K>(hint_data) {
        let seq = entry.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        let total = entry_size(size_of::<K>(), entry.value_len) as u64;
        if !tracker_update(
            tracker,
            dead_bytes,
            entry.key.as_bytes(),
            seq,
            file_id,
            total,
        ) {
            continue;
        }

        if entry.is_tombstone() {
            index.remove(entry.key.as_bytes(), guard);
        } else {
            let value_bytes = read_fn(data_file, entry.value_offset, entry.value_len as usize)?;
            let expected_crc = compute_crc32(
                entry.gsn,
                entry.value_len,
                entry.key.as_bytes(),
                &value_bytes,
            );
            if expected_crc != entry.crc32 {
                return Err(DbError::CrcMismatch {
                    expected: expected_crc,
                    actual: entry.crc32,
                });
            }
            let value: [u8; V] = value_bytes
                .try_into()
                .map_err(|_| DbError::CorruptedEntry {
                    offset: entry.value_offset,
                })?;

            let disk = DiskLoc::new(file_id, entry.value_offset as u32, entry.value_len);

            let height = random_height();
            let node_ptr = ConstNode::alloc(entry.key, value, disk, height);
            match index.insert(node_ptr, guard) {
                crate::skiplist::InsertResult::Inserted => {}
                crate::skiplist::InsertResult::Exists(existing) => {
                    existing.write_data(disk, &value);
                    unsafe {
                        ConstNode::<K, V>::dealloc_node(node_ptr);
                    }
                }
            }
        }
    }
    Ok(())
}

#[cfg(feature = "var-collections")]
/// VarTree hint recovery — pure metadata, no disk reads for values.
#[allow(clippy::too_many_arguments)]
fn recover_var_hint<K: Key>(
    _shard_id: u8,
    file_id: u32,
    hint_data: &[u8],
    index: &SkipList<VarNode<K>>,
    guard: &seize::LocalGuard<'_>,
    max_gsn: &mut u64,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<()> {
    for entry in hint::parse_hint_entries::<K>(hint_data) {
        let seq = entry.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        let total = entry_size(size_of::<K>(), entry.value_len) as u64;
        if !tracker_update(
            tracker,
            dead_bytes,
            entry.key.as_bytes(),
            seq,
            file_id,
            total,
        ) {
            continue;
        }

        if entry.is_tombstone() {
            index.remove(entry.key.as_bytes(), guard);
        } else {
            let disk = DiskLoc::new(file_id, entry.value_offset as u32, entry.value_len);

            let height = random_height();
            let node_ptr = VarNode::alloc(entry.key, disk, height);
            match index.insert(node_ptr, guard) {
                crate::skiplist::InsertResult::Inserted => {}
                crate::skiplist::InsertResult::Exists(existing) => {
                    // Recovery is single-threaded and runs before the index is published to
                    // concurrent readers, so write_loc here is the only writer (no shard mutex).
                    existing.write_loc(disk);
                    unsafe {
                        VarNode::<K>::dealloc_node(node_ptr);
                    }
                }
            }
        }
    }
    Ok(())
}

#[cfg(feature = "typed-tree")]
#[allow(clippy::too_many_arguments)]
fn recover_typed_hint<K: Key, T: Send + Sync, C: Codec<T>>(
    _shard_id: u8,
    file_id: u32,
    hint_data: &[u8],
    data_file: &std::fs::File,
    index: &SkipList<TypedNode<K, T>>,
    codec: &C,
    guard: &seize::LocalGuard<'_>,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<()> {
    for entry in hint::parse_hint_entries::<K>(hint_data) {
        let seq = entry.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        let total = entry_size(size_of::<K>(), entry.value_len) as u64;
        if !tracker_update(
            tracker,
            dead_bytes,
            entry.key.as_bytes(),
            seq,
            file_id,
            total,
        ) {
            continue;
        }

        if entry.is_tombstone() {
            index.remove(entry.key.as_bytes(), guard);
        } else {
            let value_bytes = read_fn(data_file, entry.value_offset, entry.value_len as usize)?;
            let expected_crc = compute_crc32(
                entry.gsn,
                entry.value_len,
                entry.key.as_bytes(),
                &value_bytes,
            );
            if expected_crc != entry.crc32 {
                return Err(DbError::CrcMismatch {
                    expected: expected_crc,
                    actual: entry.crc32,
                });
            }
            let value: T =
                codec
                    .decode_from(&value_bytes)
                    .map_err(|_| DbError::CorruptedEntry {
                        offset: entry.value_offset,
                    })?;

            let disk = DiskLoc::new(file_id, entry.value_offset as u32, entry.value_len);

            let height = random_height();
            let node_ptr = TypedNode::alloc(entry.key, value, disk, height);
            match index.insert(node_ptr, guard) {
                crate::skiplist::InsertResult::Inserted => {}
                crate::skiplist::InsertResult::Exists(existing) => unsafe {
                    let new_data = (*node_ptr)
                        .data
                        .swap(std::ptr::null_mut(), std::sync::atomic::Ordering::Relaxed);
                    let old_data = existing.swap_data(new_data);
                    drop(Box::from_raw(old_data));
                    TypedNode::<K, T>::dealloc_node(node_ptr);
                },
            }
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Full data scan (fallback path)
// ---------------------------------------------------------------------------

#[allow(clippy::too_many_arguments)]
fn recover_const_full_scan<K: Key, const V: usize>(
    shard_id: u8,
    file_id: u32,
    data_path: &Path,
    index: &SkipList<ConstNode<K, V>>,
    guard: &seize::LocalGuard<'_>,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<u64> {
    let file = direct::open_bulk_read(data_path)?;
    let file_len = file.metadata()?.len();
    let mut offset: u64 = 0;

    while offset + size_of::<EntryHeader>() as u64 <= file_len {
        let header_bytes = match read_fn(&file, offset, size_of::<EntryHeader>()) {
            Ok(b) => b,
            Err(e) => {
                if matches!(&e, DbError::Io(io) if io.kind() == std::io::ErrorKind::UnexpectedEof) {
                    break;
                }
                return Err(e);
            }
        };
        let header = match EntryHeader::read_from_bytes(&header_bytes) {
            Ok(h) => h,
            Err(_) => break,
        };

        // Page padding sentinel: force-flush pads the partial page with zeros.
        // If we see an all-zero header AND the rest of the current page is also
        // zero, that's clean padding — skip to the next page boundary.
        if header.gsn == 0
            && header.value_len == 0
            && header.crc32 == 0
            && crate::entry::check_page_padding_at(
                &file,
                offset + size_of::<EntryHeader>() as u64,
                read_fn,
            )?
        {
            const PAGE_SIZE: u64 = 4096;
            let next_page =
                (offset + size_of::<EntryHeader>() as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
            if next_page >= file_len {
                break;
            }
            offset = next_page;
            continue;
        }

        let total = entry_size(size_of::<K>(), header.value_len);
        if offset + total > file_len {
            tracing::warn!(
                shard_id,
                file_id,
                offset,
                "truncating partial entry at end of file"
            );
            break;
        }

        let key_value_len = size_of::<K>() + header.value_len as usize;
        let key_value_bytes = read_fn(
            &file,
            offset + size_of::<EntryHeader>() as u64,
            key_value_len,
        )?;

        let key_bytes = &key_value_bytes[..size_of::<K>()];
        let value_bytes = &key_value_bytes[size_of::<K>()..];

        let expected_crc = compute_crc32(header.gsn, header.value_len, key_bytes, value_bytes);
        if expected_crc != header.crc32 {
            if offset + total > file_len {
                tracing::warn!(shard_id, file_id, offset, "CRC mismatch at end of file");
                break;
            }
            return Err(DbError::CrcMismatch {
                expected: expected_crc,
                actual: header.crc32,
            });
        }

        let seq = header.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        if !tracker_update(tracker, dead_bytes, key_bytes, seq, file_id, total) {
            offset += total;
            continue;
        }

        let key: K = K::from_bytes(key_bytes);

        if header.is_tombstone() {
            index.remove(key.as_bytes(), guard);
        } else {
            let value: [u8; V] = value_bytes
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset })?;
            let value_offset = offset + size_of::<EntryHeader>() as u64 + size_of::<K>() as u64;
            let disk = DiskLoc::new(file_id, value_offset as u32, header.value_len);

            let height = random_height();
            let node_ptr = ConstNode::alloc(key, value, disk, height);
            match index.insert(node_ptr, guard) {
                crate::skiplist::InsertResult::Inserted => {}
                crate::skiplist::InsertResult::Exists(existing) => {
                    existing.write_data(disk, &value);
                    unsafe {
                        ConstNode::<K, V>::dealloc_node(node_ptr);
                    }
                }
            }
        }

        offset += total;
    }
    direct::fadvise_dontneed(&file, 0, file_len);
    Ok(offset)
}

#[cfg(feature = "var-collections")]
#[allow(clippy::too_many_arguments)]
fn recover_var_full_scan<K: Key>(
    shard_id: u8,
    file_id: u32,
    data_path: &Path,
    index: &SkipList<VarNode<K>>,
    guard: &seize::LocalGuard<'_>,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<u64> {
    let file = direct::open_bulk_read(data_path)?;
    let file_len = file.metadata()?.len();
    let mut offset: u64 = 0;

    while offset + size_of::<EntryHeader>() as u64 <= file_len {
        let header_bytes = match read_fn(&file, offset, size_of::<EntryHeader>()) {
            Ok(b) => b,
            Err(e) => {
                if matches!(&e, DbError::Io(io) if io.kind() == std::io::ErrorKind::UnexpectedEof) {
                    break;
                }
                return Err(e);
            }
        };
        let header = match EntryHeader::read_from_bytes(&header_bytes) {
            Ok(h) => h,
            Err(_) => break,
        };

        if header.gsn == 0
            && header.value_len == 0
            && header.crc32 == 0
            && crate::entry::check_page_padding_at(
                &file,
                offset + size_of::<EntryHeader>() as u64,
                read_fn,
            )?
        {
            const PAGE_SIZE: u64 = 4096;
            let next_page =
                (offset + size_of::<EntryHeader>() as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
            if next_page >= file_len {
                break;
            }
            offset = next_page;
            continue;
        }

        let total = entry_size(size_of::<K>(), header.value_len);
        if offset + total > file_len {
            tracing::warn!(
                shard_id,
                file_id,
                offset,
                "truncating partial entry at end of file"
            );
            break;
        }

        let key_value_len = size_of::<K>() + header.value_len as usize;
        let key_value_bytes = read_fn(
            &file,
            offset + size_of::<EntryHeader>() as u64,
            key_value_len,
        )?;

        let key_bytes = &key_value_bytes[..size_of::<K>()];
        let value_bytes = &key_value_bytes[size_of::<K>()..];

        let expected_crc = compute_crc32(header.gsn, header.value_len, key_bytes, value_bytes);
        if expected_crc != header.crc32 {
            if offset + total > file_len {
                tracing::warn!(shard_id, file_id, offset, "CRC mismatch at end of file");
                break;
            }
            return Err(DbError::CrcMismatch {
                expected: expected_crc,
                actual: header.crc32,
            });
        }

        let seq = header.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        if !tracker_update(tracker, dead_bytes, key_bytes, seq, file_id, total) {
            offset += total;
            continue;
        }

        let key: K = K::from_bytes(key_bytes);

        if header.is_tombstone() {
            index.remove(key.as_bytes(), guard);
        } else {
            let value_offset = offset + size_of::<EntryHeader>() as u64 + size_of::<K>() as u64;
            let disk = DiskLoc::new(file_id, value_offset as u32, header.value_len);

            let height = random_height();
            let node_ptr = VarNode::alloc(key, disk, height);
            match index.insert(node_ptr, guard) {
                crate::skiplist::InsertResult::Inserted => {}
                crate::skiplist::InsertResult::Exists(existing) => {
                    // Recovery is single-threaded and runs before the index is published to
                    // concurrent readers, so write_loc here is the only writer (no shard mutex).
                    existing.write_loc(disk);
                    unsafe {
                        VarNode::<K>::dealloc_node(node_ptr);
                    }
                }
            }
        }

        offset += total;
    }
    direct::fadvise_dontneed(&file, 0, file_len);
    Ok(offset)
}

#[cfg(feature = "typed-tree")]
#[allow(clippy::too_many_arguments)]
fn recover_typed_full_scan<K: Key, T: Send + Sync, C: Codec<T>>(
    shard_id: u8,
    file_id: u32,
    data_path: &Path,
    index: &SkipList<TypedNode<K, T>>,
    codec: &C,
    guard: &seize::LocalGuard<'_>,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<u64> {
    let file = direct::open_bulk_read(data_path)?;
    let file_len = file.metadata()?.len();
    let mut offset: u64 = 0;

    while offset + size_of::<EntryHeader>() as u64 <= file_len {
        let header_bytes = match read_fn(&file, offset, size_of::<EntryHeader>()) {
            Ok(b) => b,
            Err(e) => {
                if matches!(&e, DbError::Io(io) if io.kind() == std::io::ErrorKind::UnexpectedEof) {
                    break;
                }
                return Err(e);
            }
        };
        let header = match EntryHeader::read_from_bytes(&header_bytes) {
            Ok(h) => h,
            Err(_) => break,
        };

        if header.gsn == 0
            && header.value_len == 0
            && header.crc32 == 0
            && crate::entry::check_page_padding_at(
                &file,
                offset + size_of::<EntryHeader>() as u64,
                read_fn,
            )?
        {
            const PAGE_SIZE: u64 = 4096;
            let next_page =
                (offset + size_of::<EntryHeader>() as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
            if next_page >= file_len {
                break;
            }
            offset = next_page;
            continue;
        }

        let total = entry_size(size_of::<K>(), header.value_len);
        if offset + total > file_len {
            tracing::warn!(
                shard_id,
                file_id,
                offset,
                "truncating partial entry at end of file"
            );
            break;
        }

        let key_value_len = size_of::<K>() + header.value_len as usize;
        let key_value_bytes = read_fn(
            &file,
            offset + size_of::<EntryHeader>() as u64,
            key_value_len,
        )?;

        let key_bytes = &key_value_bytes[..size_of::<K>()];
        let value_bytes = &key_value_bytes[size_of::<K>()..];

        let expected_crc = compute_crc32(header.gsn, header.value_len, key_bytes, value_bytes);
        if expected_crc != header.crc32 {
            if offset + total > file_len {
                tracing::warn!(shard_id, file_id, offset, "CRC mismatch at end of file");
                break;
            }
            return Err(DbError::CrcMismatch {
                expected: expected_crc,
                actual: header.crc32,
            });
        }

        let seq = header.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        if !tracker_update(tracker, dead_bytes, key_bytes, seq, file_id, total) {
            offset += total;
            continue;
        }

        let key: K = K::from_bytes(key_bytes);

        if header.is_tombstone() {
            index.remove(key.as_bytes(), guard);
        } else {
            let value: T = codec
                .decode_from(value_bytes)
                .map_err(|_| DbError::CorruptedEntry { offset })?;
            let value_offset = offset + size_of::<EntryHeader>() as u64 + size_of::<K>() as u64;
            let disk = DiskLoc::new(file_id, value_offset as u32, header.value_len);

            let height = random_height();
            let node_ptr = TypedNode::alloc(key, value, disk, height);
            match index.insert(node_ptr, guard) {
                crate::skiplist::InsertResult::Inserted => {}
                crate::skiplist::InsertResult::Exists(existing) => unsafe {
                    let new_data = (*node_ptr)
                        .data
                        .swap(std::ptr::null_mut(), std::sync::atomic::Ordering::Relaxed);
                    let old_data = existing.swap_data(new_data);
                    drop(Box::from_raw(old_data));
                    TypedNode::<K, T>::dealloc_node(node_ptr);
                },
            }
        }

        offset += total;
    }
    direct::fadvise_dontneed(&file, 0, file_len);
    Ok(offset)
}

// ---------------------------------------------------------------------------
// ConstMap / VarMap recovery
// ---------------------------------------------------------------------------

pub fn recover_const_map<K: Key + Send + Sync + Hash + Eq, const V: usize>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    indexes: &[Mutex<ShardIndex<K, MapEntry<V, DiskLoc>>>],
    hints: bool,
    #[cfg(feature = "encryption")] cipher: Option<Arc<PageCipher>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                let index = &indexes[i];
                #[cfg(feature = "encryption")]
                let cipher = cipher.clone();
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let mut local_max_gsn: u64 = 0;
                    let mut used_hints = false;
                    let mut active_tail: Option<ActiveTail> = None;
                    let mut map = sync::lock(index);
                    let file_ids = scan_data_files(dir)?;
                    let hint_ids = if hints {
                        hint::scan_hint_files(dir)?
                    } else {
                        Vec::new()
                    };
                    let mut tracker: RecoveryTracker = std::collections::HashMap::new();
                    let mut dead_bytes: DeadBytesMap = std::collections::HashMap::new();

                    for file_id in &file_ids {
                        let data_path = dir.join(format!("{file_id:06}.data"));
                        let is_active = Some(file_id) == file_ids.last();

                        #[cfg(feature = "encryption")]
                        let read_fn = make_reader(&cipher, dir, *file_id)?;
                        #[cfg(not(feature = "encryption"))]
                        let read_fn = make_reader(dir, *file_id)?;

                        if hint_ids.contains(file_id) {
                            let hint_path = dir.join(format!("{file_id:06}.hint"));
                            if let Some(hint_data) = hint::read_hint_file(&hint_path)? {
                                if hint_data.len() % hint::hint_entry_size(size_of::<K>()) == 0 {
                                    let data_file = direct::open_bulk_read(&data_path)?;
                                    let file_len = data_file.metadata()?.len();
                                    recover_const_map_hint::<K, V>(
                                        shard_id,
                                        *file_id,
                                        &hint_data,
                                        &data_file,
                                        &mut map,
                                        &mut local_max_gsn,
                                        &read_fn,
                                        &mut tracker,
                                        &mut dead_bytes,
                                    )?;
                                    direct::fadvise_dontneed(&data_file, 0, file_len);
                                    used_hints = true;
                                    continue;
                                }
                                tracing::warn!(
                                    shard_id,
                                    file_id,
                                    "hint file has unexpected size, falling back to full scan"
                                );
                            }
                        }

                        let last_valid = recover_const_map_full_scan::<K, V>(
                            shard_id,
                            *file_id,
                            &data_path,
                            &mut map,
                            &mut local_max_gsn,
                            &read_fn,
                            &mut tracker,
                            &mut dead_bytes,
                        )?;
                        if is_active {
                            active_tail = Some(ActiveTail {
                                shard_idx: i,
                                file_id: *file_id,
                                last_valid_offset: last_valid,
                            });
                        }
                    }

                    if let Some(&last_id) = file_ids.last() {
                        let _ = fs::remove_file(dir.join(format!("{last_id:06}.hint")));
                    }

                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((local_max_gsn, active_tail, i, dead_bytes))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    let mut max_gsn: u64 = 0;
    let mut active_tails: Vec<ActiveTail> = Vec::new();
    let mut shard_dead_bytes: Vec<(usize, DeadBytesMap)> = Vec::new();
    for result in results {
        let (shard_gsn, tail, shard_idx, dead) = result?;
        if shard_gsn > max_gsn {
            max_gsn = shard_gsn;
        }
        if let Some(t) = tail {
            active_tails.push(t);
        }
        if !dead.is_empty() {
            shard_dead_bytes.push((shard_idx, dead));
        }
    }

    Ok(RecoveryOutcome {
        max_gsn,
        active_tails,
        shard_dead_bytes,
    })
}

#[cfg(feature = "var-collections")]
pub fn recover_var_map<K: Key + Send + Sync + Hash + Eq>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    indexes: &[Mutex<ShardIndex<K, DiskLoc>>],
    hints: bool,
    #[cfg(feature = "encryption")] cipher: Option<Arc<PageCipher>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                let index = &indexes[i];
                #[cfg(feature = "encryption")]
                let cipher = cipher.clone();
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let mut local_max_gsn: u64 = 0;
                    let mut used_hints = false;
                    let mut active_tail: Option<ActiveTail> = None;
                    let mut map = sync::lock(index);
                    let file_ids = scan_data_files(dir)?;
                    let hint_ids = if hints {
                        hint::scan_hint_files(dir)?
                    } else {
                        Vec::new()
                    };
                    let mut tracker: RecoveryTracker = std::collections::HashMap::new();
                    let mut dead_bytes: DeadBytesMap = std::collections::HashMap::new();

                    for file_id in &file_ids {
                        let data_path = dir.join(format!("{file_id:06}.data"));
                        let is_active = Some(file_id) == file_ids.last();

                        #[cfg(feature = "encryption")]
                        let read_fn = make_reader(&cipher, dir, *file_id)?;
                        #[cfg(not(feature = "encryption"))]
                        let read_fn = make_reader(dir, *file_id)?;

                        if hint_ids.contains(file_id) {
                            let hint_path = dir.join(format!("{file_id:06}.hint"));
                            if let Some(hint_data) = hint::read_hint_file(&hint_path)? {
                                if hint_data.len() % hint::hint_entry_size(size_of::<K>()) == 0 {
                                    recover_var_map_hint::<K>(
                                        shard_id,
                                        *file_id,
                                        &hint_data,
                                        &mut map,
                                        &mut local_max_gsn,
                                        &mut tracker,
                                        &mut dead_bytes,
                                    )?;
                                    used_hints = true;
                                    continue;
                                }
                                tracing::warn!(
                                    shard_id,
                                    file_id,
                                    "hint file has unexpected size, falling back to full scan"
                                );
                            }
                        }

                        let last_valid = recover_var_map_full_scan::<K>(
                            shard_id,
                            *file_id,
                            &data_path,
                            &mut map,
                            &mut local_max_gsn,
                            &read_fn,
                            &mut tracker,
                            &mut dead_bytes,
                        )?;
                        if is_active {
                            active_tail = Some(ActiveTail {
                                shard_idx: i,
                                file_id: *file_id,
                                last_valid_offset: last_valid,
                            });
                        }
                    }

                    if let Some(&last_id) = file_ids.last() {
                        let _ = fs::remove_file(dir.join(format!("{last_id:06}.hint")));
                    }

                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((local_max_gsn, active_tail, i, dead_bytes))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    let mut max_gsn: u64 = 0;
    let mut active_tails: Vec<ActiveTail> = Vec::new();
    let mut shard_dead_bytes: Vec<(usize, DeadBytesMap)> = Vec::new();
    for result in results {
        let (shard_gsn, tail, shard_idx, dead) = result?;
        if shard_gsn > max_gsn {
            max_gsn = shard_gsn;
        }
        if let Some(t) = tail {
            active_tails.push(t);
        }
        if !dead.is_empty() {
            shard_dead_bytes.push((shard_idx, dead));
        }
    }

    Ok(RecoveryOutcome {
        max_gsn,
        active_tails,
        shard_dead_bytes,
    })
}

// ---------------------------------------------------------------------------
// ConstMap hint-based recovery
// ---------------------------------------------------------------------------

#[allow(clippy::too_many_arguments)]
fn recover_const_map_hint<K: Key + Send + Sync + Hash + Eq, const V: usize>(
    _shard_id: u8,
    file_id: u32,
    hint_data: &[u8],
    data_file: &std::fs::File,
    map: &mut ShardIndex<K, crate::const_map::MapEntry<V, DiskLoc>>,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<()> {
    for entry in hint::parse_hint_entries::<K>(hint_data) {
        let seq = entry.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        let total = entry_size(size_of::<K>(), entry.value_len) as u64;
        if !tracker_update(
            tracker,
            dead_bytes,
            entry.key.as_bytes(),
            seq,
            file_id,
            total,
        ) {
            continue;
        }

        if entry.is_tombstone() {
            map.remove(&entry.key);
        } else {
            let value_bytes = read_fn(data_file, entry.value_offset, entry.value_len as usize)?;
            let expected_crc = compute_crc32(
                entry.gsn,
                entry.value_len,
                entry.key.as_bytes(),
                &value_bytes,
            );
            if expected_crc != entry.crc32 {
                return Err(DbError::CrcMismatch {
                    expected: expected_crc,
                    actual: entry.crc32,
                });
            }
            let value: [u8; V] = value_bytes
                .try_into()
                .map_err(|_| DbError::CorruptedEntry {
                    offset: entry.value_offset,
                })?;

            let disk = DiskLoc::new(file_id, entry.value_offset as u32, entry.value_len);

            map.upsert(entry.key, crate::const_map::MapEntry { loc: disk, value });
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// ConstMap full scan recovery
// ---------------------------------------------------------------------------

#[allow(clippy::too_many_arguments)]
fn recover_const_map_full_scan<K: Key + Send + Sync + Hash + Eq, const V: usize>(
    shard_id: u8,
    file_id: u32,
    data_path: &Path,
    map: &mut ShardIndex<K, crate::const_map::MapEntry<V, DiskLoc>>,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<u64> {
    let file = direct::open_bulk_read(data_path)?;
    let file_len = file.metadata()?.len();
    let mut offset: u64 = 0;

    while offset + size_of::<EntryHeader>() as u64 <= file_len {
        let header_bytes = match read_fn(&file, offset, size_of::<EntryHeader>()) {
            Ok(b) => b,
            Err(e) => {
                if matches!(&e, DbError::Io(io) if io.kind() == std::io::ErrorKind::UnexpectedEof) {
                    break;
                }
                return Err(e);
            }
        };
        let header = match EntryHeader::read_from_bytes(&header_bytes) {
            Ok(h) => h,
            Err(_) => break,
        };

        if header.gsn == 0
            && header.value_len == 0
            && header.crc32 == 0
            && crate::entry::check_page_padding_at(
                &file,
                offset + size_of::<EntryHeader>() as u64,
                read_fn,
            )?
        {
            const PAGE_SIZE: u64 = 4096;
            let next_page =
                (offset + size_of::<EntryHeader>() as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
            if next_page >= file_len {
                break;
            }
            offset = next_page;
            continue;
        }

        let total = entry_size(size_of::<K>(), header.value_len);
        if offset + total > file_len {
            tracing::warn!(
                shard_id,
                file_id,
                offset,
                "truncating partial entry at end of file"
            );
            break;
        }

        let key_value_len = size_of::<K>() + header.value_len as usize;
        let key_value_bytes = read_fn(
            &file,
            offset + size_of::<EntryHeader>() as u64,
            key_value_len,
        )?;

        let key_bytes = &key_value_bytes[..size_of::<K>()];
        let value_bytes = &key_value_bytes[size_of::<K>()..];

        let expected_crc = compute_crc32(header.gsn, header.value_len, key_bytes, value_bytes);
        if expected_crc != header.crc32 {
            if offset + total > file_len {
                tracing::warn!(shard_id, file_id, offset, "CRC mismatch at end of file");
                break;
            }
            return Err(DbError::CrcMismatch {
                expected: expected_crc,
                actual: header.crc32,
            });
        }

        let seq = header.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        if !tracker_update(tracker, dead_bytes, key_bytes, seq, file_id, total) {
            offset += total;
            continue;
        }

        let key: K = K::from_bytes(key_bytes);

        if header.is_tombstone() {
            map.remove(&key);
        } else {
            let value: [u8; V] = value_bytes
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset })?;
            let value_offset = offset + size_of::<EntryHeader>() as u64 + size_of::<K>() as u64;
            let disk = DiskLoc::new(file_id, value_offset as u32, header.value_len);

            map.upsert(key, crate::const_map::MapEntry { loc: disk, value });
        }

        offset += total;
    }
    direct::fadvise_dontneed(&file, 0, file_len);
    Ok(offset)
}

// ---------------------------------------------------------------------------
// VarMap hint-based recovery — pure metadata, no disk reads
// ---------------------------------------------------------------------------

#[cfg(feature = "var-collections")]
#[allow(clippy::too_many_arguments)]
fn recover_var_map_hint<K: Key + Send + Sync + Hash + Eq>(
    _shard_id: u8,
    file_id: u32,
    hint_data: &[u8],
    map: &mut ShardIndex<K, DiskLoc>,
    max_gsn: &mut u64,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<()> {
    for entry in hint::parse_hint_entries::<K>(hint_data) {
        let seq = entry.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        let total = entry_size(size_of::<K>(), entry.value_len) as u64;
        if !tracker_update(
            tracker,
            dead_bytes,
            entry.key.as_bytes(),
            seq,
            file_id,
            total,
        ) {
            continue;
        }

        if entry.is_tombstone() {
            map.remove(&entry.key);
        } else {
            let disk = DiskLoc::new(file_id, entry.value_offset as u32, entry.value_len);
            map.upsert(entry.key, disk);
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// VarMap full scan recovery
// ---------------------------------------------------------------------------

#[cfg(feature = "var-collections")]
#[allow(clippy::too_many_arguments)]
fn recover_var_map_full_scan<K: Key + Send + Sync + Hash + Eq>(
    shard_id: u8,
    file_id: u32,
    data_path: &Path,
    map: &mut ShardIndex<K, DiskLoc>,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<u64> {
    let file = direct::open_bulk_read(data_path)?;
    let file_len = file.metadata()?.len();
    let mut offset: u64 = 0;

    while offset + size_of::<EntryHeader>() as u64 <= file_len {
        let header_bytes = match read_fn(&file, offset, size_of::<EntryHeader>()) {
            Ok(b) => b,
            Err(e) => {
                if matches!(&e, DbError::Io(io) if io.kind() == std::io::ErrorKind::UnexpectedEof) {
                    break;
                }
                return Err(e);
            }
        };
        let header = match EntryHeader::read_from_bytes(&header_bytes) {
            Ok(h) => h,
            Err(_) => break,
        };

        if header.gsn == 0
            && header.value_len == 0
            && header.crc32 == 0
            && crate::entry::check_page_padding_at(
                &file,
                offset + size_of::<EntryHeader>() as u64,
                read_fn,
            )?
        {
            const PAGE_SIZE: u64 = 4096;
            let next_page =
                (offset + size_of::<EntryHeader>() as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
            if next_page >= file_len {
                break;
            }
            offset = next_page;
            continue;
        }

        let total = entry_size(size_of::<K>(), header.value_len);
        if offset + total > file_len {
            tracing::warn!(
                shard_id,
                file_id,
                offset,
                "truncating partial entry at end of file"
            );
            break;
        }

        let key_value_len = size_of::<K>() + header.value_len as usize;
        let key_value_bytes = read_fn(
            &file,
            offset + size_of::<EntryHeader>() as u64,
            key_value_len,
        )?;

        let key_bytes = &key_value_bytes[..size_of::<K>()];
        let value_bytes = &key_value_bytes[size_of::<K>()..];

        let expected_crc = compute_crc32(header.gsn, header.value_len, key_bytes, value_bytes);
        if expected_crc != header.crc32 {
            if offset + total > file_len {
                tracing::warn!(shard_id, file_id, offset, "CRC mismatch at end of file");
                break;
            }
            return Err(DbError::CrcMismatch {
                expected: expected_crc,
                actual: header.crc32,
            });
        }

        let seq = header.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        if !tracker_update(tracker, dead_bytes, key_bytes, seq, file_id, total) {
            offset += total;
            continue;
        }

        let key: K = K::from_bytes(key_bytes);

        if header.is_tombstone() {
            map.remove(&key);
        } else {
            let value_offset = offset + size_of::<EntryHeader>() as u64 + size_of::<K>() as u64;
            let disk = DiskLoc::new(file_id, value_offset as u32, header.value_len);
            map.upsert(key, disk);
        }

        offset += total;
    }
    direct::fadvise_dontneed(&file, 0, file_len);
    Ok(offset)
}

// ---------------------------------------------------------------------------
// TypedMap recovery
// ---------------------------------------------------------------------------

#[cfg(feature = "typed-tree")]
pub fn recover_typed_map<K: Key + Send + Sync + Hash + Eq, T: Send + Sync, C: Codec<T> + Sync>(
    shard_dirs: &[&Path],
    shard_ids: &[u8],
    indexes: &[Mutex<ShardIndex<K, crate::typed_map::TypedMapEntry<T>>>],
    codec: &C,
    hints: bool,
    #[cfg(feature = "encryption")] cipher: Option<Arc<PageCipher>>,
) -> DbResult<RecoveryOutcome> {
    let results: Vec<ShardRecoveryResult> = std::thread::scope(|s| {
        let handles: Vec<_> = shard_dirs
            .iter()
            .enumerate()
            .map(|(i, &dir)| {
                let shard_id = shard_ids[i];
                let index = &indexes[i];
                #[cfg(feature = "encryption")]
                let cipher = cipher.clone();
                s.spawn(move || -> ShardRecoveryResult {
                    let shard_start = std::time::Instant::now();
                    let mut local_max_gsn: u64 = 0;
                    let mut used_hints = false;
                    let mut active_tail: Option<ActiveTail> = None;
                    let mut map = sync::lock(index);
                    let file_ids = scan_data_files(dir)?;
                    let hint_ids = if hints {
                        hint::scan_hint_files(dir)?
                    } else {
                        Vec::new()
                    };
                    let mut tracker: RecoveryTracker = std::collections::HashMap::new();
                    let mut dead_bytes: DeadBytesMap = std::collections::HashMap::new();

                    for file_id in &file_ids {
                        let data_path = dir.join(format!("{file_id:06}.data"));
                        let is_active = Some(file_id) == file_ids.last();

                        #[cfg(feature = "encryption")]
                        let read_fn = make_reader(&cipher, dir, *file_id)?;
                        #[cfg(not(feature = "encryption"))]
                        let read_fn = make_reader(dir, *file_id)?;

                        if hint_ids.contains(file_id) {
                            let hint_path = dir.join(format!("{file_id:06}.hint"));
                            if let Some(hint_data) = hint::read_hint_file(&hint_path)? {
                                if hint_data.len() % hint::hint_entry_size(size_of::<K>()) == 0 {
                                    let data_file = direct::open_bulk_read(&data_path)?;
                                    let file_len = data_file.metadata()?.len();
                                    recover_typed_map_hint::<K, T, C>(
                                        shard_id,
                                        *file_id,
                                        &hint_data,
                                        &data_file,
                                        &mut map,
                                        codec,
                                        &mut local_max_gsn,
                                        &read_fn,
                                        &mut tracker,
                                        &mut dead_bytes,
                                    )?;
                                    direct::fadvise_dontneed(&data_file, 0, file_len);
                                    used_hints = true;
                                    continue;
                                }
                                tracing::warn!(
                                    shard_id,
                                    file_id,
                                    "hint file has unexpected size, falling back to full scan"
                                );
                            }
                        }

                        let last_valid = recover_typed_map_full_scan::<K, T, C>(
                            shard_id,
                            *file_id,
                            &data_path,
                            &mut map,
                            codec,
                            &mut local_max_gsn,
                            &read_fn,
                            &mut tracker,
                            &mut dead_bytes,
                        )?;
                        if is_active {
                            active_tail = Some(ActiveTail {
                                shard_idx: i,
                                file_id: *file_id,
                                last_valid_offset: last_valid,
                            });
                        }
                    }

                    if let Some(&last_id) = file_ids.last() {
                        let _ = fs::remove_file(dir.join(format!("{last_id:06}.hint")));
                    }

                    let elapsed = shard_start.elapsed().as_secs_f64();
                    metrics::histogram!("armdb.recovery.duration_seconds", "path" => if used_hints { "hint" } else { "full_scan" })
                        .record(elapsed);
                    Ok((local_max_gsn, active_tail, i, dead_bytes))
                })
            })
            .collect();

        handles
            .into_iter()
            .map(|h| match h.join() {
                Ok(r) => r,
                Err(_) => Err(DbError::RecoveryPanic),
            })
            .collect()
    });

    let mut max_gsn: u64 = 0;
    let mut active_tails: Vec<ActiveTail> = Vec::new();
    let mut shard_dead_bytes: Vec<(usize, DeadBytesMap)> = Vec::new();
    for result in results {
        let (shard_gsn, tail, shard_idx, dead) = result?;
        if shard_gsn > max_gsn {
            max_gsn = shard_gsn;
        }
        if let Some(t) = tail {
            active_tails.push(t);
        }
        if !dead.is_empty() {
            shard_dead_bytes.push((shard_idx, dead));
        }
    }

    Ok(RecoveryOutcome {
        max_gsn,
        active_tails,
        shard_dead_bytes,
    })
}

#[cfg(feature = "typed-tree")]
#[allow(clippy::too_many_arguments)]
fn recover_typed_map_hint<K: Key + Send + Sync + Hash + Eq, T: Send + Sync, C: Codec<T>>(
    _shard_id: u8,
    file_id: u32,
    hint_data: &[u8],
    data_file: &std::fs::File,
    map: &mut ShardIndex<K, crate::typed_map::TypedMapEntry<T>>,
    codec: &C,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<()> {
    for entry in hint::parse_hint_entries::<K>(hint_data) {
        let seq = entry.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        let total = entry_size(size_of::<K>(), entry.value_len) as u64;
        if !tracker_update(
            tracker,
            dead_bytes,
            entry.key.as_bytes(),
            seq,
            file_id,
            total,
        ) {
            continue;
        }

        if entry.is_tombstone() {
            if let Some(old) = map.remove(&entry.key) {
                unsafe {
                    drop(Box::from_raw(old.ptr));
                }
            }
        } else {
            let value_bytes = read_fn(data_file, entry.value_offset, entry.value_len as usize)?;
            let expected_crc = compute_crc32(
                entry.gsn,
                entry.value_len,
                entry.key.as_bytes(),
                &value_bytes,
            );
            if expected_crc != entry.crc32 {
                return Err(DbError::CrcMismatch {
                    expected: expected_crc,
                    actual: entry.crc32,
                });
            }
            let value = codec
                .decode_from(&value_bytes)
                .map_err(|_| DbError::CorruptedEntry {
                    offset: entry.value_offset,
                })?;

            let disk = DiskLoc::new(file_id, entry.value_offset as u32, entry.value_len);

            let ptr = Box::into_raw(Box::new(TypedData { disk, value }));
            if let Some(old) = map.upsert(entry.key, crate::typed_map::TypedMapEntry { ptr }) {
                unsafe {
                    drop(Box::from_raw(old.ptr));
                }
            }
        }
    }
    Ok(())
}

#[cfg(feature = "typed-tree")]
#[allow(clippy::too_many_arguments)]
fn recover_typed_map_full_scan<K: Key + Send + Sync + Hash + Eq, T: Send + Sync, C: Codec<T>>(
    shard_id: u8,
    file_id: u32,
    data_path: &Path,
    map: &mut ShardIndex<K, crate::typed_map::TypedMapEntry<T>>,
    codec: &C,
    max_gsn: &mut u64,
    read_fn: &ReadFn,
    tracker: &mut RecoveryTracker,
    dead_bytes: &mut DeadBytesMap,
) -> DbResult<u64> {
    let file = direct::open_bulk_read(data_path)?;
    let file_len = file.metadata()?.len();
    let mut offset: u64 = 0;

    while offset + size_of::<EntryHeader>() as u64 <= file_len {
        let header_bytes = match read_fn(&file, offset, size_of::<EntryHeader>()) {
            Ok(b) => b,
            Err(e) => {
                if matches!(&e, DbError::Io(io) if io.kind() == std::io::ErrorKind::UnexpectedEof) {
                    break;
                }
                return Err(e);
            }
        };
        let header = match EntryHeader::read_from_bytes(&header_bytes) {
            Ok(h) => h,
            Err(_) => break,
        };

        if header.gsn == 0
            && header.value_len == 0
            && header.crc32 == 0
            && crate::entry::check_page_padding_at(
                &file,
                offset + size_of::<EntryHeader>() as u64,
                read_fn,
            )?
        {
            const PAGE_SIZE: u64 = 4096;
            let next_page =
                (offset + size_of::<EntryHeader>() as u64).div_ceil(PAGE_SIZE) * PAGE_SIZE;
            if next_page >= file_len {
                break;
            }
            offset = next_page;
            continue;
        }

        let total = entry_size(size_of::<K>(), header.value_len);
        if offset + total > file_len {
            tracing::warn!(
                shard_id,
                file_id,
                offset,
                "truncating partial entry at end of file"
            );
            break;
        }

        let key_value_len = size_of::<K>() + header.value_len as usize;
        let key_value_bytes = read_fn(
            &file,
            offset + size_of::<EntryHeader>() as u64,
            key_value_len,
        )?;

        let key_bytes = &key_value_bytes[..size_of::<K>()];
        let value_bytes = &key_value_bytes[size_of::<K>()..];

        let expected_crc = compute_crc32(header.gsn, header.value_len, key_bytes, value_bytes);
        if expected_crc != header.crc32 {
            if offset + total > file_len {
                tracing::warn!(shard_id, file_id, offset, "CRC mismatch at end of file");
                break;
            }
            return Err(DbError::CrcMismatch {
                expected: expected_crc,
                actual: header.crc32,
            });
        }

        let seq = header.sequence();
        if seq > *max_gsn {
            *max_gsn = seq;
        }

        if !tracker_update(tracker, dead_bytes, key_bytes, seq, file_id, total) {
            offset += total;
            continue;
        }

        let key: K = K::from_bytes(key_bytes);

        if header.is_tombstone() {
            if let Some(old) = map.remove(&key) {
                unsafe {
                    drop(Box::from_raw(old.ptr));
                }
            }
        } else {
            let value = codec
                .decode_from(value_bytes)
                .map_err(|_| DbError::CorruptedEntry { offset })?;

            let value_offset = offset + size_of::<EntryHeader>() as u64 + size_of::<K>() as u64;
            let disk = DiskLoc::new(file_id, value_offset as u32, header.value_len);

            let ptr = Box::into_raw(Box::new(TypedData { disk, value }));
            if let Some(old) = map.upsert(key, crate::typed_map::TypedMapEntry { ptr }) {
                unsafe {
                    drop(Box::from_raw(old.ptr));
                }
            }
        }

        offset += total;
    }
    direct::fadvise_dontneed(&file, 0, file_len);
    Ok(offset)
}

// ---------------------------------------------------------------------------
// Utilities
// ---------------------------------------------------------------------------

/// Scan a directory for `.data` files and return sorted file IDs.
fn scan_data_files(dir: &Path) -> DbResult<Vec<u32>> {
    let mut ids = Vec::new();
    if !dir.exists() {
        return Ok(ids);
    }
    for entry in fs::read_dir(dir)? {
        let entry = entry?;
        let name = entry.file_name();
        let name = name.to_string_lossy();
        if name.ends_with(".data")
            && let Ok(id) = name.trim_end_matches(".data").parse::<u32>()
        {
            ids.push(id);
        }
    }
    ids.sort();
    Ok(ids)
}

#[cfg(test)]
mod tests {
    use crate::entry::check_page_padding_at;

    #[test]
    fn page_padding_detected_for_zero_tail() {
        let dir = std::env::temp_dir().join(format!("armdb_pad_{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("p.data");

        // One page: 100 bytes of 0xAB, rest zero (padding).
        let mut page = vec![0u8; 4096];
        for b in page[..100].iter_mut() {
            *b = 0xAB;
        }
        let f = crate::io::direct::open_bulk_write(&path).unwrap();
        crate::io::direct::pwrite_at(&f, &page, 0).unwrap();
        crate::io::direct::fsync(&f).unwrap();
        drop(f);

        let r = crate::io::direct::open_bulk_read(&path).unwrap();
        let read = |file: &std::fs::File, off: u64, len: usize| {
            crate::io::direct::pread_value(file, off, len)
        };
        // From offset 100 to page end is all zero → padding.
        assert!(check_page_padding_at(&r, 100, &read).unwrap());
        // From offset 50 it includes 0xAB → not padding.
        assert!(!check_page_padding_at(&r, 50, &read).unwrap());

        std::fs::remove_dir_all(&dir).ok();
    }
}
