#![doc(html_no_source)] // remove it upon open-source

use std::collections::HashMap;

use std::path::Path;

pub use rocksdb;

use rocksdb::{OptimisticTransactionDB, Options};

use lyquor_api::{

    anyhow::Context,

    kvstore::{KVStore, KVStoreError, Key, Value},

    parking_lot::RwLock,

};

pub struct MemDB(RwLock<HashMap<Bytes, Bytes>>);

impl MemDB {

    pub fn new() -> Self {

        Self(RwLock::new(HashMap::new()))

    }

}

impl Default for MemDB {

    fn default() -> Self {

        Self::new()

    }

}

impl KVStore for MemDB {

    fn atomic_write<'a>(

        &'a self, changes: Box<dyn Iterator<Item = (Key, Option<Value>)> + 'a>,

    ) -> Result<(), KVStoreError> {

        let mut map = self.0.write();

        for (k, v) in 
changes15.4k
 {

            match v {

                Some(v) => {

                    map.insert(k.consolidated(), v);

                }

                None => {

                    map.remove(&k.consolidated());

                }

            }

        }

        Ok(())

    }

    fn contains(&self, key: Key) -> bool {

        self.0.read().contains_key(&key.consolidated())

    }

    fn get(&self, key: Key) -> Result<Option<Value>, KVStoreError> {

        Ok(self.0.read().get(&key.consolidated()).cloned())

    }

}

pub struct RocksDB(rocksdb::OptimisticTransactionDB);

impl KVStore for RocksDB {

    fn atomic_write<'a>(

        &'a self, changes: Box<dyn Iterator<Item = (Key, Option<Value>)> + 'a>,

    ) -> Result<(), KVStoreError> {

        let mut wb = rocksdb::WriteBatchWithTransaction::default();

        for (k, v) in 
changes7
 {

            match v {

                Some(v) => wb.put(k.consolidated(), v),

                None => wb.delete(k.consolidated()),

            }

        }

        Ok(self.0.write(wb).context("RocksDB write error.")
?0
)

    }

    fn contains(&self, key: Key) -> bool {

        self.0.key_may_exist(key.consolidated())

    }

    fn get(&self, key: Key) -> Result<Option<Value>, KVStoreError> {

        Ok(self

            .0

            .get(key.consolidated())

            .context("RocksDB read error.")
?0

            .map(|v| 
v371
.
into371
()))

    }

}

impl RocksDB {

    pub fn new(path: &Path) -> Result<Self, KVStoreError> {

        let mut opts = Options::default();

        opts.create_if_missing(true);

        let db = OptimisticTransactionDB::open(&opts, path).context("RocksDB open error")
?0
;

        Ok(Self(db))

    }

}

use lyquor_api::subkey_builder;

use lyquor_primitives::{Bytes, decode_object, encode_object};

use serde::{Deserialize, Serialize};

use std::hash::Hash;

pub struct SortedMappingStore<

    S: KVStore,

    K: Hash + Eq + Ord + Serialize + for<'a> Deserialize<'a> + Clone,

    V: Serialize + for<'a> Deserialize<'a> + Clone,

> {

    cache: lru::LruCache<K, V>,

    max: Option<K>,

    first_idx: u64,

    next_idx: u64,

    store: S,

}

subkey_builder!(SortedMappingSubkey(

    ([0x00])-value(&[u8]) => Key,

    ([0x01])-index(&u64) => Key,

    ([0x02])-max() => Key,

    ([0x03])-first_idx() => Key,

    ([0x04])-next_idx() => Key,

));

impl<

    S: KVStore,

    K: Hash + Eq + Ord + Serialize + for<'a> Deserialize<'a> + Clone,

    V: Serialize + for<'a> Deserialize<'a> + Clone,

> SortedMappingStore<S, K, V>

{

    fn subkey() -> &'static SortedMappingSubkey {

        static KEY: std::sync::OnceLock<SortedMappingSubkey> = std::sync::OnceLock::new();

        KEY.get_or_init(|| 
SortedMappingSubkey::new5
(
Value::new5
().
into5
()))

    }

    pub fn new(store: S, cache_size: std::num::NonZeroUsize) -> Result<Self, KVStoreError> {

        let max: Option<K> = match store.get(Self::subkey().max())
?0
 {

            Some(raw) => decode_object(&raw),

            None => None,

        };

        let first_idx: u64 = store

            .get(Self::subkey().first_idx())
?0

            .and_then(|raw| 
decode_object0
(
&raw0
))

            .unwrap_or(0);

        let next_idx: u64 = store

            .get(Self::subkey().next_idx())
?0

            .and_then(|raw| 
decode_object0
(
&raw0
))

            .unwrap_or(0);

        Ok(Self {

            cache: lru::LruCache::new(cache_size),

            max,

            first_idx,

            next_idx,

            store,

        })

    }

    pub fn max_key(&self) -> Option<K> {

        self.max.clone()

    }

    pub fn append(&mut self, key: K, value: V) -> Result<bool, KVStoreError> {

        if !self.max_key().map(|max| 
key1
 > 
max1
).unwrap_or(true) {

            return Ok(false);

        }

        let idx = self.next_idx;

        self.next_idx += 1;

        self.cache.push(key.clone(), value.clone());

        self.max = Some(key.clone());

        let key_raw: Bytes = encode_object(&key).into();

        let mapping = [

            (Self::subkey().value(&key_raw), Some(encode_object(&value).into())),

            (Self::subkey().index(&idx), Some(key_raw.clone())),

            (Self::subkey().max(), Some(key_raw)),

            (Self::subkey().next_idx(), Some(encode_object(&self.next_idx).into())),

        ]

        .into_iter();

        self.store.atomic_write(Box::new(mapping))
?0
;

        Ok(true)

    }

    pub fn append_or_update_max(&mut self, key: K, value: V) -> Result<bool, KVStoreError> {

        match self.max_key() {

            Some(
max1
) if key > ma
x1
 => 
self1
.
append1
(
key1
, 
value1
),

            Some(max) if key == max => {

                self.cache.push(key.clone(), value.clone());

                let key_raw: Bytes = encode_object(&key).into();

                self.store.atomic_write(Box::new(

                    [(Self::subkey().value(&key_raw), Some(encode_object(&value).into()))].into_iter(),

                ))
?0
;

                Ok(true)

            }

            Some(_) => Ok(false),

            None => self.append(key, value),

        }

    }

    pub fn get_le(&mut self, key: K) -> Option<V> {

        let mut l = self.first_idx;

        let mut r = self.next_idx;

        let 
mut l_key8
 = self

            .store

            .get(Self::subkey().index(&l))

            .ok()

            .flatten()

            .and_then(|b| 
decode_object8
(
&b8
))
?6
;

        if key < l_key {

            return None;

        }

        while r > l + 1 {

            let mid = (l + r) >> 1;

            let mid_key: K = self

                .store

                .get(Self::subkey().index(&mid))

                .ok()

                .flatten()

                .and_then(|b| decode_object(&b))
?0
;

            if key < mid_key {

                r = mid

            } else {

                l = mid;

                l_key = mid_key;

            }

        }

        let value = self.get(l_key.clone());

        if let Some(value) = &value &&

            l_key != key

        {

            self.cache.push(key, value.clone());

        }

        value

    }

    pub fn get(&mut self, key: K) -> Option<V> {

        match self.cache.get(&key) {

            Some(value) => Some(value.clone()),

            None => {

                let value: V = self

                    .store

                    .get(Self::subkey().value(&encode_object(&key)))

                    .ok()

                    .flatten()

                    .and_then(|b| decode_object(&b))?;

                self.cache.push(key.clone(), value.clone());

                Some(value)

            }

        }

    }

}

#[cfg(test)]

mod tests {

    use super::SortedMappingStore;

    use crate::MemDB;

    #[test]

    fn sorted_mapping_store_updates_existing_max_key() {

        let mut store = SortedMappingStore::new(MemDB::new(), 4.try_into().unwrap()).unwrap();

        assert!(store.append(7_u64, 10_u64).unwrap());

        assert!(store.append_or_update_max(7, 11).unwrap());

        assert_eq!(store.max_key(), Some(7));

        assert_eq!(store.get(7), Some(11));

        assert_eq!(store.get_le(7), Some(11));

    }

}