use std::borrow::Borrow;
use std::ops::Bound;
use std::path::{Path, PathBuf};

use common::counter::conditioned_counter::ConditionedCounter;
use common::counter::hardware_counter::HardwareCounterCell;
use common::counter::iterator_hw_measurement::HwMeasurementIteratorExt;
use common::fs::{atomic_save_json, clear_disk_cache, read_json};
use common::mmap;
use common::mmap::{AdviceSetting, MmapBitSlice, MmapSlice, create_and_ensure_length};
use common::types::PointOffsetType;
use common::universal_io::UniversalRead;
use common::universal_io::mmap::MmapUniversal;
use fs_err as fs;
use memmap2::MmapMut;
use serde::{Deserialize, Serialize};

use super::Encodable;
use super::mutable_numeric_index::InMemoryNumericIndex;
use crate::common::Flusher;
use crate::common::mmap_bitslice_buffered_update_wrapper::MmapBitSliceBufferedUpdateWrapper;
use crate::common::operation_error::{OperationError, OperationResult};
use crate::index::field_index::histogram::{Histogram, Numericable, Point};
use crate::index::field_index::stored_point_to_values::{StoredPointToValues, StoredValue};

const PAIRS_PATH: &str = "data.bin";
const DELETED_PATH: &str = "deleted.bin";
const CONFIG_PATH: &str = "mmap_field_index_config.json";

pub struct MmapNumericIndex<T: Encodable + Numericable + Default + StoredValue + 'static> {
    path: PathBuf,
    pub(super) storage: Storage<T, MmapUniversal<u8>>,
    histogram: Histogram<T>,
    deleted_count: usize,
    max_values_per_point: usize,
    is_on_disk: bool,
}

pub(super) struct Storage<
    T: Encodable + Numericable + Default + StoredValue + 'static,
    S: UniversalRead<u8>,
> {
    deleted: MmapBitSliceBufferedUpdateWrapper,
    // sorted pairs (id + value), sorted by value (by id if values are equal)
    pairs: MmapSlice<Point<T>>,
    pub(super) point_to_values: StoredPointToValues<T, S>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct MmapNumericIndexConfig {
    max_values_per_point: usize,
}

pub(super) struct NumericIndexPairsIterator<'a, T: Encodable + Numericable> {
    pairs: &'a [Point<T>],
    deleted: &'a MmapBitSliceBufferedUpdateWrapper,
    start_index: usize,
    end_index: usize,
}

impl<T: Encodable + Numericable> Iterator for NumericIndexPairsIterator<'_, T> {
    type Item = Point<T>;

    fn next(&mut self) -> Option<Self::Item> {
        while self.start_index < self.end_index {
            let key = self.pairs[self.start_index].clone();
            let deleted = self.deleted.get(key.idx as usize).unwrap_or(true);
            self.start_index += 1;
            if deleted {
                continue;
            }
            return Some(key);
        }
        None
    }
}

impl<T: Encodable + Numericable> DoubleEndedIterator for NumericIndexPairsIterator<'_, T> {
    fn next_back(&mut self) -> Option<Self::Item> {
        while self.start_index < self.end_index {
            let key = self.pairs[self.end_index - 1].clone();
            let deleted = self.deleted.get(key.idx as usize).unwrap_or(true);
            self.end_index -= 1;
            if deleted {
                continue;
            }
            return Some(key);
        }
        None
    }
}

impl<T: Encodable + Numericable + Default + StoredValue> MmapNumericIndex<T> {
    pub fn build(
        in_memory_index: InMemoryNumericIndex<T>,
        path: &Path,
        is_on_disk: bool,
    ) -> OperationResult<Self> {
        fs::create_dir_all(path)?;

        let pairs_path = path.join(PAIRS_PATH);
        let deleted_path = path.join(DELETED_PATH);
        let config_path = path.join(CONFIG_PATH);

        atomic_save_json(
            &config_path,
            &MmapNumericIndexConfig {
                max_values_per_point: in_memory_index.max_values_per_point,
            },
        )?;

        in_memory_index.histogram.save(path)?;

        StoredPointToValues::<T, MmapUniversal<u8>>::from_iter(
            path,
            in_memory_index
                .point_to_values
                .iter()
                .enumerate()
                .map(|(idx, values)| (idx as PointOffsetType, values.iter().map(|v| v.borrow()))),
        )?;

        {
            let pairs_file = create_and_ensure_length(
                &pairs_path,
                in_memory_index.map.len() * std::mem::size_of::<Point<T>>(),
            )?;
            let pairs_mmap = unsafe { MmapMut::map_mut(&pairs_file)? };
            let mut pairs = unsafe { MmapSlice::<Point<T>>::try_from(pairs_mmap)? };
            for (src, dst) in in_memory_index.map.iter().zip(pairs.iter_mut()) {
                *dst = src.clone();
            }
        }

        {
            const BITS_IN_BYTE: usize = 8;
            let deleted_flags_count = in_memory_index.point_to_values.len();
            let deleted_file = create_and_ensure_length(
                &deleted_path,
                BITS_IN_BYTE
                    * BITS_IN_BYTE
                    * deleted_flags_count.div_ceil(BITS_IN_BYTE * BITS_IN_BYTE),
            )?;
            let mut deleted_mmap = unsafe { MmapMut::map_mut(&deleted_file)? };
            deleted_mmap.fill(0);
            let mut deleted_bitflags = MmapBitSlice::from(deleted_mmap, 0);
            for (idx, values) in in_memory_index.point_to_values.iter().enumerate() {
                if values.is_empty() {
                    deleted_bitflags.set(idx, true);
                }
            }
        }

        Self::open(path, is_on_disk)?.ok_or_else(|| {
            OperationError::service_error("Failed to open MmapNumericIndex after building it")
        })
    }

    /// Open and load mmap numeric index from the given path
    pub fn open(path: &Path, is_on_disk: bool) -> OperationResult<Option<Self>> {
        let pairs_path = path.join(PAIRS_PATH);
        let deleted_path = path.join(DELETED_PATH);
        let config_path = path.join(CONFIG_PATH);

        // If config doesn't exist, assume the index doesn't exist on disk
        if !config_path.is_file() {
            return Ok(None);
        }

        let histogram = Histogram::<T>::load(path)?;
        let config: MmapNumericIndexConfig = read_json(&config_path)?;
        let deleted = mmap::open_write_mmap(&deleted_path, AdviceSetting::Global, false)?;
        let deleted = MmapBitSlice::from(deleted, 0);
        let deleted_count = deleted.count_ones();
        let do_populate = !is_on_disk;
        let map = unsafe {
            MmapSlice::try_from(mmap::open_write_mmap(
                &pairs_path,
                AdviceSetting::Global,
                do_populate,
            )?)?
        };
        let point_to_values = StoredPointToValues::open(path, do_populate)?;

        Ok(Some(Self {
            path: path.to_path_buf(),
            storage: Storage {
                pairs: map,
                deleted: MmapBitSliceBufferedUpdateWrapper::new(deleted),
                point_to_values,
            },
            histogram,
            deleted_count,
            max_values_per_point: config.max_values_per_point,
            is_on_disk,
        }))
    }

    pub fn wipe(self) -> OperationResult<()> {
        let files = self.files();
        let path = self.path.clone();
        // drop mmap handles before deleting files
        drop(self);
        for file in files {
            fs::remove_file(file)?;
        }
        let _ = fs::remove_dir(path);
        Ok(())
    }

    pub fn files(&self) -> Vec<PathBuf> {
        let mut files = vec![
            self.path.join(PAIRS_PATH),
            self.path.join(DELETED_PATH),
            self.path.join(CONFIG_PATH),
        ];
        files.extend(self.storage.point_to_values.files());
        files.extend(Histogram::<T>::files(&self.path));
        files
    }

    pub fn immutable_files(&self) -> Vec<PathBuf> {
        let mut files = vec![self.path.join(PAIRS_PATH), self.path.join(CONFIG_PATH)];
        files.extend(self.storage.point_to_values.immutable_files());
        files.extend(Histogram::<T>::immutable_files(&self.path));
        files
    }

    pub fn flusher(&self) -> Flusher {
        self.storage.deleted.flusher()
    }

    pub fn check_values_any(
        &self,
        idx: PointOffsetType,
        check_fn: impl Fn(&T) -> bool,
        hw_counter: &HardwareCounterCell,
    ) -> OperationResult<bool> {
        let hw_counter = self.make_conditioned_counter(hw_counter);

        if self.storage.deleted.get(idx as usize) == Some(false) {
            self.storage
                .point_to_values
                .check_values_any(idx, |v| check_fn(v), &hw_counter)
        } else {
            Ok(false)
        }
    }

    pub fn get_values(&self, idx: PointOffsetType) -> Option<Box<dyn Iterator<Item = T> + '_>> {
        if self.storage.deleted.get(idx as usize) == Some(false) {
            Some(Box::new(
                self.storage
                    .point_to_values
                    // TODO: Propagate counter upwards
                    .values_iter(idx, ConditionedCounter::never())
                    .ok()??
                    .map(|v| *v),
            ))
        } else {
            None
        }
    }

    pub fn values_count(&self, idx: PointOffsetType) -> Option<usize> {
        if self.storage.deleted.get(idx as usize) == Some(false) {
            self.storage.point_to_values.get_values_count(idx).ok()?
        } else {
            None
        }
    }

    /// Returns the number of key-value pairs in the index.
    /// Note that is doesn't count deleted pairs.
    pub(super) fn total_unique_values_count(&self) -> usize {
        self.storage.pairs.len()
    }

    pub(super) fn values_range<'a>(
        &'a self,
        start_bound: Bound<Point<T>>,
        end_bound: Bound<Point<T>>,
        hw_counter: &'a HardwareCounterCell,
    ) -> impl Iterator<Item = PointOffsetType> + 'a {
        let hw_counter = self.make_conditioned_counter(hw_counter);

        self.values_range_iterator(start_bound, end_bound)
            .map(|Point { idx, .. }| idx)
            .measure_hw_with_condition_cell(hw_counter, size_of::<Point<T>>(), |i| {
                i.payload_index_io_read_counter()
            })
    }

    pub(super) fn orderable_values_range(
        &self,
        start_bound: Bound<Point<T>>,
        end_bound: Bound<Point<T>>,
    ) -> impl DoubleEndedIterator<Item = (T, PointOffsetType)> + '_ {
        self.values_range_iterator(start_bound, end_bound)
            .map(|Point { val, idx }| (val, idx))
    }

    pub fn remove_point(&mut self, idx: PointOffsetType) {
        let idx = idx as usize;
        if idx < self.storage.deleted.len() && !self.storage.deleted.get(idx).unwrap_or(true) {
            self.storage.deleted.set(idx, true);
            self.deleted_count += 1;
        }
    }

    pub(super) fn get_histogram(&self) -> &Histogram<T> {
        &self.histogram
    }

    pub(super) fn get_points_count(&self) -> usize {
        self.storage.point_to_values.len() - self.deleted_count
    }

    pub(super) fn get_max_values_per_point(&self) -> usize {
        self.max_values_per_point
    }

    pub(super) fn values_range_size(
        &self,
        start_bound: Bound<Point<T>>,
        end_bound: Bound<Point<T>>,
    ) -> usize {
        let iter = self.values_range_iterator(start_bound, end_bound);
        iter.end_index - iter.start_index
    }

    // get iterator
    fn values_range_iterator(
        &self,
        start_bound: Bound<Point<T>>,
        end_bound: Bound<Point<T>>,
    ) -> NumericIndexPairsIterator<'_, T> {
        let start_index = match start_bound {
            Bound::Included(bound) => self
                .storage
                .pairs
                .binary_search(&bound)
                .unwrap_or_else(|idx| idx),
            Bound::Excluded(bound) => match self.storage.pairs.binary_search(&bound) {
                Ok(idx) => idx + 1,
                Err(idx) => idx,
            },
            Bound::Unbounded => 0,
        };

        if start_index >= self.storage.pairs.len() {
            return NumericIndexPairsIterator {
                pairs: &self.storage.pairs,
                deleted: &self.storage.deleted,
                start_index: self.storage.pairs.len(),
                end_index: self.storage.pairs.len(),
            };
        }

        let end_index = match end_bound {
            Bound::Included(bound) => match self.storage.pairs[start_index..].binary_search(&bound)
            {
                Ok(idx) => idx + 1 + start_index,
                Err(idx) => idx + start_index,
            },
            Bound::Excluded(bound) => {
                let end_bound = self.storage.pairs[start_index..].binary_search(&bound);
                end_bound.unwrap_or_else(|idx| idx) + start_index
            }
            Bound::Unbounded => self.storage.pairs.len(),
        };

        NumericIndexPairsIterator {
            pairs: &self.storage.pairs,
            deleted: &self.storage.deleted,
            start_index,
            end_index,
        }
    }

    fn make_conditioned_counter<'a>(
        &self,
        hw_counter: &'a HardwareCounterCell,
    ) -> ConditionedCounter<'a> {
        ConditionedCounter::new(self.is_on_disk, hw_counter)
    }

    pub fn is_on_disk(&self) -> bool {
        self.is_on_disk
    }

    /// Populate all pages in the mmap.
    /// Block until all pages are populated.
    pub fn populate(&self) -> OperationResult<()> {
        self.storage.pairs.populate()?;
        self.storage.point_to_values.populate()?;
        Ok(())
    }

    /// Drop disk cache.
    pub fn clear_cache(&self) -> OperationResult<()> {
        let pairs_path = self.path.join(PAIRS_PATH);
        let deleted_path = self.path.join(DELETED_PATH);

        clear_disk_cache(&pairs_path)?;
        clear_disk_cache(&deleted_path)?;

        self.storage.point_to_values.clear_cache()?;

        Ok(())
    }
}
