// Copyright 2026 PingCAP, Inc.

use std::{marker::PhantomData, sync::Arc};

use super::{
    arena::Arena,
    crossbeam_skl,
    skl::{self as arena_skl, Hint, WriteBatch},
};
use crate::{
    SnapAccess,
    table::{InnerKey, table::Value},
};

/// Selects the in-memory ordered-map implementation for a memtable.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub(crate) enum MemTableImplementation {
    /// The original arena-backed skiplist. It cannot store values of 16 MiB or
    /// larger.
    Arena,
    /// The Crossbeam skiplist, which has no 16 MiB value-size limit.
    #[default]
    Crossbeam,
}

impl MemTableImplementation {
    pub(crate) fn from_crossbeam_enabled(enabled: bool) -> Self {
        if enabled {
            Self::Crossbeam
        } else {
            Self::Arena
        }
    }
}

#[derive(Clone)]
/// Backend-neutral memtable skiplist facade.
pub struct SkipList {
    inner: SkipListInner,
}

// Arena returns a raw view tied to its skiplist, while Crossbeam owns an
// Arc-backed guard. PhantomData preserves the stricter Arena lifetime for
// callers of either backend.
pub(crate) struct ValueGuard<'a> {
    inner: ValueGuardInner,
    _list: PhantomData<&'a SkipListInner>,
}

enum ValueGuardInner {
    Arena(Value),
    Crossbeam(crossbeam_skl::ValueGuard),
}

impl ValueGuard<'_> {
    fn value(&self) -> &Value {
        match &self.inner {
            ValueGuardInner::Arena(value) => value,
            ValueGuardInner::Crossbeam(value) => value.value(),
        }
    }

    #[cfg(test)]
    pub fn is_empty(&self) -> bool {
        self.value().is_empty()
    }

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

    pub fn version(&self) -> u64 {
        self.value().version
    }

    pub(crate) fn encode_to_outer_owner(&self, outer_owner: &mut Vec<u8>) -> Value {
        crate::table::encode_val_to_outer_val_owner(*self.value(), outer_owner)
    }
}

#[derive(Clone)]
enum SkipListInner {
    Arena(arena_skl::SkipList),
    Crossbeam(crossbeam_skl::SkipList),
}

impl Default for SkipList {
    fn default() -> Self {
        Self::with_implementation(MemTableImplementation::default(), None)
    }
}

impl SkipList {
    /// Creates an arena-backed skiplist.
    pub fn new(arena: Option<Arc<Arena>>) -> Self {
        Self::with_implementation(MemTableImplementation::Arena, arena)
    }

    pub(super) fn with_implementation(
        implementation: MemTableImplementation,
        arena: Option<Arc<Arena>>,
    ) -> Self {
        let inner = match implementation {
            MemTableImplementation::Arena => SkipListInner::Arena(arena_skl::SkipList::new(arena)),
            MemTableImplementation::Crossbeam => {
                debug_assert!(arena.is_none());
                SkipListInner::Crossbeam(crossbeam_skl::SkipList::new())
            }
        };
        Self { inner }
    }

    pub fn new_iterator(&self, reversed: bool) -> Box<dyn crate::table::Iterator> {
        match &self.inner {
            SkipListInner::Arena(list) => Box::new(list.new_iterator(reversed)),
            SkipListInner::Crossbeam(list) => Box::new(list.new_iterator(reversed)),
        }
    }

    /// Creates a forward iterator optimized for flushing a sealed memtable.
    ///
    /// Arena already borrows from immutable arena storage, while Crossbeam uses
    /// its dedicated refcount-free version-chain iterator.
    pub fn new_flush_iterator(&self) -> Box<dyn crate::table::Iterator + '_> {
        match &self.inner {
            SkipListInner::Arena(list) => Box::new(list.new_iterator(false)),
            SkipListInner::Crossbeam(list) => Box::new(list.new_flush_iterator()),
        }
    }

    /// Permanently marks the backing map immutable. Arena needs no transition.
    pub(crate) fn seal(&self) {
        if let SkipListInner::Crossbeam(list) = &self.inner {
            list.seal();
        }
    }

    pub fn has_over_bound_data(&self, start: InnerKey<'_>, end: InnerKey<'_>) -> bool {
        match &self.inner {
            SkipListInner::Arena(list) => list.has_over_bound_data(start, end),
            SkipListInner::Crossbeam(list) => list.has_over_bound_data(start, end),
        }
    }

    pub fn put_batch(&self, batch: &mut WriteBatch, snap: Option<&SnapAccess>, cf: usize) {
        match &self.inner {
            SkipListInner::Arena(list) => list.put_batch(batch, snap, cf),
            SkipListInner::Crossbeam(list) => list.put_batch(batch, snap, cf),
        }
    }

    /// Inserts a remotely reconstructed batch without eliding tombstones.
    pub fn put_batch_preserve_tombstones(&self, batch: &mut WriteBatch, cf: usize) {
        match &self.inner {
            SkipListInner::Arena(list) => list.put_batch_preserve_tombstones(batch, cf),
            SkipListInner::Crossbeam(list) => list.put_batch_preserve_tombstones(batch, cf),
        }
    }

    pub(crate) fn get(&self, key: &[u8], version: u64) -> ValueGuard<'_> {
        let inner = match &self.inner {
            SkipListInner::Arena(list) => ValueGuardInner::Arena(list.get(key, version)),
            SkipListInner::Crossbeam(list) => ValueGuardInner::Crossbeam(list.get(key, version)),
        };
        ValueGuard {
            inner,
            _list: PhantomData,
        }
    }

    pub(crate) fn get_with_hint(
        &self,
        key: &[u8],
        version: u64,
        hint: &mut Hint,
    ) -> ValueGuard<'_> {
        let inner = match &self.inner {
            SkipListInner::Arena(list) => {
                ValueGuardInner::Arena(list.get_with_hint(key, version, hint))
            }
            SkipListInner::Crossbeam(list) => ValueGuardInner::Crossbeam(list.get(key, version)),
        };
        ValueGuard {
            inner,
            _list: PhantomData,
        }
    }

    pub fn supports_get_hint(&self) -> bool {
        matches!(self.inner, SkipListInner::Arena(_))
    }

    pub(crate) fn get_newer(&self, key: &[u8], version: u64) -> ValueGuard<'_> {
        let inner = match &self.inner {
            SkipListInner::Arena(list) => ValueGuardInner::Arena(list.get_newer(key, version)),
            SkipListInner::Crossbeam(list) => {
                ValueGuardInner::Crossbeam(list.get_newer(key, version))
            }
        };
        ValueGuard {
            inner,
            _list: PhantomData,
        }
    }

    pub fn is_empty(&self) -> bool {
        match &self.inner {
            SkipListInner::Arena(list) => list.is_empty(),
            SkipListInner::Crossbeam(list) => list.is_empty(),
        }
    }

    /// Returns deterministic cumulative submitted bytes used for switching.
    pub fn size(&self) -> u64 {
        match &self.inner {
            SkipListInner::Arena(list) => list.size(),
            SkipListInner::Crossbeam(list) => list.size(),
        }
    }

    pub fn entries(&self) -> usize {
        match &self.inner {
            SkipListInner::Arena(list) => list.entries(),
            SkipListInner::Crossbeam(list) => list.entries(),
        }
    }

    /// Returns the retained bytes charged to memory limiters for this skiplist.
    ///
    /// Arena memory cannot be reclaimed before the whole skiplist is dropped,
    /// so its submitted size remains the corresponding memory charge.
    pub fn memory_usage(&self) -> u64 {
        match &self.inner {
            SkipListInner::Arena(list) => list.size(),
            SkipListInner::Crossbeam(list) => list.memory_usage(),
        }
    }

    pub(crate) fn data_max_ts(&self) -> u64 {
        match &self.inner {
            SkipListInner::Arena(list) => list.data_max_ts(),
            SkipListInner::Crossbeam(list) => list.data_max_ts(),
        }
    }
}