use std::{collections::HashSet, ops::Deref};

use anyhow::{Result, anyhow};
use log::info;
use serde::{Deserialize, Serialize};
use tici_grpc::FragMeta;

use crate::common::split;

#[derive(PartialEq, Eq, Clone, serde::Serialize, serde::Deserialize)]
pub enum RangeBound {
    LeftMost,
    Key(Vec<u8>),
    RightMost,
}

impl std::fmt::Debug for RangeBound {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self {
            Self::LeftMost => f.write_str("-inf"),
            Self::RightMost => f.write_str("inf"),
            Self::Key(v) => f.write_str(&hex::encode(v)),
        }
    }
}

impl RangeBound {
    #[allow(clippy::comparison_chain)]
    pub fn effective_eq(&self, other: &RangeBound) -> bool {
        match (self, other) {
            (RangeBound::Key(x), RangeBound::Key(y)) => {
                for (xi, yi) in x.iter().zip(y.iter()) {
                    if xi != yi {
                        return false;
                    }
                }
                if x.len() > y.len() {
                    x[y.len()..].iter().all(|&b| b == 0)
                } else if x.len() < y.len() {
                    y[x.len()..].iter().all(|&b| b == 0)
                } else {
                    true
                }
            }
            _ => self == other,
        }
    }

    pub fn make_start(start: Vec<u8>) -> Self {
        if start.is_empty() {
            RangeBound::LeftMost
        } else {
            RangeBound::Key(start)
        }
    }

    pub fn make_end(end: Vec<u8>) -> Self {
        if end.is_empty() {
            RangeBound::RightMost
        } else {
            RangeBound::Key(end)
        }
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn encoded_as_start(&self) -> Vec<u8> {
        match self {
            RangeBound::LeftMost => vec![],
            RangeBound::Key(k) => k.clone(),
            RangeBound::RightMost => vec![],
        }
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn encoded_as_end(&self) -> Vec<u8> {
        match self {
            RangeBound::LeftMost => vec![],
            RangeBound::Key(k) => k.clone(),
            RangeBound::RightMost => vec![],
        }
    }
}

impl From<RangeBound> for Vec<u8> {
    fn from(value: RangeBound) -> Vec<u8> {
        match value {
            RangeBound::LeftMost => vec![],
            RangeBound::Key(k) => k,
            RangeBound::RightMost => vec![],
        }
    }
}

impl Ord for RangeBound {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        match &self {
            Self::LeftMost => match other {
                Self::LeftMost => std::cmp::Ordering::Equal,
                Self::Key(_) => std::cmp::Ordering::Less,
                Self::RightMost => std::cmp::Ordering::Less,
            },
            Self::Key(k1) => match other {
                Self::LeftMost => std::cmp::Ordering::Greater,
                Self::Key(k2) => k1.cmp(k2),
                Self::RightMost => std::cmp::Ordering::Less,
            },
            Self::RightMost => match other {
                Self::LeftMost => std::cmp::Ordering::Greater,
                Self::Key(_) => std::cmp::Ordering::Greater,
                Self::RightMost => std::cmp::Ordering::Equal,
            },
        }
    }
}

impl PartialOrd for RangeBound {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

// TODO(P2): Merge it with `components/tici_shard/src/utils/handle.rs`,
// consider:
//       - Should we keep the two inf bounds in RangeBound?
//       - We'd better find a way to represent the Included/Excluded bounds
//         semantic. Otherwise it is easy to make mistakes when calling
//         following methods:
//          + `range_overlaps_range`
//          + `range_overlaps_minmax`
//          + `minmax_overlaps_minmax`
//          + `range_contains_minmax`
//       - Provide a HandleRange over &[u8]
#[derive(Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct HandleRange {
    start: RangeBound,
    end: RangeBound,
}

impl std::fmt::Debug for HandleRange {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_fmt(format_args!("{:?}..{:?}", &self.start, &self.end))
    }
}

impl HandleRange {
    pub fn new(start: Vec<u8>, end: Vec<u8>) -> Self {
        Self {
            start: RangeBound::make_start(start),
            end: RangeBound::make_end(end),
        }
    }

    pub fn new_from_sorted(start: RangeBound, end: RangeBound) -> Self {
        Self { start, end }
    }

    pub fn sortable_start_ref(&self) -> &RangeBound {
        &self.start
    }

    pub fn sortable_end_ref(&self) -> &RangeBound {
        &self.end
    }

    pub fn sortable_start(&self) -> RangeBound {
        self.start.clone()
    }

    pub fn sortable_end(&self) -> RangeBound {
        self.end.clone()
    }

    pub fn encoded_start(&self) -> Vec<u8> {
        self.start.clone().into()
    }

    pub fn encoded_end(&self) -> Vec<u8> {
        self.end.clone().into()
    }

    pub fn contains_key_not_included(&self, key: &RangeBound) -> bool {
        key > &self.start && key < &self.end
    }

    // if `key` is in [start, end)
    pub fn contains_key(&self, key: &RangeBound) -> bool {
        key >= &self.start && key < &self.end
    }

    // Check if `other` is subrange of `self`
    // Note that the fragment has range [min_handle, max_handle],
    // While the shard has range [start, end).
    pub fn range_contains_minmax(&self, other: &HandleRange) -> bool {
        other.start >= self.start && other.end < self.end
    }

    // if `self` is subrange of `other`
    pub fn is_subrange_of(&self, other: &HandleRange) -> bool {
        self.start >= other.start && self.end <= other.end
    }

    /// [a, b) and [c, d) overlap if: a < d && c < b
    pub fn range_overlaps_range(&self, key_start: &[u8], key_end: &[u8]) -> bool {
        let key_start_bound = RangeBound::make_start(key_start.to_vec());
        let key_end_bound = RangeBound::make_end(key_end.to_vec());
        self.start < key_end_bound && key_start_bound < self.end
    }

    /// [a, b) and [c, d] overlap if: a <= d && c < b
    pub fn range_overlaps_minmax(&self, key_start: &[u8], key_end: &[u8]) -> bool {
        let key_start_bound = RangeBound::make_start(key_start.to_vec());
        let key_end_bound = RangeBound::make_end(key_end.to_vec());
        self.start <= key_end_bound && key_start_bound < self.end
    }

    /// [a, b] and [c, d] overlap if: a <= d && c <= b
    pub fn minmax_overlaps_minmax(&self, key_start: &[u8], key_end: &[u8]) -> bool {
        let key_start_bound = RangeBound::make_start(key_start.to_vec());
        let key_end_bound = RangeBound::make_end(key_end.to_vec());
        self.start <= key_end_bound && key_start_bound <= self.end
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn debug_derive_frag_range(&self) -> Self {
        use tici_common::codec::previous_key_helper;

        Self::new(
            self.encoded_start(),
            previous_key_helper(self.encoded_end().as_slice(), 1),
        )
    }

    pub fn is_valid(&self) -> bool {
        self.start <= self.end
    }

    pub fn intersect(&self, other: &HandleRange) -> HandleRange {
        HandleRange::new_from_sorted(
            if self.start >= other.start {
                self.start.clone()
            } else {
                other.start.clone()
            },
            if self.end >= other.end {
                other.end.clone()
            } else {
                self.end.clone()
            },
        )
    }
}

/// If the whole fragment can't be splitted by a split key,
/// then we can only compute reduce its size by half,
/// before the next compaction.
/// Otherwise, split can be repeatedly performed.
#[derive(Clone, Serialize, Deserialize)]
pub struct SplittableFrag {
    pub f: FragMeta,
    // (shard_id, seq)
    pub c: (u64, u64),
    /// Currently, after a split, all fragments' p will be multiplied by
    /// split ratio, which means `p` is estimated for the most of the
    /// fragments. Also, all `p` of the same fragments will not always add
    /// up to 1.0. Consider s1 splitted from [f1:1.0] to s2 [f1:0.5] and s3
    /// [f1:0.5]. If s3 is then compacted into `s3 [f5:1.0]`, then `f1`` is
    /// partially invalid. Furthermore, if `p` is zero, then all subranges
    /// of the fragment is compacted, we can safely gc this fragment.
    pub p: f64,
}

impl heapsize::HeapSizeOf for SplittableFrag {
    fn heap_size_of_children(&self) -> usize {
        self.f.heap_size_of_children()
    }
}

impl PartialEq for SplittableFrag {
    fn eq(&self, other: &Self) -> bool {
        // A frag doesn't belong to a shard, it is created by a shard, and referenced by
        // multiple shards.
        self.f == other.f && self.p == other.p
    }
}

impl std::fmt::Debug for SplittableFrag {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if let Some(p) = self.f.property.as_ref() {
            f.debug_struct("SF")
                .field("p", &self.p)
                .field("c", &self.c)
                .field("path", &self.f.frag_path)
                .field("stats", &(p.size, p.count, self.f.seg_count()))
                .field(
                    "range",
                    &format!(
                        "{}..{}",
                        &hex::encode(&p.min_handle),
                        &hex::encode(&p.max_handle)
                    ),
                )
                .field("mid", &p.format_split_handles())
                .finish()
        } else {
            f.debug_struct("SF")
                .field("p", &self.p)
                .field("c", &self.c)
                .field("path", &self.f.frag_path)
                .field("stats", &(0usize, 0usize, self.f.seg_count()))
                .finish()
        }
    }
}

impl SplittableFrag {
    pub fn new(f: FragMeta, shard_id: u64, seq: u64) -> Self {
        Self {
            f,
            p: 1.0,
            c: (shard_id, seq),
        }
    }

    pub fn weighted_bytes(&self) -> u64 {
        let size = self.f.property.as_ref().map_or(0, |p| p.size);
        split::weighted_u64(size, self.p)
    }
}

impl Deref for SplittableFrag {
    type Target = FragMeta;
    fn deref(&self) -> &Self::Target {
        &self.f
    }
}

// Fragment location in the new two-vector structure
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FragmentLocation {
    Base(usize),
    Delta(usize),
}

// Remove Serialize, Deserialize from derive - will implement manually for
// backward compatibility
#[derive(Debug, Clone, PartialEq)]
pub struct ShardManifest {
    // Base layer (L2) - immutable, no deleted set
    pub(super) base_fragments: Vec<SplittableFrag>,

    // Delta layer - active compaction area
    pub(super) delta_fragments: Vec<SplittableFrag>,
}

// Custom Serialize implementation for compatibility with legacy first_l1 and
// new b field.
impl Serialize for ShardManifest {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        use serde::ser::SerializeStruct;

        // Concatenate base + delta for persistence
        let mut all_fragments =
            Vec::with_capacity(self.base_fragments.len() + self.delta_fragments.len());
        all_fragments.extend(self.base_fragments.iter().cloned());
        all_fragments.extend(self.delta_fragments.iter().cloned());

        // Calculate absolute indices for persistence (old format)
        let persisted_b = self.base_fragments.len();

        let mut state = serializer.serialize_struct("ShardManifest", 2)?;
        state.serialize_field("fragments", &all_fragments)?;
        state.serialize_field("b", &persisted_b)?;
        state.end()
    }
}

// Custom Deserialize implementation for compatibility with legacy first_l1 and
// new b field.
impl<'de> Deserialize<'de> for ShardManifest {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        #[derive(Deserialize)]
        struct PersistedManifest {
            fragments: Vec<SplittableFrag>,
            #[serde(default)]
            b: Option<usize>,
            #[serde(default)]
            first_l1: Option<usize>,
        }

        let persisted = PersistedManifest::deserialize(deserializer)?;

        let first_l1 = match (persisted.b, persisted.first_l1) {
            (Some(b), Some(first_l1)) => {
                log::error!(
                    "ShardManifest has both 'b' and legacy 'first_l1'; ignoring first_l1 (b={}, first_l1={})",
                    b,
                    first_l1
                );
                b
            }
            (Some(b), None) => b,
            (None, Some(first_l1)) => first_l1,
            (None, None) => {
                return Err(serde::de::Error::custom(
                    "ShardManifest missing 'b'/'first_l1' boundary",
                ));
            }
        };

        // Split at first_l1 boundary
        let mut fragments = persisted.fragments;
        let delta_fragments = fragments.split_off(first_l1);
        let base_fragments = fragments;

        Ok(ShardManifest {
            base_fragments,
            delta_fragments,
        })
    }
}

impl From<ShardManifest> for Vec<SplittableFrag> {
    fn from(manifest: ShardManifest) -> Self {
        // Concatenate base + delta
        let mut all =
            Vec::with_capacity(manifest.base_fragments.len() + manifest.delta_fragments.len());
        all.extend(manifest.base_fragments);
        all.extend(manifest.delta_fragments);
        all
    }
}

impl heapsize::HeapSizeOf for ShardManifest {
    fn heap_size_of_children(&self) -> usize {
        self.base_fragments.heap_size_of_children() + self.delta_fragments.heap_size_of_children()
    }
}

impl Default for ShardManifest {
    fn default() -> Self {
        Self::new_for_first()
    }
}

impl ShardManifest {
    // First of the table
    pub fn new_for_first() -> Self {
        Self {
            base_fragments: vec![],
            delta_fragments: vec![],
        }
    }

    pub fn get_total_datafile_count(&self) -> usize {
        self.base_fragments.iter().fold(0, |a, b| a + b.seg_count())
            + self
                .delta_fragments
                .iter()
                .fold(0, |a, b| a + b.seg_count())
    }

    pub fn get_total_estimated_bytes(&self) -> u64 {
        let base: u64 = self
            .base_fragments
            .iter()
            .map(|x| (x.bytes() as f64 * x.p).round() as u64)
            .sum();
        let delta: u64 = self
            .delta_fragments
            .iter()
            .map(|x| (x.bytes() as f64 * x.p).round() as u64)
            .sum();
        base + delta
    }

    pub fn get_total_estimated_rows(&self) -> u64 {
        let base: u64 = self
            .base_fragments
            .iter()
            .map(|x| (x.rows() as f64 * x.p).round() as u64)
            .sum();
        let delta: u64 = self
            .delta_fragments
            .iter()
            .map(|x| (x.rows() as f64 * x.p).round() as u64)
            .sum();
        base + delta
    }

    fn generate_split(
        mut fragments: Vec<SplittableFrag>,
        new_range: &HandleRange,
        splits: u64,
    ) -> Vec<SplittableFrag> {
        fragments
            .drain(..)
            .filter(|x| {
                let prop = x.property.as_ref().unwrap();
                new_range.range_overlaps_minmax(&prop.min_handle, &prop.max_handle)
            })
            .map(|x| SplittableFrag {
                f: x.f.clone(),
                p: {
                    let prop = x.property.as_ref().unwrap();
                    let handle =
                        HandleRange::new(prop.min_handle.to_vec(), prop.max_handle.to_vec());
                    if new_range.range_contains_minmax(&handle) {
                        x.p
                    } else {
                        x.p / (splits as f64)
                    }
                },
                c: x.c,
            })
            .collect()
    }

    /// Split the shard:
    /// - If a fragment entirely belongs to a split, move it into that split.
    /// - Otherwise, both new shards own the same fragment, with evenly splitted
    ///   propotion `p`.
    pub fn trim_fragments(&mut self, new_range: &HandleRange, splits: u64) {
        let base =
            Self::generate_split(std::mem::take(&mut self.base_fragments), new_range, splits);
        let delta =
            Self::generate_split(std::mem::take(&mut self.delta_fragments), new_range, splits);

        self.base_fragments = base;
        self.delta_fragments = delta;

        if self.base_fragments.is_empty() && self.delta_fragments.is_empty() {
            log::debug!("trim to empty, new_range={:?}", new_range);
        }
    }

    pub fn cloned_fragments(&self) -> Vec<FragMeta> {
        let mut result = Vec::with_capacity(self.base_fragments.len() + self.delta_fragments.len());
        result.extend(self.base_fragments.iter().map(|x| x.f.clone()));
        result.extend(self.delta_fragments.iter().map(|x| x.f.clone()));
        result
    }

    pub fn cloned_splittable_fragments(&self) -> Vec<SplittableFrag> {
        let mut result = Vec::with_capacity(self.base_fragments.len() + self.delta_fragments.len());
        result.extend(self.base_fragments.iter().cloned());
        result.extend(self.delta_fragments.iter().cloned());
        result
    }

    // Helper to get all fragments concatenated (for compatibility)
    pub fn get_fragments(&self) -> Vec<SplittableFrag> {
        let mut all = Vec::with_capacity(self.base_fragments.len() + self.delta_fragments.len());
        all.extend(self.base_fragments.iter().cloned());
        all.extend(self.delta_fragments.iter().cloned());
        all
    }

    pub fn append_delta_fragments(&mut self, frags: Vec<FragMeta>, shard_id: u64, seq: u64) {
        self.delta_fragments.extend(
            frags
                .into_iter()
                .map(|f| SplittableFrag::new(f, shard_id, seq)),
        );
    }

    fn get_duplicated(&self, frag: &[FragMeta]) -> HashSet<String> {
        let s: HashSet<&str> = frag.iter().map(|x| x.frag_path.as_str()).collect();
        let mut res = HashSet::new();
        for f in self
            .base_fragments
            .iter()
            .chain(self.delta_fragments.iter())
        {
            if s.contains(f.frag_path.as_str()) {
                res.insert(f.frag_path.to_string());
            }
        }
        res
    }

    pub fn make_deduplicated(
        &self,
        frags: Vec<FragMeta>,
        tag: &str,
        shard_id: u64,
        seq: u64,
    ) -> Vec<FragMeta> {
        let duplicated = self.get_duplicated(frags.as_slice());
        if !duplicated.is_empty() {
            info!(
                "{}: duplicated frags, shard_id={}, new_seq={}, frags={:?}",
                tag, shard_id, seq, frags,
            );
        }
        frags
            .into_iter()
            .filter(|x| !duplicated.contains(&x.frag_path))
            .collect()
    }

    pub fn append_base_fragments(&mut self, frags: Vec<FragMeta>, shard_id: u64, seq: u64) {
        let new_frags: Vec<_> = frags
            .into_iter()
            .map(|f| SplittableFrag::new(f, shard_id, seq))
            .collect();
        // Insert at beginning of base_fragments
        self.base_fragments.splice(0..0, new_frags);
    }

    /// Switch imported base fragments into the delta layer while preserving the
    /// externally visible fragment order: `base + delta`.
    pub fn switch_base_to_delta(&mut self) -> bool {
        if self.base_fragments.is_empty() {
            return false;
        }

        let mut delta_fragments = std::mem::take(&mut self.delta_fragments);
        self.base_fragments.append(&mut delta_fragments);
        self.delta_fragments = std::mem::take(&mut self.base_fragments);
        true
    }

    #[allow(clippy::len_without_is_empty)]
    pub fn len(&self) -> usize {
        self.base_fragments.len() + self.delta_fragments.len()
    }

    pub fn total_weighted_delta_size(&self) -> u64 {
        self.delta_fragments
            .iter()
            .map(SplittableFrag::weighted_bytes)
            .sum()
    }

    pub fn total_weighted_base_size(&self) -> u64 {
        self.base_fragments
            .iter()
            .map(SplittableFrag::weighted_bytes)
            .sum()
    }

    pub fn find_pos_by_frag_path(&self, frag_path: &String) -> Option<FragmentLocation> {
        if let Some(pos) = self
            .base_fragments
            .iter()
            .position(|x| &x.frag_path == frag_path)
        {
            return Some(FragmentLocation::Base(pos));
        }
        if let Some(pos) = self
            .delta_fragments
            .iter()
            .position(|x| &x.frag_path == frag_path)
        {
            return Some(FragmentLocation::Delta(pos));
        }
        None
    }

    // TODO(P1) Support n-fold split.
    pub fn compute_split_points(
        &self,
        range: &HandleRange,
        split_point_scan_size_ratio: f64,
    ) -> Result<Vec<RangeBound>> {
        let sanitize = |split_points: Vec<RangeBound>| -> Result<Vec<RangeBound>> {
            if split_points.iter().any(|point| {
                point.effective_eq(range.sortable_start_ref())
                    || point.effective_eq(range.sortable_end_ref())
            }) {
                return Ok(Vec::new());
            }
            Ok(split_points)
        };

        let fallback = || -> Result<Vec<RangeBound>> {
            let start = range.sortable_start_ref();
            let end = range.sortable_end_ref();
            if matches!(start, RangeBound::LeftMost | RangeBound::RightMost)
                || matches!(end, RangeBound::LeftMost | RangeBound::RightMost)
            {
                return Err(anyhow!(
                    "handle range contains inf bound: start={:?}, end={:?}",
                    start,
                    end
                ));
            }
            let split_points = split::midpoint_key(&range.encoded_start(), &range.encoded_end())
                .map(|handle| vec![RangeBound::Key(handle)])
                .unwrap_or_default();
            sanitize(split_points)
        };

        let total_size = self.total_weighted_delta_size() + self.total_weighted_base_size();
        if total_size == 0 {
            return fallback();
        }
        let mut scanned_size = 0u64;
        let mut has_any_handle = false;
        let mut candidates: Vec<(Vec<u8>, u64)> = Vec::new();

        for frag in self
            .delta_fragments
            .iter()
            .chain(self.base_fragments.iter())
        {
            if (scanned_size as f64) / (total_size as f64) >= split_point_scan_size_ratio {
                break;
            }
            let Some(prop) = frag.property.as_ref() else {
                continue;
            };
            scanned_size = scanned_size.saturating_add(frag.weighted_bytes());
            let Some(handle) = split::fragment_split_handle(prop, range) else {
                continue;
            };
            has_any_handle = true;
            let weight = split::weighted_u64(prop.count, frag.p);
            if weight > 0 {
                candidates.push((handle, weight));
            }
        }

        if !has_any_handle || candidates.is_empty() {
            return fallback();
        }

        match split::weighted_average_keys(&candidates) {
            Some(handle) => sanitize(vec![RangeBound::Key(handle)]),
            None => fallback(),
        }
    }

    pub fn merge(&mut self, other: &ShardManifest) -> Option<(SplittableFrag, SplittableFrag)> {
        if let Some((left, right)) = self.find_first_overlapped_fragment(other) {
            return Some((left, right));
        }
        // Merge base fragments
        self.base_fragments
            .extend(other.base_fragments.iter().cloned());

        let our_delta = std::mem::take(&mut self.delta_fragments);
        self.delta_fragments
            .extend(other.delta_fragments.iter().cloned());
        self.delta_fragments.extend(our_delta);
        None
    }

    fn find_first_overlapped_fragment(
        &self,
        other: &ShardManifest,
    ) -> Option<(SplittableFrag, SplittableFrag)> {
        for s1 in self.get_fragments().iter() {
            let range1 = HandleRange::new(
                s1.property.as_ref().unwrap().min_handle.clone(),
                s1.property.as_ref().unwrap().max_handle.clone(),
            );
            for s2 in other.get_fragments().iter() {
                if range1.minmax_overlaps_minmax(
                    s2.property.as_ref().unwrap().min_handle.as_slice(),
                    s2.property.as_ref().unwrap().max_handle.as_slice(),
                ) {
                    return Some((s1.clone(), s2.clone()));
                }
            }
        }
        None
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn remove_frag(&mut self, frag_path: &str) {
        if let Some(pos) = self
            .base_fragments
            .iter()
            .position(|x| x.frag_path == frag_path)
        {
            self.base_fragments.remove(pos);
            return;
        }

        if let Some(pos) = self
            .delta_fragments
            .iter()
            .position(|x| x.frag_path == frag_path)
        {
            self.delta_fragments.remove(pos);
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{
        collections::BTreeMap,
        sync::atomic::{AtomicU64, Ordering},
    };

    use tici_grpc::FragProperty;
    use tici_shard::test_utils::test_mock_frag_storage_refs_total;

    use super::*;
    use crate::common::test_utils::*;

    static FRAG_ID_COUNTER: AtomicU64 = AtomicU64::new(0);

    fn make_frag_with_handles(
        min_handle: &[u8],
        max_handle: &[u8],
        split_handles: &[&[u8]],
        size: u64,
        count: u64,
    ) -> FragMeta {
        let frag_id = FRAG_ID_COUNTER.fetch_add(1, Ordering::SeqCst);
        FragMeta::new(
            0,
            format!("test_frag_{frag_id}/f_1"),
            FragProperty {
                min_handle: min_handle.to_vec(),
                split_handles: split_handles.iter().map(|h| h.to_vec()).collect(),
                max_handle: max_handle.to_vec(),
                size,
                count,
                delete_count: 0,
            },
            test_mock_frag_storage_refs_total(size),
            1,
        )
        .unwrap()
    }

    #[test]
    fn test_range_cound() {
        // Validates RangeBound ordering and BTreeMap lower_bound behavior.
        assert!(RangeBound::make_end(vec![]) > RangeBound::make_start(vec![]));
        assert!(RangeBound::make_end(vec![]) != RangeBound::make_start(vec![]));
        assert!(RangeBound::make_end(vec![]) > RangeBound::make_end(vec![b'A']));
        assert!(RangeBound::make_start(vec![]) < RangeBound::make_start(vec![b'A']));
        assert!(RangeBound::make_start(vec![b'A']) == RangeBound::make_end(vec![b'A']));
        assert!(RangeBound::make_start(vec![b'A']) < RangeBound::make_start(vec![b'B']));
        assert!(RangeBound::make_end(vec![b'A']) < RangeBound::make_end(vec![b'B']));

        let mut bm = BTreeMap::new();
        bm.insert(RangeBound::make_start(vec![b'A']), 1);
        assert_eq!(
            bm.lower_bound(std::collections::Bound::Included(&RangeBound::make_start(
                vec![]
            )))
            .next()
            .unwrap()
            .1,
            &1
        );
        assert!(
            bm.lower_bound(std::collections::Bound::Included(&RangeBound::make_end(
                vec![]
            )))
            .next()
            .is_none()
        );
    }

    #[test]
    fn test_handle_range_overlaps_with() {
        // Exercises HandleRange overlap helpers against overlapping and disjoint
        // ranges. Test range [b"aaaa", b"zzzz")
        let handle_range = HandleRange::new(b"aaaa".to_vec(), b"zzzz".to_vec());

        // Test overlaps
        assert!(handle_range.range_overlaps_range(b"aaaa", b"zzzz")); // Exact match
        assert!(handle_range.range_overlaps_range(b"bbbb", b"cccc")); // Within range
        assert!(handle_range.range_overlaps_range(b"bbbb", b"zzzz")); // Within range
        assert!(handle_range.range_overlaps_range(b"0000", b"bbbb")); // Partial start
        assert!(handle_range.range_overlaps_range(b"yyyy", b"zzzzz")); // Partial end
        assert!(handle_range.range_overlaps_range(b"0000", b"zzzzz")); // Contains range
        // Test disjoint
        assert!(!handle_range.range_overlaps_range(b"0000", b"aaaa")); // Before range
        assert!(!handle_range.range_overlaps_range(b"zzzz", b"zzzzz")); // After range

        // Test overlaps
        assert!(handle_range.range_overlaps_minmax(b"aaaa", b"zzzz")); // Exact match
        assert!(handle_range.range_overlaps_minmax(b"bbbb", b"cccc")); // Within range
        assert!(handle_range.range_overlaps_minmax(b"bbbb", b"zzzz")); // Within range
        assert!(handle_range.range_overlaps_minmax(b"0000", b"bbbb")); // Partial start
        assert!(handle_range.range_overlaps_minmax(b"yyyy", b"zzzzz")); // Partial end
        assert!(handle_range.range_overlaps_minmax(b"0000", b"zzzzz")); // Contains range
        assert!(handle_range.range_overlaps_minmax(b"0000", b"aaaa"));
        // Test disjoint
        assert!(!handle_range.range_overlaps_minmax(b"zzzz", b"zzzzz"));

        // Test overlaps
        assert!(handle_range.minmax_overlaps_minmax(b"aaaa", b"zzzz")); // Exact match
        assert!(handle_range.minmax_overlaps_minmax(b"bbbb", b"cccc")); // Within range
        assert!(handle_range.minmax_overlaps_minmax(b"bbbb", b"zzzz")); // Within range
        assert!(handle_range.minmax_overlaps_minmax(b"0000", b"bbbb")); // Partial start
        assert!(handle_range.minmax_overlaps_minmax(b"yyyy", b"zzzzz")); // Partial end
        assert!(handle_range.minmax_overlaps_minmax(b"0000", b"zzzzz")); // Contains range
        assert!(handle_range.minmax_overlaps_minmax(b"0000", b"aaaa"));
        assert!(handle_range.minmax_overlaps_minmax(b"zzzz", b"zzzzz"));

        // Test no overlaps

        {
            // HandleRange for table_id=1144, [t1144, t1145)
            let h = HandleRange::new(
                hex::decode("748000000000000478").unwrap(),
                hex::decode("748000000000000479").unwrap(),
            );
            // A HandleRange inside table_id=1144
            let kr = tici_grpc::KeyRange {
                start_key: hex::decode(
                    "7480000000000004785f72038000000000000001038000000000000050",
                )
                .unwrap(),
                end_key: hex::decode("7480000000000004785f72038000000000000064038000000000000031")
                    .unwrap(),
            };
            assert!(h.range_overlaps_range(&kr.start_key, &kr.end_key),);
        }
    }

    #[test]
    fn test_find_first_overlapped() {
        // Ensures find_first_overlapped_fragment detects overlaps across base/delta
        // fragments.
        let mut m1 = ShardManifest::default();
        m1.append_delta_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"C")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_delta_fragments(vec![make_sf(b"C", b"D")], 0, 0);
        assert!(m1.find_first_overlapped_fragment(&m2).is_some());
        assert!(m2.find_first_overlapped_fragment(&m1).is_some());

        let mut m1 = ShardManifest::default();
        m1.append_delta_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"C")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_delta_fragments(vec![make_sf(b"C1", b"D")], 0, 0);
        assert!(m1.find_first_overlapped_fragment(&m2).is_none());
        assert!(m2.find_first_overlapped_fragment(&m1).is_none());

        let mut m1 = ShardManifest::default();
        m1.append_delta_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_delta_fragments(vec![make_sf(b"B", b"C")], 0, 0);
        assert!(m1.find_first_overlapped_fragment(&m2).is_some());
        assert!(m2.find_first_overlapped_fragment(&m1).is_some());

        let mut m1 = ShardManifest::default();
        m1.append_delta_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_delta_fragments(vec![make_sf(b"B", b"D")], 0, 0);
        assert!(m1.find_first_overlapped_fragment(&m2).is_some());
        assert!(m2.find_first_overlapped_fragment(&m1).is_some());

        let mut m1 = ShardManifest::default();
        m1.append_delta_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_delta_fragments(vec![make_sf(b"B", b"D")], 0, 0);
        assert!(m1.find_first_overlapped_fragment(&m2).is_some());
        assert!(m2.find_first_overlapped_fragment(&m1).is_some());

        let mut m1 = ShardManifest::default();
        m1.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        m1.append_delta_fragments(vec![make_sf(b"D", b"E")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_base_fragments(vec![make_sf(b"E1", b"F")], 0, 0);
        m2.append_delta_fragments(vec![make_sf(b"F", b"G")], 0, 0);
        assert!(m1.find_first_overlapped_fragment(&m2).is_none());
        assert!(m2.find_first_overlapped_fragment(&m1).is_none());

        let mut m1 = ShardManifest::default();
        m1.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        m1.append_delta_fragments(vec![make_sf(b"D", b"E")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_base_fragments(vec![make_sf(b"E", b"F")], 0, 0);
        m2.append_delta_fragments(vec![make_sf(b"F", b"G")], 0, 0);
        assert!(m1.find_first_overlapped_fragment(&m2).is_some());
        assert!(m2.find_first_overlapped_fragment(&m1).is_some());

        let mut m1 = ShardManifest::default();
        m1.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        m1.append_delta_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_base_fragments(vec![make_sf(b"B", b"D")], 0, 0);
        m2.append_delta_fragments(vec![make_sf(b"B", b"D")], 0, 0);
        assert!(m1.find_first_overlapped_fragment(&m2).is_some());
        assert!(m2.find_first_overlapped_fragment(&m1).is_some());
    }

    #[test]
    fn test_merge() {
        // Verifies merge behavior for overlapping vs non-overlapping manifests.
        // delta fragments without overlap
        let mut m1 = ShardManifest::default();
        m1.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        m1.append_delta_fragments(vec![make_sf(b"D", b"E")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_base_fragments(vec![make_sf(b"E1", b"F")], 0, 0);
        m2.append_delta_fragments(vec![make_sf(b"F", b"G")], 0, 0);
        assert!(m1.merge(&m2).is_none());
        assert_eq!(m1.len(), 5);
        assert_eq!(m1.base_fragments.len(), 3);
        assert_eq!(m1.delta_fragments.len(), 2);

        // delta fragments with overlap 1
        let mut m1 = ShardManifest::default();
        m1.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        m1.append_delta_fragments(vec![make_sf(b"D", b"E")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_base_fragments(vec![make_sf(b"E", b"F")], 0, 0);
        m2.append_delta_fragments(vec![make_sf(b"F", b"G")], 0, 0);
        assert!(m1.merge(&m2).is_some());

        // delta fragments with overlap 2
        let mut m1 = ShardManifest::default();
        m1.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        m1.append_delta_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        let mut m2 = ShardManifest::default();
        m2.append_base_fragments(vec![make_sf(b"B", b"D")], 0, 0);
        m2.append_delta_fragments(vec![make_sf(b"B", b"D")], 0, 0);
        assert!(m1.merge(&m2).is_some());
    }

    #[test]
    fn test_append_fragments() {
        // Confirms fragment append behavior for base and delta layers.
        let mut m = ShardManifest::default();
        m.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"D")], 0, 0);
        assert_eq!(m.len(), 2);
        assert_eq!(m.base_fragments.len(), 2);
        assert_eq!(m.delta_fragments.len(), 0);

        m.append_base_fragments(vec![make_sf(b"C", b"E")], 0, 0);
        assert_eq!(m.len(), 3);
        assert_eq!(m.base_fragments.len(), 3);

        m.append_delta_fragments(vec![make_sf(b"A", b"B")], 0, 0);
        assert_eq!(m.len(), 4);
        assert_eq!(m.delta_fragments.len(), 1);

        m.append_delta_fragments(vec![make_sf(b"B", b"D")], 0, 0);
        assert_eq!(m.len(), 5);
        assert_eq!(m.delta_fragments.len(), 2);

        m.append_base_fragments(vec![make_sf(b"E", b"F")], 0, 0);
        assert_eq!(m.len(), 6);
        assert_eq!(m.base_fragments.len(), 4);
    }

    #[test]
    fn test_switch_base_to_delta_preserves_fragment_order() {
        let mut manifest = ShardManifest::default();
        manifest.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"C")], 0, 0);
        manifest.append_base_fragments(vec![make_sf(b"C", b"D")], 0, 0);
        manifest.append_delta_fragments(vec![make_sf(b"D", b"E"), make_sf(b"E", b"F")], 0, 0);

        let before = manifest
            .get_fragments()
            .into_iter()
            .map(|frag| frag.frag_path.clone())
            .collect::<Vec<_>>();

        assert!(manifest.switch_base_to_delta());
        assert!(manifest.base_slice().is_empty());
        assert_eq!(manifest.delta_slice().len(), before.len());

        let after = manifest
            .get_fragments()
            .into_iter()
            .map(|frag| frag.frag_path.clone())
            .collect::<Vec<_>>();
        assert_eq!(after, before);
        assert!(!manifest.switch_base_to_delta());
    }

    #[test]
    fn test_serde_round_trip() {
        // Ensures manifest serializes and deserializes without data loss.
        let mut manifest = ShardManifest::new_for_first();
        manifest.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"C")], 0, 0);
        manifest.append_delta_fragments(vec![make_sf(b"D", b"E"), make_sf(b"E", b"F")], 0, 0);
        manifest.append_delta_fragments(vec![make_sf(b"G", b"H"), make_sf(b"H", b"I")], 0, 0);

        let json = serde_json::to_string(&manifest).unwrap();
        let deserialized: ShardManifest = serde_json::from_str(&json).unwrap();

        assert_eq!(manifest, deserialized);
        assert_eq!(deserialized.base_fragments.len(), 2);
        assert_eq!(deserialized.delta_fragments.len(), 4);
    }

    #[test]
    fn test_manifest_serde_compatibility() {
        // Verifies compatibility with legacy first_l1 while persisting only the
        // current b boundary.
        let mut manifest = ShardManifest::new_for_first();
        manifest.append_base_fragments(vec![make_sf(b"A", b"B"), make_sf(b"B", b"C")], 0, 0);
        manifest.append_delta_fragments(vec![make_sf(b"D", b"E"), make_sf(b"E", b"F")], 0, 0);
        manifest.append_delta_fragments(vec![make_sf(b"G", b"H")], 0, 0);

        let json_str = serde_json::to_string(&manifest).unwrap();
        let json_val: serde_json::Value = serde_json::from_str(&json_str).unwrap();
        let obj = json_val.as_object().unwrap();

        assert_eq!(obj.len(), 2);
        assert_eq!(obj["b"].as_u64().unwrap(), 2);
        assert!(obj.contains_key("fragments"));
        assert!(!obj.contains_key("first_l0"));
        assert!(!obj.contains_key("first_l1"));

        let deserialized: ShardManifest = serde_json::from_value(json_val.clone()).unwrap();
        assert_eq!(manifest, deserialized);

        let mut both_val = json_val.clone();
        if let Some(obj) = both_val.as_object_mut() {
            let b = obj.get("b").cloned().unwrap();
            obj.insert("first_l1".to_string(), b);
        }
        let both: ShardManifest = serde_json::from_value(both_val).unwrap();
        assert_eq!(manifest, both);

        let mut legacy_val = json_val.clone();
        if let Some(obj) = legacy_val.as_object_mut() {
            let b = obj.remove("b").unwrap();
            obj.insert("first_l1".to_string(), b);
        }
        let legacy: ShardManifest = serde_json::from_value(legacy_val).unwrap();
        assert_eq!(manifest, legacy);
    }

    #[test]
    fn test_splice_delta_windows() {
        // Validates splice behavior across arbitrary delta windows.
        let mut m = ShardManifest::new_for_first();
        m.append_base_fragments(vec![make_sf(b"A", b"B")], 0, 0);
        m.append_delta_fragments(
            vec![
                make_sf(b"C", b"D"),
                make_sf(b"D", b"E"),
                make_sf(b"E", b"F"),
                make_sf(b"F", b"G"),
                make_sf(b"G", b"H"),
            ],
            0,
            0,
        );

        let input = m.delta_fragments[1..4].to_vec();
        let output = vec![SplittableFrag::new(make_sf(b"D", b"G"), 0, 0)];
        m.splice(&input, output).unwrap();
        assert_eq!(m.delta_fragments.len(), 3);

        let input = m.delta_fragments[0..2].to_vec();
        let output = vec![SplittableFrag::new(make_sf(b"C", b"G"), 0, 0)];
        m.splice(&input, output).unwrap();
        assert_eq!(m.delta_fragments.len(), 2);
    }

    #[test]
    fn test_splice_base_and_errors() {
        // Validates splice on base fragments and error handling for invalid inputs.
        let mut m = ShardManifest::new_for_first();
        m.append_base_fragments(
            vec![
                make_sf(b"A", b"B"),
                make_sf(b"B", b"C"),
                make_sf(b"C", b"D"),
            ],
            0,
            0,
        );
        m.append_delta_fragments(vec![make_sf(b"E", b"F")], 0, 0);

        let input = m.base_fragments[0..2].to_vec();
        let output = vec![SplittableFrag::new(make_sf(b"A", b"C"), 0, 0)];
        m.splice(&input, output).unwrap();
        assert_eq!(m.base_fragments.len(), 2);

        let input = vec![SplittableFrag::new(make_sf(b"X", b"Y"), 0, 0)];
        let output = vec![SplittableFrag::new(make_sf(b"X", b"Z"), 0, 0)];
        assert!(m.splice(&input, output).is_err());

        let wrong_input = vec![
            m.base_fragments[0].clone(),
            SplittableFrag::new(make_sf(b"X", b"Y"), 0, 0),
        ];
        let output = vec![SplittableFrag::new(make_sf(b"A", b"D"), 0, 0)];
        assert!(m.splice(&wrong_input, output).is_err());
    }

    #[test]
    fn test_compute_split_points_cases() {
        // Covers weighted average, scan ratio, handle selection, overlap, and fallback
        // cases.
        {
            let mut manifest = ShardManifest::default();
            manifest.append_delta_fragments(
                vec![
                    make_frag_with_handles(&[0x00], &[0xFF], &[&[0x10]], 10, 10),
                    make_frag_with_handles(&[0x00], &[0xFF], &[&[0x20]], 10, 10),
                    make_frag_with_handles(&[0x00], &[0xFF], &[&[0x30]], 10, 10),
                ],
                0,
                0,
            );
            let range = HandleRange::new(vec![0x00], vec![0xFF]);
            let split_points = manifest.compute_split_points(&range, 1.0).unwrap();
            // Equal weights average 0x10/0x20/0x30 -> 0x20.
            assert_eq!(split_points, vec![RangeBound::Key(vec![0x20])]);
        }

        {
            let mut manifest = ShardManifest::default();
            manifest.append_delta_fragments(
                vec![
                    make_frag_with_handles(&[0x00], &[0xFF], &[&[0x10]], 10, 1),
                    make_frag_with_handles(&[0x00], &[0xFF], &[&[0xF0]], 10, 1),
                    make_frag_with_handles(&[0x00], &[0xFF], &[&[0x20]], 10, 1),
                ],
                0,
                0,
            );
            let range = HandleRange::new(vec![0x00], vec![0xFF]);
            let split_points = manifest.compute_split_points(&range, 0.5).unwrap();
            // Scan ratio 0.5 covers first two frags: midpoint(0x10, 0xF0) -> 0x80.
            assert_eq!(split_points, vec![RangeBound::Key(vec![0x80])]);
        }

        {
            let mut manifest = ShardManifest::default();
            manifest.append_delta_fragments(
                vec![make_frag_with_handles(
                    &[0x00],
                    &[0xFF],
                    &[&[0x10], &[0x20], &[0x30]],
                    10,
                    10,
                )],
                0,
                0,
            );
            let range = HandleRange::new(vec![0x00], vec![0xFF]);
            let split_points = manifest.compute_split_points(&range, 1.0).unwrap();
            // Odd handle list picks the middle handle 0x20.
            assert_eq!(split_points, vec![RangeBound::Key(vec![0x20])]);
        }

        {
            let mut manifest = ShardManifest::default();
            manifest.append_delta_fragments(
                vec![make_frag_with_handles(
                    &[0x10],
                    &[0x40],
                    &[&[0x10], &[0x20], &[0x30], &[0x40]],
                    10,
                    10,
                )],
                0,
                0,
            );
            let range = HandleRange::new(vec![0x10], vec![0x40]);
            let split_points = manifest.compute_split_points(&range, 1.0).unwrap();
            // Even handle list uses midpoint(0x20, 0x30) -> 0x28.
            assert_eq!(split_points, vec![RangeBound::Key(vec![0x28])]);
        }

        {
            let mut manifest = ShardManifest::default();
            manifest.append_delta_fragments(
                vec![make_frag_with_handles(
                    &[0x10],
                    &[0x60],
                    &[&[0x10], &[0x20], &[0x50], &[0x60]],
                    10,
                    10,
                )],
                0,
                0,
            );
            let range = HandleRange::new(vec![0x30], vec![0x40]);
            let split_points = manifest.compute_split_points(&range, 1.0).unwrap();
            // No handles in range; midpoint of overlap [0x30, 0x40] -> 0x38.
            assert_eq!(split_points, vec![RangeBound::Key(vec![0x38])]);
        }

        {
            let mut manifest = ShardManifest::default();
            manifest.append_delta_fragments(
                vec![make_frag_with_handles(&[0x00], &[0x10], &[&[0x08]], 10, 10)],
                0,
                0,
            );
            let range = HandleRange::new(vec![0x30], vec![0x40]);
            let split_points = manifest.compute_split_points(&range, 1.0).unwrap();
            // No overlap; fallback to shard midpoint of [0x30, 0x40] -> 0x38.
            assert_eq!(split_points, vec![RangeBound::Key(vec![0x38])]);
        }

        {
            let mut manifest = ShardManifest::default();
            manifest.append_delta_fragments(
                vec![make_frag_with_handles(&[0x00], &[0xFF], &[&[0x20]], 10, 0)],
                0,
                0,
            );
            let range = HandleRange::new(vec![0x00], vec![0xFF]);
            let split_points = manifest.compute_split_points(&range, 1.0).unwrap();
            // All weights zero; fallback midpoint of [0x00, 0xFF] -> 0x7F.
            assert_eq!(split_points, vec![RangeBound::Key(vec![0x7F])]);
        }

        {
            let manifest = ShardManifest::default();
            let range = HandleRange::new(vec![b'a'], vec![b'b']);
            let split_points = manifest.compute_split_points(&range, 1.0).unwrap();
            // Midpoint of ['a', 'b'] rounds to 'a', so we drop it as a boundary split.
            assert_eq!(split_points, Vec::<RangeBound>::new());
        }
    }
}