use core::borrow::Borrow;
#[cfg(feature = "bytemuck")]
use core::marker::PhantomData;
use std::{
    fmt::Debug,
    io::Result as IoResult,
    iter::FilterMap,
    ops::{Bound, RangeBounds},
};

use rayon::prelude::*;
use sled::{CompareAndSwapError, CompareAndSwapSuccess, InlineArray};
use tracing::debug;

use super::{
    ByteValue, Iter, KeyValue,
    error::{DbError, DbResult},
    replication::ChangeEvent,
    tree::MIN_PAR_SIZE,
};
#[cfg(feature = "bytemuck")]
use crate::database::byte_as_slice::BytesAsValSlice;
use crate::{Cid, CollectionTree, Entry, Record, database::updater::Updater, utils::MaybeParIter};

pub type IterCollection<T> = FilterMap<Iter, fn(IoResult<KeyValue>) -> Option<T>>;
pub type ItemWithRaw<Item> = (ByteValue, ByteValue, Entry<Item>);

// update methods
impl<Item> crate::CollectionTree<Item>
where
    Item: Record<Value = InlineArray> + Debug,
{
    /// Insert or update value by id.
    /// Return old value if exist
    #[doc(alias = "insert")]
    #[doc(alias = "update")]
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    pub fn upsert(
        &self,
        id: impl Borrow<Item::SelfId> + Debug,
        item: &Item,
    ) -> DbResult<Option<Item>> {
        let bytes = Item::ser(item);
        let borrowed_id = id.borrow();

        let key_bytes = Cid::encode(borrowed_id);
        let key_bytes = InlineArray::from(key_bytes.as_ref());

        let old = self
            .raw
            .tree
            .insert(&key_bytes, bytes.clone())
            .map(|i| i.map(|item| Item::deser(&item)))?;

        self.update_indexes(borrowed_id, old.as_ref(), Some(item));
        self.upsert_event(borrowed_id, old.as_ref(), item);
        self.invalidate_hash(borrowed_id);
        self.send_event(|| ChangeEvent::Upsert((key_bytes, bytes)));

        Ok(old)
    }

    /// Update, create or delete value by id with Fn. Can call closure multiple times.
    ///
    /// Return new value and closure result
    #[tracing::instrument(level = "trace", skip(self, f), fields(name = Item::NAME), ret)]
    pub fn upsert_fn<F, Ret>(
        &self,
        id: impl Borrow<Item::SelfId> + Debug,
        f: F,
    ) -> DbResult<(Option<Item>, Ret)>
    where
        F: Fn(Option<Item>) -> (Option<Item>, Ret),
        Item: Clone,
        Ret: Debug,
    {
        let borrowed_id = id.borrow();
        let key = Cid::encode(borrowed_id);
        let mut old_bytes = self.raw.tree.get(&key)?;

        loop {
            let old_ref = old_bytes.as_ref();
            let old = old_ref.map(|old| Item::deser(old));
            let old_cloned = old.clone();

            let (new, ret) = f(old);
            let new_bytes = new.as_ref().map(|new| Item::ser(new));

            let res = self.raw.tree.compare_and_swap(&key, old_ref, new_bytes)?;

            match res {
                Ok(CompareAndSwapSuccess {
                    new_value,
                    previous_value,
                }) => {
                    if new_value == previous_value {
                        return Ok((new, ret));
                    }
                    self.update_indexes(borrowed_id, old_cloned.as_ref(), new.as_ref());
                    self.invalidate_hash(borrowed_id);

                    self.send_event(|| match new_value {
                        Some(bytes) => ChangeEvent::Upsert((ByteValue::from(key.as_ref()), bytes)),
                        None => ChangeEvent::Delete(ByteValue::from(key.as_ref())),
                    });

                    match &new {
                        Some(new) => {
                            self.upsert_event(borrowed_id, old_cloned.as_ref(), new);
                        }
                        None => {
                            if let Some(old) = old_cloned {
                                self.remove_event(borrowed_id, &old);
                            }
                        }
                    }
                    return Ok((new, ret));
                }
                Err(CompareAndSwapError {
                    current,
                    proposed: _,
                }) => {
                    old_bytes = current;
                }
            }
        }
    }

    /// Update item with closure if exist, return new value. Tried only once.
    #[tracing::instrument(level = "trace", skip(self, f), fields(name = Item::NAME), ret)]
    pub fn update_fn<F>(&self, id: impl Borrow<Item::SelfId> + Debug, f: F) -> DbResult<Item>
    where
        F: FnOnce(Item) -> DbResult<Item>,
        Item: Clone,
    {
        let borrowed_id = id.borrow();
        let key = Cid::encode(borrowed_id);
        let old_bytes = self.raw.tree.get(&key)?;
        let old = old_bytes.as_ref().map(|old| Item::deser(old));

        match (old, old_bytes) {
            (Some(old), Some(old_bytes)) => {
                let old_cloned = old.clone();

                let new = f(old)?;

                let new_bytes = Item::ser(&new);

                let res = self.raw.tree.compare_and_swap(
                    &key,
                    Some(old_bytes.as_ref()),
                    Some(new_bytes),
                )?;

                match res {
                    Ok(CompareAndSwapSuccess {
                        new_value,
                        previous_value: _,
                    }) => {
                        let new_bytes = new_value.expect("new value should be present");
                        self.invalidate_hash(borrowed_id);
                        self.update_indexes(borrowed_id, Some(&old_cloned), Some(&new));
                        self.upsert_event(borrowed_id, Some(&old_cloned), &new);
                        self.send_event(|| {
                            let key_bytes = ByteValue::from(key.as_ref());
                            ChangeEvent::Upsert((key_bytes, new_bytes))
                        });

                        Ok(new)
                    }
                    Err(_) => Err(DbError::Transaction),
                }
            }
            _ => Err(DbError::NotFound),
        }
    }

    pub fn updater(&self, id: Item::SelfId) -> DbResult<Updater<Item>>
    where
        Item: Clone,
    {
        let key = Cid::encode(&id);
        let val = self.tree.get(key)?.ok_or(DbError::NotFound)?;
        let item = Item::deser(&val);
        let updater = Updater {
            id,
            val: item.clone(),
            bytes: val,
            old: item,
        };
        Ok(updater)
    }

    /// Tried only once.
    pub fn apply_updater(&self, updater: Updater<Item>) -> DbResult<Entry<Item>> {
        let Updater {
            id,
            val,
            bytes,
            old,
        } = updater;
        let borrowed_id = id.borrow();
        let key = Cid::encode(borrowed_id);
        let new = val.ser();
        let CompareAndSwapSuccess {
            new_value: _,
            previous_value: _,
        } = self
            .tree
            .compare_and_swap(&key, Some(bytes), Some(new.clone()))?
            .map_err(|_| DbError::Transaction)?;

        self.invalidate_hash(borrowed_id);
        self.update_indexes(borrowed_id, Some(&old), Some(&val));
        self.upsert_event(borrowed_id, Some(&old), &val);
        let key_bytes = ByteValue::from(key.as_ref());
        let event = ChangeEvent::Upsert((key_bytes, new));
        self.send_event(|| event);

        Ok(Entry::new(id, val))
    }

    /// change status in item key with closure,
    /// if key not exist, does nothing.
    /// not use for change id!
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME), err)]
    #[deprecated]
    pub fn change_status(&self, old_id: &Item::SelfId, new_id: &Item::SelfId) -> DbResult<()> {
        let old_key_bytes = Cid::encode(old_id);
        let new_key_bytes = Cid::encode(new_id);
        // TODO: run in transaction
        let value = self
            .raw
            .tree
            .remove(&old_key_bytes)?
            .ok_or(DbError::NotFound)?;
        let item = Item::deser(&value);
        if let Some(val) = self.raw.tree.insert(&new_key_bytes, value)? {
            let item = Item::deser(&val);
            tracing::error!(?item, "unexpected value");
        };
        self.update_indexes(old_id, Some(&item), None);
        self.update_indexes(new_id, None, Some(&item));
        self.id_change_event(&item, old_id, new_id);
        self.invalidate_hash(old_id);
        self.invalidate_hash(new_id);
        self.send_event(|| {
            let old_key_bytes = ByteValue::from(old_key_bytes.as_ref());
            let new_key_bytes = ByteValue::from(new_key_bytes.as_ref());
            ChangeEvent::ChangeId(old_key_bytes, new_key_bytes)
        });
        Ok(())
    }

    /// permanently remove item. Return old value or None if not exist
    #[doc(alias = "delete")]
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    pub fn remove(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<Option<Item>> {
        let borrowed_id = id.borrow();
        let key_bytes = Cid::encode(borrowed_id);

        let res = self
            .raw
            .tree
            .remove(&key_bytes)
            .map(|res| res.map(|val| Item::deser_owned(val)))
            .map_err(DbError::from);

        if let Ok(Some(item)) = &res {
            self.update_indexes(borrowed_id, Some(item), None);
            self.remove_event(borrowed_id, item);
            self.invalidate_hash(borrowed_id);
            self.send_event(|| {
                let key_bytes = ByteValue::from(key_bytes.as_ref());
                ChangeEvent::Delete(key_bytes)
            });
        }
        res
    }

    #[deprecated]
    pub fn remove_item(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<Option<Item>> {
        self.remove(id)
    }

    /// Remove item to other ('removed') tree. Return old value or None if not exist
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    pub fn soft_remove(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<Option<Item>> {
        let borrowed_id = id.borrow();
        let key_bytes = Cid::encode(borrowed_id);

        let res = self.raw.tree.remove(&key_bytes)?;

        match res {
            Some(res) => {
                let item = Item::deser(&res);
                self.update_indexes(borrowed_id, Some(&item), None);
                self.remove_event(borrowed_id, &item);
                self.invalidate_hash(borrowed_id);
                self.send_event(|| {
                    let key_bytes = ByteValue::from(key_bytes.as_ref());
                    ChangeEvent::Delete(key_bytes)
                });
                self.removed.insert(key_bytes, res)?;
                Ok(Some(item))
            }
            None => Ok(None),
        }
    }
}

// read methods
impl<Item> CollectionTree<Item>
where
    Item: Record<Value = InlineArray> + Debug,
{
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    fn get_by_slice(&self, key: &[u8]) -> DbResult<Option<Item>> {
        // self.raw.tree.get_zero_copy(key, |bytes| bytes.map(Self::deser)).map_err(DbError::from)
        let val = self.raw.tree.get(key)?;
        match val {
            Some(val) => Ok(Some(Item::deser_owned(val))),
            None => Ok(None),
        }
    }

    /// get value by id
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    pub fn get(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<Option<Item>> {
        let key = Cid::encode(id.borrow());
        self.get_by_slice(key.as_ref())
    }

    #[deprecated]
    pub fn get_item(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<Option<Item>> {
        self.get(id)
    }

    /// get value by id or return not found
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    pub fn get_or_err(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<Item> {
        let key = Cid::encode(id.borrow());
        self.get_by_slice(key.as_ref())
            .and_then(|item| item.ok_or(DbError::NotFound))
    }

    /// check exist by id
    #[doc(alias = "exist")]
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    pub fn contains(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<bool> {
        let key = Cid::encode(id.borrow());
        // self.raw.tree.get_zero_copy(key, |v| v.is_some()).map_err(DbError::from)
        self.raw
            .tree
            .contains_key(key.as_ref())
            .map_err(DbError::from)
    }

    #[deprecated]
    pub fn contains_item(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<bool> {
        self.contains(id)
    }

    /// get iterator for collection
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn iter(&self) -> IterCollection<Entry<Item>> {
        self.range_iter::<&[u8], _>(..)
    }

    /// get iterator by prefix for collection
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn scan_prefix<P>(&self, prefix: P) -> IterCollection<Entry<Item>>
    where
        P: AsRef<[u8]> + Debug,
    {
        let from = prefix.as_ref();
        let mut upper = from.to_vec();

        // [0,0,100,100] -> [0,0,100,100]..[0,0,100,101]
        // [0,0,100,255] -> [0,0,100,255]..[0,0,101]
        // [255,255] -> [255,255]..
        while let Some(last) = upper.pop() {
            if last < u8::MAX {
                upper.push(last + 1);
                return self.range_iter(from..upper.as_ref());
            }
        }

        self.range_iter(from..)
    }

    /// get iterator by range for collection
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn range_iter<K, R>(&self, range: R) -> IterCollection<Entry<Item>>
    where
        K: AsRef<[u8]>,
        R: RangeBounds<K> + Debug,
    {
        self.raw.tree.range(range).filter_map(|item| match item {
            Ok((key, value)) => {
                let id = Item::deser_key(&key);

                let value = Item::deser_owned(value);
                Some(Entry {
                    key: id,
                    val: value,
                })
            }
            Err(e) => {
                tracing::error!(%e);
                None
            }
        })
    }

    /// get iterator by range for collection
    /// if item error, return Error
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn try_range_iter<K, R>(
        &self,
        range: R,
    ) -> impl DoubleEndedIterator<Item = DbResult<Entry<Item>>> + use<K, R, Item>
    where
        K: AsRef<[u8]>,
        R: RangeBounds<K> + Debug,
    {
        self.raw.tree.range(range).map(|item| {
            item.map_err(DbError::from).map(|item| {
                let (key, value) = item;
                let id = Item::deser_key(&key);
                let val = Item::deser_owned(value);
                Entry { key: id, val }
            })
        })
    }

    /// get iterator by range for collection
    /// if item error, return Error
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn try_range_iter_with_raw<K, R>(
        &self,
        range: R,
    ) -> impl DoubleEndedIterator<Item = DbResult<ItemWithRaw<Item>>> + use<K, R, Item>
    where
        K: AsRef<[u8]>,
        R: RangeBounds<K> + Debug,
    {
        self.raw.tree.range(range).map(|item| {
            item.map_err(DbError::from).map(|item| {
                let (key, value) = item;
                let id = Item::deser_key(&key);
                let val = Item::deser(&value);
                (key, value, Entry { key: id, val })
            })
        })
    }

    /// iter without deserialize value
    #[tracing::instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn iter_keys<P>(
        &self,
        prefix: P,
    ) -> impl DoubleEndedIterator<Item = Item::SelfId> + use<P, Item>
    where
        P: AsRef<[u8]> + Debug,
    {
        self.raw
            .tree
            .scan_prefix(prefix)
            .filter_map(|res| match res {
                Ok((key, _)) => Some(Item::deser_key(&key)),
                Err(e) => {
                    tracing::error!(%e);
                    None
                }
            })
    }

    /// Split collection into groups for parallel processing
    ///
    /// used rayon for parallelism
    #[tracing::instrument(level = "debug", skip(self), fields(name = Item::NAME), err)]
    pub fn par_iter(
        &self,
        seq: Option<u64>,
    ) -> DbResult<MaybeParIter<impl IndexedParallelIterator<Item = (Iter, Self)> + '_, Iter>> {
        if self.raw.tree.is_empty()? {
            return Ok(MaybeParIter::Empty);
        }

        let seq = seq.unwrap_or_else(|| self.count()) as usize;

        if seq < MIN_PAR_SIZE {
            let res = self.raw.tree.iter();
            return Ok(MaybeParIter::Seq(res));
        }

        // 10k items / 512 = 20
        let chunk_count = seq / MIN_PAR_SIZE;

        let cpus = num_cpus::get();
        // 20 or threads count
        let workers_count = chunk_count.min(cpus);

        if workers_count < 2 {
            let res = self.raw.tree.iter();
            return Ok(MaybeParIter::Seq(res));
        }

        let (first, _) = self.raw.tree.first()?.expect("tree len < 2");
        let (last, _) = self.raw.tree.last()?.expect("tree len < 2");

        let first = Item::deser_key(first.as_ref());
        let last = Item::deser_key(last.as_ref());

        let first_prefix = first.group_id();
        let last_prefix = last.group_id();

        let diff = last_prefix - first_prefix;

        if diff < 2 {
            let res = self.raw.tree.iter();
            return Ok(MaybeParIter::Seq(res));
        }

        let step = diff / (workers_count as u32);

        if step == 0 {
            let res = self.raw.tree.iter();
            return Ok(MaybeParIter::Seq(res));
        }

        let mut bounds_arr = Vec::with_capacity(workers_count);

        let mut start = first_prefix;

        while start <= last_prefix {
            let end = start + step;

            let start_bound = Bound::Included(start.to_be_bytes());
            let end_bound = Bound::Excluded(end.to_be_bytes());
            let bounds = (start_bound, end_bound);

            // last step
            // if end + step > last_prefix {
            //     let end_bound = Bound::Unbounded;
            //     let bounds = (start_bound, end_bound);
            //     let range = self.raw.tree.range(bounds);
            //     arr.push(range);
            //     break;
            // }

            bounds_arr.push(bounds);
            start = end;
        }

        tracing::info!(workers_count, step, first_prefix, last_prefix, "par_iter");
        debug!(?bounds_arr);

        let arr = bounds_arr
            .into_iter()
            .map(|bounds| {
                let range = self.raw.tree.range(bounds);
                (range, self.clone())
            })
            .collect::<Vec<_>>();

        let res = arr.into_par_iter();

        Ok(MaybeParIter::Par(res))
    }

    #[cfg(feature = "bytemuck")]
    pub fn cast_slice<K: AsRef<[u8]>>(
        &self,
        id: K,
    ) -> DbResult<Option<BytesAsValSlice<InlineArray, Item>>> {
        let val = self.raw.tree.get(id)?;
        match val {
            Some(val) => Ok(Some(BytesAsValSlice(val, PhantomData))),
            None => Ok(None),
        }
    }
}
