pub use bytes::Bytes;
use emitter_and_signal::signal::{JoinError, Signal};
pub use fjall::{Config, Keyspace, Partition, PartitionCreateOptions};
use serde::{Deserialize, Serialize};
use snafu::{OptionExt, ResultExt, Snafu};
use std::{fmt::Debug, future::Future, num::NonZeroUsize, ops::Deref, sync::Arc};
use tokio::{
    select,
    sync::{mpsc, oneshot},
    task::spawn_blocking,
};

#[derive(Debug, Clone, Snafu)]
pub enum PersistedError {
    Missing,
    CreationError {
        // Wrapped in [`Arc`] to make this [`Clone`]
        source: Arc<fjall::Error>,
    },
    DeserializationError {
        source: postcard::Error,
    },
}

pub async fn persisted<T: Debug + Send + Sync + 'static + Serialize + for<'a> Deserialize<'a>>(
    partition: Partition,
    identifier: Bytes,
    buffer: NonZeroUsize,
) -> (
    Setter<T>,
    Signal<Result<T, PersistedError>>,
    impl Future<Output = Result<(), JoinError>>,
) {
    let initial = spawn_blocking({
        let partition = partition.clone();
        let identifier = identifier.clone();
        move || partition.get(identifier.deref())
    })
    .await
    .unwrap()
    .map_err(Arc::new)
    .context(CreationSnafu)
    .and_then(|op| op.context(MissingSnafu))
    .and_then(|slice| postcard::from_bytes(&slice).context(DeserializationSnafu));

    let (set_tx, mut set_rx) = mpsc::channel(buffer.get());
    let setter = Setter { sender: set_tx };

    let (signal, task) = Signal::new(initial, move |mut publisher_stream| async move {
        while let Some(publisher) = publisher_stream.wait().await {
            loop {
                select! {
                    biased;
                    _ = publisher.all_unsubscribed() => {
                        break;
                    }
                    new_value_opt = set_rx.recv() => {
                        let Some((new_value, callback)) = new_value_opt else { return };
                        let serialized_res = postcard::to_stdvec(&new_value).context(SerializationSnafu);
                        // Stand-in for Option::async_and_then
                        let persisted_res = match serialized_res {
                            Ok(serialized) => spawn_blocking({
                                let partition = partition.clone();
                                let identifier = identifier.clone();
                                move || partition.insert(identifier.deref(), serialized)
                            }).await.unwrap().context(SavingSnafu),
                            Err(error) => Err(error),
                        };

                        if persisted_res.is_ok() {
                            publisher.publish(Ok(new_value));
                        }

                        let _ = callback.send(persisted_res);
                    }
                }
            }
        }
    });

    (setter, signal, task)
}

#[derive(Debug, Clone)]
pub struct Setter<T: 'static> {
    sender: mpsc::Sender<(T, oneshot::Sender<Result<(), SetError<T>>>)>,
}

// TODO: add doc comments functioning as error Display
#[derive(Debug, Snafu)]
pub enum SetError<T: 'static> {
    Closed { source: mpsc::error::SendError<T> },
    NoFeedback { source: oneshot::error::RecvError },
    SerializationError { source: postcard::Error },
    SavingError { source: fjall::Error },
}

impl<T: Debug> Setter<T> {
    pub async fn set(&self, value: T) -> Result<(), SetError<T>> {
        let (callback_tx, callback_rx) = oneshot::channel();

        self.sender
            .send((value, callback_tx))
            .await
            .map_err(|send_error| mpsc::error::SendError(send_error.0 .0))
            .context(ClosedSnafu)?;

        let set_res = callback_rx.await.context(NoFeedbackSnafu)?;
        set_res
    }
}