use std::fmt::Display;
use std::hash::Hash;

use indexmap::map::Entry;
use indexmap::{IndexMap, IndexSet};
use log::{error, info, warn};
use rand::{thread_rng, Rng};
use smallvec::SmallVec;

use tokio::stream::StreamExt;
use tokio::sync::mpsc;
use tokio::time::{interval, Instant, Interval};

use super::commit_queue::{CommitQueue, EnqueuedCommit, TakeOwnershipResult};
use super::lru_candidates::{InsertionStatus, LruCandidates};
use super::{
    InnerCommitPolicy, InternalMessage, PendingCallback, ServiceHandleMessage, ShardConfig,
    ShardStats, TakeDataSender, TakenData,
};
use crate::{
    CommitToUpstream, DataCommitRequest, DataLoadRequest, LoadFromUpstream, ServiceData, ShardError,
};

pub(super) struct Shard<Key, Data, Upstream> {
    running: bool,
    last_message_was_expedited: bool,
    config: ShardConfig,

    shard_receiver: mpsc::Receiver<ServiceHandleMessage<Key, Data>>,

    internal_receiver: mpsc::UnboundedReceiver<InternalMessage<Key, Data>>,
    internal_sender: mpsc::UnboundedSender<InternalMessage<Key, Data>>,

    data: IndexMap<Key, DataState<Key, Data>>,
    expiring_keys: IndexSet<Key>,
    lru_candidates: LruCandidates<Key>,
    stats: ShardStats,
    upstream: Upstream,
    commit_queue: Option<CommitQueue<Key, Data>>,
    access_offset: u64,
    expedited_expiration_probe_enqueued: bool,
    cache_expiration_probe_interval: Interval,
}

enum DataState<Key, Data> {
    // a SmallVec is used here, so we do not need to heap-allocate in the majority use-case,
    // which is there is going to be a load, and only one pending callback ever for it.
    Loading(
        SmallVec<[PendingCallback<Data>; 1]>,
        Option<TakeDataSender<Key, Data>>,
    ),
    Loaded(Data, u64, bool),
}

pub(super) mod marker {
    use std::marker::PhantomData;

    pub struct NotMutable<Key, Data>(pub PhantomData<(Key, Data)>);
    impl<Key, Data> Default for NotMutable<Key, Data> {
        fn default() -> Self {
            Self(PhantomData)
        }
    }

    pub struct Immutable<Key, Data, Upstream>(Upstream, NotMutable<Key, Data>);

    impl<Key, Data, Upstream> Immutable<Key, Data, Upstream> {
        pub fn new(upstream: Upstream) -> Self {
            Self(upstream, NotMutable::default())
        }
    }
    pub struct Mutable<Key, Data, Upstream>(Upstream, PhantomData<(Key, Data)>);

    impl<Key, Data, Upstream> Mutable<Key, Data, Upstream> {
        pub fn new(upstream: Upstream) -> Self {
            Self(upstream, PhantomData)
        }
    }

    pub trait MaybeMutableUpstream<Key, Data> {
        type Immutable: super::LoadFromUpstream<Key, Data>;
        type Mutable: super::CommitToUpstream<Key, Data>;

        fn as_immutable(&mut self) -> &mut Self::Immutable;
        fn as_mutable(&mut self) -> Option<&mut Self::Mutable>;
    }

    impl<Key: Send + 'static, Data: super::ServiceData> super::CommitToUpstream<Key, Data>
        for NotMutable<Key, Data>
    {
        fn commit<'a>(&mut self, data_commit_request: super::DataCommitRequest<'a, Key, Data>) {
            panic!("invariant: tried to mutate, with an immutable upstream.");
        }
    }

    impl<
            Key: Send + 'static,
            Data: super::ServiceData,
            Upstream: super::LoadFromUpstream<Key, Data>,
        > MaybeMutableUpstream<Key, Data> for Immutable<Key, Data, Upstream>
    {
        type Immutable = Upstream;
        type Mutable = NotMutable<Key, Data>;

        fn as_immutable(&mut self) -> &mut Self::Immutable {
            &mut self.0
        }

        fn as_mutable(&mut self) -> Option<&mut Self::Mutable> {
            None
        }
    }

    impl<
            Key: Send + 'static,
            Data: super::ServiceData,
            Upstream: super::LoadFromUpstream<Key, Data> + super::CommitToUpstream<Key, Data>,
        > MaybeMutableUpstream<Key, Data> for Mutable<Key, Data, Upstream>
    {
        type Immutable = Upstream;
        type Mutable = Upstream;

        fn as_immutable(&mut self) -> &mut Self::Immutable {
            &mut self.0
        }

        fn as_mutable(&mut self) -> Option<&mut Self::Mutable> {
            Some(&mut self.0)
        }
    }
}

impl<
        Key: Clone + Hash + Eq + Display + Send + 'static,
        Data: ServiceData,
        Upstream: LoadFromUpstream<Key, Data>,
        MaybeMutableUpstream: marker::MaybeMutableUpstream<Key, Data, Immutable = Upstream>,
    > Shard<Key, Data, MaybeMutableUpstream>
{
    pub(super) fn new(
        shard_receiver: mpsc::Receiver<ServiceHandleMessage<Key, Data>>,
        mut upstream: MaybeMutableUpstream,
        config: ShardConfig,
    ) -> Self {
        // we use an unbounded channel here, as we want the upstream implementation to be "fire and forget" which is to say,
        // the data loaded callback should not need to be awaited (which is the case with the regular Sender),
        // and we don't necessarily need back-pressure here, as the demand for loaded data originates from the external
        // shard message.
        let (internal_sender, internal_receiver) = mpsc::unbounded_channel();
        let commit_queue = if upstream.as_mutable().is_some() {
            Some(CommitQueue::with_capacity(config.persist_queue_capacity))
        } else {
            None
        };

        let cache_expiration_probe_interval = interval(config.cache_expiration_probe_interval);
        let data = IndexMap::with_capacity(config.max_data_capacity);
        let expiring_keys = IndexSet::with_capacity(config.max_data_capacity);

        Self {
            config,
            shard_receiver,
            internal_receiver,
            internal_sender,
            upstream,
            commit_queue,
            cache_expiration_probe_interval,
            data,
            expiring_keys,
            stats: Default::default(),
            lru_candidates: Default::default(),
            access_offset: 0,
            running: true,
            last_message_was_expedited: false,
            expedited_expiration_probe_enqueued: false,
        }
    }

    #[inline]
    fn next_access_offset(&mut self) -> u64 {
        self.access_offset += 1;
        self.access_offset
    }

    pub(super) async fn run(mut self) {
        let shard_id = self.config.shard_id;
        info!("shard={} is starting", shard_id);

        // todo: abort on panic as well.
        if let Err(error) = self.shard_loop().await {
            eprintln!("Shard {} has failed with error: {:?}", shard_id, error);
            std::process::abort();
        }

        info!(
            "shard={} has stopped, flushing writes to storage.",
            shard_id
        );

        let persist_all_start = Instant::now();
        match self.persist_all().await {
            Ok(()) => {
                info!(
                    "shard={} has finished flushing, took {:?}.",
                    shard_id,
                    persist_all_start.elapsed()
                );
            }
            Err(err) => {
                error!(
                    "shard={} has failed to persist all with error: {:?}",
                    shard_id, err
                );
            }
        }
    }

    async fn shard_loop(&mut self) -> Result<(), failure::Error> {
        while self.running {
            // todo: some notes on stopping:
            // - we need to make sure that we stop handling any shard messages,
            //   but we need to continue to handle messages on the internal
            //   receiver, until all the send ends have been closed.
            //   (the tricky part here, however, is that we also hold an internal sender,
            //    which means it wont close, so we need to do ... something, about that?!)
            tokio::select! {
                Some(shard_message) = self.shard_receiver.next() => {
                    self.last_message_was_expedited = false;
                    self.handle_shard_message(shard_message);
                }
                Some(internal_message) = self.internal_receiver.next() => {
                    // todo: remove me when this is "fixed" upstream?? https://github.com/tokio-rs/tokio/pull/2160
                    // the tokio scheduler can be a bit "unfair", in that if we send ourselves a message, and then try
                    // to immediately read it out, we may starve the scheduler for other tasks. These types of messages
                    // are called "expedited" requests, and when we receive one, we should yield quickly, to allow other
                    // tasks to be scheduled, before progressing with our work. We should only do this if the last message
                    // we've processed was also an expedited request. This lets us be a bit more "fair" than the scheduler
                    // allows us to be.
                    if internal_message.is_expedited_request() {
                        if self.last_message_was_expedited {
                            tokio::task::yield_now().await;
                        } else {
                            self.last_message_was_expedited = true;
                        }
                    } else {
                        self.last_message_was_expedited = false;
                    }
                    self.handle_internal_message(internal_message)
                }
                _interval = self.cache_expiration_probe_interval.tick(), if !self.expedited_expiration_probe_enqueued => {
                    self.last_message_was_expedited = false;
                    self.probe_expired_entries();
                }
                Ok(data_ref) = self.commit_queue.as_mut().unwrap().next(), if self.commit_queue.is_some() => {
                    self.last_message_was_expedited = false;
                    self.handle_commit_data(data_ref);
                }
                else => break,
            }
        }

        Ok(())
    }

    #[inline]
    fn handle_commit_data(&mut self, enqueued_commit: EnqueuedCommit<Key, Data>) {
        let (key, mut data) = enqueued_commit.into_inner();
        let data = match &mut data {
            Some(data) => data,
            None => match self.data.get_mut(&key) {
                Some(DataState::Loaded(data, _, _)) => data,
                _ => panic!("invariant: referenced key from persist queue does not exist."),
            },
        };
        let data_commit_request = DataCommitRequest::new(self.internal_sender.clone(), key, data);
        self.upstream
            .as_mutable()
            .expect("invariant: persist queue polled with no mutable upstream.")
            .commit(data_commit_request);
    }

    #[inline]
    fn handle_internal_message(&mut self, message: InternalMessage<Key, Data>) {
        match message {
            InternalMessage::DataLoadResult(key, Ok(data)) => {
                let next_access_offset = self.next_access_offset();

                match self.data.get_mut(&key) {
                    // The demand for the data is gone? Wat.
                    None | Some(DataState::Loaded(_, _, _)) => {}
                    // We're in a sane state, and we can now run callbacks.
                    Some(entry @ DataState::Loading(_, _)) => {
                        // This is some borrow checker fuckery here. Basically, we need to swap our entry in the map with the loaded data,
                        // this gives us now a stable `&mut Data` somewhere in memory that we can now invoke our callbacks upon.
                        let mut data = std::mem::replace(
                            entry,
                            DataState::Loaded(data, next_access_offset, false),
                        );

                        // Now we can go ahead and match out our loaded/loading variants.
                        match (entry, &mut data) {
                            (
                                DataState::Loaded(data, _access_offset, _is_lru_candidate),
                                DataState::Loading(pending_callbacks, take_data_sender),
                            ) => {
                                let pending_executions_completed = pending_callbacks.len();
                                // todo: we need to handle the commit policy decision here too.
                                for pending_callback in pending_callbacks.drain(..) {
                                    pending_callback.resolve(data);
                                }

                                let expires = data.get_expires_at().is_some();
                                self.stats.loads_in_progress -= 1;
                                self.stats.loads_complete += 1;
                                self.stats.executions_pending -= pending_executions_completed;
                                self.stats.executions_complete +=
                                    pending_executions_completed as u64;

                                // We have finished processing the data load, and performed all pending callbacks,
                                // do we have a request to now take ownership of the data elsewhere? if so, let's
                                // go ahead and respect that request.
                                match take_data_sender.take() {
                                    Some(take_data_sender) => self.take_data(key, take_data_sender),

                                    None => {
                                        if expires {
                                            self.expiring_keys.insert(key);
                                        }

                                        // We only execute this in this code-branch, as take_data above, will always remove the item, meaning,
                                        // we'll never be at capacity.
                                        if self.is_at_capacity() {
                                            self.evict_lru();
                                        }
                                    }
                                }
                            }

                            // This is unreachable, but the compiler doesn't know that.
                            _ => unreachable!(),
                        }
                    }
                };
            }
            InternalMessage::DataLoadResult(key, Err(err)) => {
                match self.data.swap_remove(&key) {
                    // Data isn't loaded, or is already loaded?
                    None | Some(DataState::Loaded(_, _, _)) => {}
                    Some(DataState::Loading(pending_callbacks, take_data_sender)) => {
                        if err.is_not_found() {
                            self.stats.loads_not_found += 1;
                        } else {
                            self.stats.loads_failed += 1;
                        }

                        for pending_callback in pending_callbacks {
                            pending_callback.reject(err.clone().into());
                        }

                        // If the load failed, there's nothing to take.
                        match take_data_sender {
                            Some(take_data_sender) => {
                                take_data_sender.send(None).ok();
                            }
                            None => {}
                        }
                    }
                }
            }
            InternalMessage::DoExpeditedExpirationProbe => {
                self.probe_expired_entries();
            }
        }
    }

    #[inline]
    fn handle_shard_message(&mut self, message: ServiceHandleMessage<Key, Data>) {
        match message {
            ServiceHandleMessage::Execute(key, func) => {
                self.execute_or_enqueue_load(key, PendingCallback::Execute(func));
            }
            ServiceHandleMessage::ExecuteMut(key, func) => {
                self.execute_or_enqueue_load(key, PendingCallback::ExecuteMut(func));
            }
            ServiceHandleMessage::ExecuteIfCached(key, func) => {
                self.execute_if_cached(key, func);
            }
            ServiceHandleMessage::GetStats(result_tx) => {
                self.stats.data_size = self.data.len();
                self.stats.expiring_keys = self.expiring_keys.len();
                result_tx.send(self.stats.clone()).ok();
            }
            ServiceHandleMessage::TakeData(key, result_tx) => {
                self.take_data(key, result_tx);
            }
            ServiceHandleMessage::Stop => {
                self.running = false;
                warn!("shard {} is stopping", self.config.shard_id);
            }
        }
    }

    #[inline]
    fn is_at_capacity(&self) -> bool {
        self.data.len() == self.config.max_data_capacity
    }

    #[inline]
    fn execute_if_cached(&mut self, key: Key, func: Box<dyn FnOnce(Option<&Data>) + Send>) {
        let next_access_offset = self.next_access_offset();

        match self.data.get_mut(&key) {
            Some(DataState::Loaded(data, access_offset, is_lru_candidate)) => {
                if *is_lru_candidate {
                    *is_lru_candidate = false;
                    self.lru_candidates
                        .remove_candidate_by_access_offset(*access_offset);
                }
                *access_offset = next_access_offset;
                (func)(Some(data));
            }
            _ => {
                (func)(None);
            }
        }
    }

    #[inline]
    fn execute_or_enqueue_load(&mut self, key: Key, pending_callback: PendingCallback<Data>) {
        // If we are at capacity, and we are unable to evict, and this isn't a known key,
        // we will reject the request. Future executions (once loads are complete),
        // should allow us to evict items.
        if self.is_at_capacity() && !self.evict_lru() && !self.data.contains_key(&key) {
            pending_callback.reject(ShardError::ShardAtCapacity.into());
            return;
        }

        let next_access_offset = self.next_access_offset();

        // todo: kill this clone
        let key_c = key.clone();

        // Figure out what the data state for the given key is.
        let (data_index, data_state, key_to_load) = match self.data.entry(key) {
            Entry::Vacant(ent) => {
                // If the data was evicted, but scheduled to be committed, we want to grab it back,
                // into the data hash-map. So check with the persist queue of whether or not we can
                // ask it to relinquish ownership of the given key.
                let take_ownership_result = self
                    .commit_queue
                    .as_mut()
                    .map(|queue| queue.take_ownership(ent.key()))
                    .unwrap_or(TakeOwnershipResult::NotEnqueued);

                match take_ownership_result {
                    TakeOwnershipResult::NotEnqueued => {
                        // The persist queue did not own the, we need to schedule it to be fetched from upstream.

                        // This is a bit of a clever hack, essentially, we've had to give up ownership of `key` when
                        // we called `data.entry()`, however, since we're going from vacant -> filled, we know we'll
                        // need to load the key, so we clone it out, and return it. It will be `None` otherwise. This
                        // avoids the clone in the cache-hit case, which for more complicated keys (non copy types)
                        // can be a bit expensive.
                        let key = ent.key().clone();
                        (
                            ent.index(),
                            ent.insert(DataState::Loading(SmallVec::new(), None)),
                            Some(key),
                        )
                    }
                    TakeOwnershipResult::Transferred(data) => {
                        // We have successfully taken the ownership of the data, so insert it into the
                        // data map.
                        (
                            ent.index(),
                            ent.insert(DataState::Loaded(data, next_access_offset, false)),
                            None,
                        )
                    }
                    TakeOwnershipResult::NotOwned => unreachable!(
                        "invariant: tried to relinquish ownership of key that was already in the hash-map"
                    ),
                }
            }
            Entry::Occupied(ent) => (ent.index(), ent.into_mut(), None),
        };

        match data_state {
            DataState::Loading(pending_callbacks, _take_data_sender) => {
                pending_callbacks.push(pending_callback);
                self.stats.executions_pending += 1;
            }
            DataState::Loaded(data, access_offset, is_lru_candidate) => {
                let prev_expires = data.get_expires_at().is_some();
                let inner_commit_policy = pending_callback.resolve(data);
                match inner_commit_policy {
                    InnerCommitPolicy::Noop => {}
                    // todo: properly handle.
                    _ => {
                        let data_commit_request =
                            DataCommitRequest::new(self.internal_sender.clone(), key_c, data);
                        self.upstream
                            .as_mutable()
                            .expect("invariant: data commit attempted with no mutable upstream.")
                            .commit(data_commit_request);
                    }
                };
                let next_expires = data.get_expires_at().is_some();
                self.stats.executions_complete += 1;

                // If we were an LRU candidate, we've just been accessed, which means, we're probably
                // not a good candidate anymore, and we should bail out.
                if *is_lru_candidate {
                    *is_lru_candidate = false;
                    self.lru_candidates
                        .remove_candidate_by_access_offset(*access_offset);
                }

                // Update the access offset, to say we've been recently accessed, so we don't become a
                // candidate for eviction any time soon.
                *access_offset = next_access_offset;

                match (prev_expires, next_expires) {
                    // was expiring, but no longer is.
                    (true, false) => {
                        let key = self
                            .data
                            .get_index(data_index)
                            .expect("invariant: data at previously discovered index does not exist")
                            .0;
                        self.expiring_keys.swap_remove(key);
                    }
                    // wasn't expiring, but is now expiring.
                    (false, true) => {
                        let key = self
                            .data
                            .get_index(data_index)
                            .expect("invariant: data at previously discovered index does not exist")
                            .0;
                        self.expiring_keys.insert(key.clone());
                    }
                    (false, false) | (true, true) => {}
                };
            }
        };

        if let Some(key) = key_to_load {
            let result_channel = DataLoadRequest::new(key, self.internal_sender.clone());
            self.upstream.as_immutable().load(result_channel);
            self.stats.loads_in_progress += 1;
        }
    }

    fn take_data(&mut self, key: Key, result_tx: TakeDataSender<Key, Data>) {
        let mut occupied_entry = match self.data.entry(key) {
            // The data does not exist in the main hash-map, check if it exists in the persist queue.
            Entry::Vacant(ent) => {
                // The data exists in the persist queue, we'll yank it from there, this can happen
                // if the data has been evicted from the main hash-map and is pending a commit.
                let taken_data =
                    self.commit_queue
                        .as_mut()
                        .and_then(|queue| match queue.cancel(ent.key()) {
                            Some(EnqueuedCommit::Owned(key, data, was_enqueued_at)) => {
                                Some(TakenData {
                                    key,
                                    data,
                                    was_enqueued_at: Some(was_enqueued_at),
                                })
                            }
                            // This should never happen, as we've already checked the main hash-map
                            // above.
                            Some(EnqueuedCommit::References(_, _)) => unreachable!(),
                            None => None,
                        });

                if taken_data.is_some() {
                    self.stats.keys_taken += 1;
                }

                result_tx.send(taken_data).ok();
                return;
            }
            Entry::Occupied(ent) => ent,
        };

        match occupied_entry.get_mut() {
            DataState::Loading(_pending_callbacks, take_data_sender) => {
                match take_data_sender {
                    // There is no enqueued take data sender, so we'll go and re-insert, with the result_tx, so when the data is loaded, we'll
                    // have something to yank.
                    Some(_) => {
                        result_tx.send(None).ok();
                    }
                    // We already have an existing sender, so return `None` here.
                    // todo: maybe an AlreadyTakenError?? none is fine for now tho.
                    None => {
                        *take_data_sender = Some(result_tx);
                    }
                };
            }
            DataState::Loaded(_, _, _) => {
                // Take ownership of the entry now that we know we're in the loaded state.
                let (key, data, access_offset, is_lru_candidate) =
                    match occupied_entry.swap_remove_entry() {
                        (key, DataState::Loaded(data, access_offset, is_lru_candidate)) => {
                            (key, data, access_offset, is_lru_candidate)
                        }
                        // unreachable because we checked the data state's variant above.
                        _ => unreachable!(),
                    };

                // todo: this book-keeping exists in multiple places, but is becoming increasingly error prone,
                // think up a way to abstract this state management.
                if is_lru_candidate {
                    self.lru_candidates
                        .remove_candidate_by_access_offset(access_offset);
                }

                if data.get_expires_at().is_some() {
                    self.expiring_keys.remove(&key);
                }

                // big todo: if we are persisting data due to an immediate write commit policy,
                // we need to ensure that take doesn't move the data before it's persisted.
                let was_enqueued_at = self
                    .commit_queue
                    .as_mut()
                    .and_then(|queue| queue.cancel(&key).map(|data| data.at()));

                let taken_data = TakenData {
                    key,
                    data,
                    was_enqueued_at,
                };
                self.stats.keys_taken += 1;
                result_tx.send(Some(taken_data)).ok();
            }
        }
    }

    /// Updates the best LRU candidates, by looking at a few random keys, and updating `lru_candidates`
    /// with the least recently used items in the bunch.
    fn update_lru_candidates(&mut self) {
        if self.data.is_empty() {
            return;
        }

        let mut rng = thread_rng();
        for _ in 0..self.config.lru_candidates_num_probes {
            let index = rng.gen_range(0, self.data.len());
            match &mut self.data.get_index_mut(index) {
                // If the data we've probed is not an lru-candidate, try to see if it's fit to be a candidate.
                Some((key, DataState::Loaded(_, access_offset, is_lru_candidate)))
                    if !*is_lru_candidate =>
                {
                    match self
                        .lru_candidates
                        .try_insert_candidate(&key, *access_offset)
                    {
                        // We have become a candidate.
                        InsertionStatus::Inserted(replaced_key) => {
                            *is_lru_candidate = true;
                            // Remove the LRU candidate status from the key that was bumped off.
                            if let Some(replaced_key) = replaced_key {
                                match self.data.get_mut(&replaced_key) {
                                    Some(DataState::Loaded(
                                        _data,
                                        _access_offset,
                                        is_lru_candidate,
                                    )) => {
                                        *is_lru_candidate = false;
                                    }
                                    _ => panic!("invariant: lru probing replaced a key that doesn't exist in the main data-map in the loaded state."),
                                }
                            }
                        }
                        InsertionStatus::Ineligible => {}
                    }
                }
                Some(_) => {}
                None => break,
            };
        }
    }

    /// Evicts what is "probably" the least recently used data from storage.
    ///
    /// Returns true if data was able to be successfully evicted, otherwise, false.
    fn evict_lru(&mut self) -> bool {
        self.update_lru_candidates();

        // Tries to grab the best candidate for eviction. This can be none, even after
        // update_lru_candidates has run, if for some weird reason no candidates
        // could be located, because most of the data is in the loading state,
        // or the data is simply empty.
        match self.lru_candidates.take_best_candidate() {
            Some(key_to_evict) => {
                self.evict_key(key_to_evict, EvictionReason::Lru);
                true
            }
            None => false,
        }
    }

    fn evict_key(&mut self, key_to_evict: Key, eviction_reason: EvictionReason) {
        let data = self
            .data
            .remove(&key_to_evict)
            .expect("evict_key tried to evict a key that does not exist.");

        match data {
            DataState::Loaded(data, access_offset, is_lru_candidate) => {
                match eviction_reason {
                    // If we are being evicted because of the expiration ttl-ing out,
                    // we'll remove ourselves from LRU candidates, because the
                    // ttl prober does not remove us from lru, but does remove us from
                    // the expiring key-set.
                    EvictionReason::Ttl => {
                        self.stats.keys_ttl_evicted += 1;
                        if is_lru_candidate {
                            self.lru_candidates
                                .remove_candidate_by_access_offset(access_offset);
                        }
                    }
                    // If we're being evicted because of LRU, we need to remove ourselves
                    // from the expiring keys set, because we don't remove ourselves from
                    // the expiring set in the lru prober.
                    EvictionReason::Lru => {
                        self.stats.keys_lru_evicted += 1;
                        if data.get_expires_at().is_some() {
                            self.expiring_keys.remove(&key_to_evict);
                        }
                    }
                }

                // If we have data that we have removed from the main data map,
                // try and give it to the persist queue. If the data is not
                // pending commit, then we'll drop it here, otherwise, it will live
                // on until the data is committed.
                if let Some(persist_queue) = &mut self.commit_queue {
                    persist_queue.give_ownership(key_to_evict, data);
                }
            }
            DataState::Loading(_, _) => {
                unreachable!("evict_key tried to evict data that was loading.")
            }
        }
    }

    /// Probes the expiring keys to see if any have expired. We do this rather than maintain a rather (costly) delay queue of all the
    /// expiring keys, applying a probabilistic approach to key expiry.
    ///
    /// Every time this probe is called, we look at a random sample of N many keys within the expiring keys set and expire them. If the
    /// number of keys expired during the probe (X) exceed 25% of N, we immediately schedule another probe, until the % of expired keys
    /// is under 25% per probe, then we go back to probing at the regular tick interval.
    fn probe_expired_entries(&mut self) {
        let mut rng = thread_rng();
        let mut num_expired_keys = 0;
        let now = Instant::now();

        for _ in 0..self.config.cache_expiration_probe_keys_per_tick {
            // There are no keys to evict, so we have nothing left to do.
            if self.expiring_keys.is_empty() {
                break;
            }

            let idx = rng.gen_range(0, self.expiring_keys.len());
            let maybe_expiring_key = self
                .expiring_keys
                .get_index(idx)
                .expect("invariant: Tried to get_index on an invalid range.");

            let data = self.data.get(maybe_expiring_key).expect(
                "invariant: tried to get key that existed in `expiring_keys` but not `data`",
            );
            let data = match data {
                // data is loading, and thus can't expire...
                DataState::Loading(_, _) => continue,
                DataState::Loaded(data, _, _) => data,
            };
            let expiry = match data.get_expires_at() {
                // data doesn't have an expiry?
                // todo: should this be a panic, or a warn?
                None => {
                    warn!("warn: a key was in `expiring_keys` that does not have an expiration.");
                    continue;
                }
                Some(expiry) => expiry,
            };

            if expiry > &now {
                continue;
            }

            let expiring_key = self.expiring_keys.swap_remove_index(idx).expect("invariant: tried to `swap_remove_index` on an index that no longer exists in `expiring_keys`");
            self.evict_key(expiring_key, EvictionReason::Ttl);
            num_expired_keys += 1;
        }

        // Did we expire enough keys to make us enter the expedited loop?
        // The thought here is that if we've expired over 25% of the keys when we probed, we should expedite the next tick, hopefully,
        // to the next tick of the event loop for this shard, going until the number of expired keys is probably under 25%. Then we'll
        // go back to probing at the configured probe interval. We do this instead of looping, so we can avoid "blocking" the event
        // loop processing expiration probes for too long.
        let expedited_expiration_threshold = self.config.cache_expiration_probe_keys_per_tick / 4;
        self.expedited_expiration_probe_enqueued =
            num_expired_keys >= expedited_expiration_threshold;

        if self.expedited_expiration_probe_enqueued {
            self.internal_sender
                .send(InternalMessage::DoExpeditedExpirationProbe)
                .ok();
        }
    }

    async fn persist_all(self) -> Result<(), ShardError> {
        let (upstream, persist_queue) = match self.commit_queue {
            None => return Ok(()),
            Some(persist_queue) => (self.upstream, persist_queue),
        };
        let cancelled_commits = persist_queue.consume();

        todo!();
    }
}

enum EvictionReason {
    Lru,
    Ttl,
}
