// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
// Modifications Copyright 2022 PingCAP, Inc.

use bytes::BufMut;
use engine_traits::CfName;
use kvengine::{EXTRA_CF, WRITE_CF};
use txn_types::Key;

use crate::storage::{
    kv::{CfStatistics, Snapshot},
    mvcc::Result,
};

#[maybe_async::both]
pub async fn has_data_in_range<S: Snapshot>(
    snapshot: S,
    _cf: CfName,
    left: &Key,
    right: &Key,
    _statistic: &mut CfStatistics,
) -> Result<bool> {
    if let Some(snap) = snapshot.get_kvengine_snap() {
        let raw_left = std::cmp::max(left.to_raw().unwrap(), snap.clone_start_key().to_vec());
        let mut raw_right = std::cmp::min(right.to_raw().unwrap(), snap.clone_end_key().to_vec());
        let mut iter = snap
            .new_iterator(WRITE_CF, false, false, Some(u64::MAX), true)
            .await;
        iter.seek(&raw_left).await;
        if iter.valid() && iter.key() < raw_right.as_slice() {
            return Ok(true);
        }
        let mut extra_iter = snap.new_iterator(EXTRA_CF, false, false, Some(u64::MAX), true);
        extra_iter.seek(&raw_left);
        raw_right.put_u64(u64::MAX);
        return Ok(extra_iter.valid() && extra_iter.key() < raw_right.as_slice());
    }
    unimplemented!()
}

#[cfg(test)]
mod tests {
    use engine_traits::{CF_WRITE as WRITE_CF_NAME, IterOptions, ReadOptions};
    use kvengine::{
        WriteBatch,
        test_engine::{ApplyTask, new_test_engine_api_v2},
    };
    use tikv_util::mpsc;

    use super::*;
    use crate::storage::kv::{DummySnapshotExt, Iterator, Result as KvResult};

    #[derive(Clone)]
    struct KvEngineSnapshot {
        snap: kvengine::SnapAccess,
    }

    struct NoopIterator;

    impl Iterator for NoopIterator {
        fn next(&mut self) -> KvResult<bool> {
            unimplemented!()
        }

        fn prev(&mut self) -> KvResult<bool> {
            unimplemented!()
        }

        fn seek(&mut self, _key: &Key) -> KvResult<bool> {
            unimplemented!()
        }

        fn seek_for_prev(&mut self, _key: &Key) -> KvResult<bool> {
            unimplemented!()
        }

        fn seek_to_first(&mut self) -> KvResult<bool> {
            unimplemented!()
        }

        fn seek_to_last(&mut self) -> KvResult<bool> {
            unimplemented!()
        }

        fn valid(&self) -> KvResult<bool> {
            unimplemented!()
        }

        fn key(&self) -> &[u8] {
            unimplemented!()
        }

        fn value(&self) -> &[u8] {
            unimplemented!()
        }
    }

    impl Snapshot for KvEngineSnapshot {
        type Iter = NoopIterator;
        type Ext<'a> = DummySnapshotExt;

        fn get(&self, _key: &Key) -> KvResult<Option<Vec<u8>>> {
            unimplemented!()
        }

        fn get_cf(&self, _cf: CfName, _key: &Key) -> KvResult<Option<Vec<u8>>> {
            unimplemented!()
        }

        fn get_cf_opt(
            &self,
            _opts: ReadOptions,
            _cf: CfName,
            _key: &Key,
        ) -> KvResult<Option<Vec<u8>>> {
            unimplemented!()
        }

        fn iter(&self, _cf: CfName, _iter_opt: IterOptions) -> KvResult<Self::Iter> {
            unimplemented!()
        }

        fn ext(&self) -> DummySnapshotExt {
            DummySnapshotExt
        }

        fn get_kvengine_snap(&self) -> Option<&kvengine::SnapAccess> {
            Some(&self.snap)
        }
    }

    fn apply_kvengine_write(applier_tx: &mpsc::Sender<ApplyTask>, key: &[u8]) {
        let mut wb = WriteBatch::default();
        wb.reset(1);
        wb.put(WRITE_CF, key, b"value", 0, &[], 1);
        let (result_tx, result_rx) = mpsc::bounded(1);
        let task = ApplyTask::new_wb(wb, result_tx);
        applier_tx.send(task).unwrap();
        result_rx.recv().unwrap().unwrap();
    }

    #[maybe_async::test]
    async fn test_has_data_in_range_respects_snapshot_end() {
        let (engine, applier_tx) = new_test_engine_api_v2();
        let key = engine.key_builder.i_to_outer_key(1);
        apply_kvengine_write(&applier_tx, &key);

        let shard = engine.get_shard(1).unwrap();
        let snap = kvengine::SnapAccess::new(&shard);
        let end_key = snap.get_end_key().to_vec();
        let snapshot = KvEngineSnapshot { snap };
        let left = Key::from_raw(&key);
        let right_inside = Key::from_raw(&end_key);
        let mut right_outside = end_key.clone();
        right_outside.push(0);
        let right_outside = Key::from_raw(&right_outside);

        let mut stats = CfStatistics::default();
        let has_inside = has_data_in_range(
            snapshot.clone(),
            WRITE_CF_NAME,
            &left,
            &right_inside,
            &mut stats,
        )
        .await
        .unwrap();
        assert!(has_inside);

        let has_outside =
            has_data_in_range(snapshot, WRITE_CF_NAME, &left, &right_outside, &mut stats)
                .await
                .unwrap();
        assert!(has_outside);
    }
}