use std::{
    borrow::Borrow,
    fmt::Debug,
    ops::{Deref, RangeBounds},
    sync::Arc,
};

use fjall::{Keyspace, KeyspaceCreateOptions, Slice};

use super::{
    ByteValue, ItemWithRaw, KeyValue, RawTree, builder::LogBuilder, db::LogDb, events::change::*,
    read::ReadTree,
};
use crate::{Cid, DbError, DbResult, Entry, Record, indexes::IndexUpdateCollection};
use arrayvec::ArrayVec;

pub type IndexesMap<Item> = ArrayVec<Box<dyn IndexUpdateCollection<Item>>, 8>;

#[derive(derive_more::Debug)]
pub struct LogTree<Item: Record> {
    pub inner: ReadTree<Item>,
    pub(crate) handlers: Arc<LogDbEventHandlers<Item>>,
}

impl<Item: Record> Clone for LogTree<Item> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            handlers: self.handlers.clone(),
        }
    }
}

impl<Item: Record> Deref for LogTree<Item> {
    type Target = ReadTree<Item>;

    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

impl<Item> LogTree<Item>
where
    Item: Record<Value = Slice> + std::fmt::Debug,
{
    #[doc(alias = "new")]
    #[doc(alias = "create")]
    pub fn open(db: &LogDb, options: Option<KeyspaceCreateOptions>) -> Self {
        LogBuilder::<Item>::new(db, options).build()
    }

    pub fn with_name(db: &LogDb, name: &str, options: Option<KeyspaceCreateOptions>) -> Self {
        LogBuilder::<Item>::with_name(db, name, options).build()
    }

    pub fn builder(db: &LogDb, options: Option<KeyspaceCreateOptions>) -> LogBuilder<Item> {
        LogBuilder::<Item>::new(db, options)
    }

    #[doc(alias = "inner")]
    pub fn raw(&self) -> &RawTree {
        &self.inner.raw
    }

    pub(crate) fn tree(&self) -> &Keyspace {
        &self.inner.raw.tree
    }

    #[inline]
    pub(super) fn invalidate_hash(&self, id: &Item::SelfId) {
        let group = id.group_id();
        self.inner.raw.inner.invalidate_hash(group);
    }

    /// update fn when update item body or create item
    #[inline]
    pub(super) fn upsert_event(&self, id: &Item::SelfId, new: &Item) {
        if let Some(f) = self.handlers.on_upsert.as_ref() {
            f(id, new);
        }
    }

    /// call when item removed (soft or hard)
    #[inline]
    pub(super) fn remove_event(&self, id: &Item::SelfId) {
        if let Some(f) = self.handlers.on_remove.as_ref() {
            f(id);
        }
    }
}

impl<Item: Record> From<LogTree<Item>> for RawTree {
    fn from(value: LogTree<Item>) -> Self {
        value.inner.raw
    }
}

// update methods
impl<Item> LogTree<Item>
where
    Item: Record<Value = Slice> + Debug,
{
    /// Insert or update value by id.
    #[doc(alias = "insert")]
    #[doc(alias = "update")]
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    pub fn upsert(&self, id: impl Borrow<Item::SelfId> + Debug, item: &Item) -> DbResult<()> {
        let start = std::time::Instant::now();
        let bytes = Item::ser(item);

        let key = Cid::encode(id.borrow());
        let key_bytes = ByteValue::from(key.as_ref());

        self.tree().insert(key_bytes, bytes)?;

        self.upsert_event(id.borrow(), item);
        self.invalidate_hash(id.borrow());

        histogram!("armour_logdb_tree_upsert_duration", "name" => self.raw().name.clone())
            .record(start.elapsed().as_secs_f64());
        counter!("armour_logdb_tree_upsert_total", "name" => self.raw().name.clone()).increment(1);

        Ok(())
    }

    /// permanently remove item
    #[doc(alias = "delete")]
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    pub fn remove(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<()> {
        let start = std::time::Instant::now();
        let key = Cid::encode(id.borrow());
        let key_bytes = ByteValue::from(key.as_ref());

        self.tree().remove(key_bytes).map_err(DbError::from)?;
        self.remove_event(id.borrow());
        self.invalidate_hash(id.borrow());

        histogram!("armour_logdb_tree_remove_duration", "name" => self.raw().name.clone())
            .record(start.elapsed().as_secs_f64());
        counter!("armour_logdb_tree_remove_total", "name" => self.raw().name.clone()).increment(1);

        Ok(())
    }

    /// Remove item to other ('removed') tree
    #[doc(alias = "delete")]
    #[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 start = std::time::Instant::now();
        let key = Cid::encode(id.borrow());
        let key_bytes = ByteValue::from(key.as_ref());
        let res = match self.tree().get(key.as_ref())? {
            Some(val) => {
                let item = Item::deser(&val);

                self.raw().removed.insert(key_bytes.clone(), val)?;
                self.remove_event(id.borrow());
                self.tree().remove(key_bytes).map_err(DbError::from)?;
                self.invalidate_hash(id.borrow());
                Ok(Some(item))
            }
            None => Ok(None),
        };

        histogram!("armour_logdb_tree_soft_remove_duration", "name" => self.raw().name.clone())
            .record(start.elapsed().as_secs_f64());
        counter!("armour_logdb_tree_soft_remove_total", "name" => self.raw().name.clone())
            .increment(1);

        res
    }

    /// doesn't change indexes
    pub fn apply_batch<K, V>(&self, iter: impl Iterator<Item = (K, Option<V>)>) -> DbResult<()>
    where
        K: AsRef<Item::SelfId>,
        V: AsRef<Item>,
    {
        let start = std::time::Instant::now();
        let res = self.raw().apply_batch(iter.map(|(id, item)| {
            let key_bytes = Cid::encode(id.as_ref());
            let key_bytes = ByteValue::from(key_bytes.as_ref());
            let val_bytes = item.map(|item| Item::ser(item.as_ref()));
            (key_bytes, val_bytes)
        }));

        histogram!("armour_logdb_tree_apply_batch_duration", "name" => self.raw().name.clone())
            .record(start.elapsed().as_secs_f64());
        counter!("armour_logdb_tree_apply_batch_total", "name" => self.raw().name.clone())
            .increment(1);

        res
    }

    pub fn next_id(&self) -> DbResult<u64> {
        self.raw().next_id()
    }
}

// read methods
impl<Item> LogTree<Item>
where
    Item: Record<Value = Slice> + Debug,
{
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    fn get_by_slice(&self, key: &[u8]) -> DbResult<Option<Item>> {
        let start = std::time::Instant::now();
        let val = self.tree().get(key)?;
        let res = match val {
            Some(val) => Ok(Some(Item::deser_owned(val))),
            None => Ok(None),
        };

        histogram!("armour_logdb_tree_get_by_slice_duration", "name" => self.raw().name.clone())
            .record(start.elapsed().as_secs_f64());
        counter!("armour_logdb_tree_get_by_slice_total", "name" => self.raw().name.clone())
            .increment(1);

        res
    }

    /// get value by id
    #[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())
    }

    /// get value by id or return not found
    #[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 = "exists")]
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME), ret)]
    pub fn contains(&self, id: impl Borrow<Item::SelfId> + Debug) -> DbResult<bool> {
        let start = std::time::Instant::now();
        let key = Cid::encode(id.borrow());
        let res = self
            .tree()
            .contains_key(key.as_ref())
            .map_err(DbError::from);

        histogram!("armour_logdb_tree_contains_duration", "name" => self.raw().name.clone())
            .record(start.elapsed().as_secs_f64());
        counter!("armour_logdb_tree_contains_total", "name" => self.raw().name.clone())
            .increment(1);

        res
    }

    /// get iterator for collection
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn iter(&self) -> impl DoubleEndedIterator<Item = Entry<Item>> {
        counter!("armour_logdb_tree_iter_total", "name" => self.raw().name.clone()).increment(1);
        self.tree()
            .iter()
            .filter_map(|item| entry_filter_map(item.into_inner()))
    }

    /// get iterator by prefix for collection
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn scan_prefix<'a, P>(&'a self, prefix: P) -> impl DoubleEndedIterator<Item = Entry<Item>>
    where
        P: AsRef<[u8]> + Debug + 'a,
    {
        counter!("armour_logdb_tree_scan_prefix_total", "name" => self.raw().name.clone())
            .increment(1);
        self.tree()
            .prefix(prefix)
            .filter_map(|item| entry_filter_map(item.into_inner()))
    }

    /// get iterator by prefix for collection
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn scan_prefix_key<'a, P>(
        &'a self,
        prefix: P,
    ) -> impl DoubleEndedIterator<Item = Item::SelfId> + 'a
    where
        P: AsRef<[u8]> + Debug + 'a,
    {
        counter!("armour_logdb_tree_scan_prefix_key_total", "name" => self.raw().name.clone())
            .increment(1);
        self.tree()
            .prefix(prefix)
            .filter_map(|entry| match entry.into_inner() {
                Ok((key, _)) => Some(Item::deser_key(&key)),
                Err(e) => {
                    error!(%e);
                    None
                }
            })
    }

    /// get iterator by range for collection
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn range_iter<'a, K, R>(&'a self, range: R) -> impl DoubleEndedIterator<Item = Entry<Item>>
    where
        K: AsRef<[u8]> + 'a,
        R: RangeBounds<K> + Debug + 'a,
    {
        counter!("armour_logdb_tree_range_iter_total", "name" => self.raw().name.clone())
            .increment(1);
        self.tree()
            .range(range)
            .filter_map(|item| entry_filter_map(item.into_inner()))
    }

    /// get iterator by prefix for collection
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn try_scan_prefix<'a, P>(
        &'a self,
        prefix: P,
    ) -> impl DoubleEndedIterator<Item = DbResult<Entry<Item>>>
    where
        P: AsRef<[u8]> + Debug + 'a,
    {
        counter!("armour_logdb_tree_try_scan_prefix_total", "name" => self.raw().name.clone())
            .increment(1);
        self.tree()
            .prefix(prefix)
            .map(|item| entry_map(item.into_inner()))
    }

    /// get iterator by prefix for collection
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn try_scan_prefix_with_raw<'a, P>(
        &'a self,
        prefix: P,
    ) -> impl DoubleEndedIterator<Item = DbResult<ItemWithRaw<Item>>>
    where
        P: AsRef<[u8]> + Debug + 'a,
    {
        counter!("armour_logdb_tree_try_scan_prefix_with_raw_total", "name" => self.raw().name.clone()).increment(1);
        self.tree()
            .prefix(prefix)
            .map(|item| entry_map_with_raw(item.into_inner()))
    }

    /// get iterator by range for collection
    /// if item error, return Error
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn try_range_iter<'a, K, R>(
        &'a self,
        range: R,
    ) -> impl DoubleEndedIterator<Item = DbResult<ItemWithRaw<Item>>>
    where
        K: AsRef<[u8]> + 'a,
        R: RangeBounds<K> + Debug + 'a,
    {
        counter!("armour_logdb_tree_try_range_iter_total", "name" => self.raw().name.clone())
            .increment(1);
        self.tree().range(range).map(|item| {
            item.into_inner().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
    #[instrument(level = "trace", skip(self), fields(name = Item::NAME))]
    pub fn iter_keys(&self) -> impl DoubleEndedIterator<Item = Item::SelfId> {
        counter!("armour_logdb_tree_iter_keys_total", "name" => self.raw().name.clone())
            .increment(1);
        self.tree().iter().filter_map(|item| match item.key() {
            Ok(key) => Some(Item::deser_key(&key)),
            Err(e) => {
                error!(%e);
                None
            }
        })
    }
}

fn entry_filter_map<Item: Record<Value = Slice>>(
    item: fjall::Result<KeyValue>,
) -> Option<Entry<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) => {
            error!(%e);
            None
        }
    }
}

fn entry_map<Item: Record<Value = Slice>>(item: fjall::Result<KeyValue>) -> DbResult<Entry<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 }
    })
}

fn entry_map_with_raw<Item: Record<Value = Slice>>(
    item: fjall::Result<KeyValue>,
) -> DbResult<ItemWithRaw<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 })
    })
}
