use std::future::Future;
use tokio::sync::mpsc;

use crate::shard::InternalMessage;
use crate::{ServiceData, UpstreamError};

struct DataCommitGuard<Key, Data> {
    key: Option<Key>,
    sender: mpsc::UnboundedSender<InternalMessage<Key, Data>>,
}

pub struct DataCommitRequest<'a, Key, Data> {
    data: &'a mut Data,
    commit_guard: DataCommitGuard<Key, Data>,
}

impl<'a, Key, Data> DataCommitRequest<'a, Key, Data> {
    pub(crate) fn new(
        sender: mpsc::UnboundedSender<InternalMessage<Key, Data>>,
        key: Key,
        data: &'a mut Data,
    ) -> Self {
        Self {
            commit_guard: DataCommitGuard {
                key: Some(key),
                sender,
            },
            data,
        }
    }

    pub fn key(&self) -> &Key {
        self.commit_guard.key()
    }

    pub fn into_processing(self) -> ProcessingDataCommitRequest<Key, Data> {
        ProcessingDataCommitRequest {
            drop_guard: self.commit_guard,
        }
    }

    pub fn resolve(self) {
        self.into_processing().resolve();
    }

    pub fn reject<E: Into<UpstreamError>>(self, error: E) {
        self.into_processing().reject(error);
    }
}

impl<Key: Send + 'static, Data: ServiceData> DataCommitRequest<'_, Key, Data> {
    pub fn spawn<F: Future<Output = Result<(), UpstreamError>> + Send + 'static>(self, fut: F) {
        self.into_processing().spawn(fut);
    }
}

impl<Key, Data: Clone> DataCommitRequest<'_, Key, Data> {
    pub fn into_owned(self) -> OwnedDataCommitRequest<Key, Data> {
        OwnedDataCommitRequest {
            data: self.data.clone(),
            commit_guard: self.commit_guard,
        }
    }
}

pub struct OwnedDataCommitRequest<Key, Data> {
    data: Data,
    commit_guard: DataCommitGuard<Key, Data>,
}

impl<Key, Data> OwnedDataCommitRequest<Key, Data> {
    pub fn resolve(self) {
        self.commit_guard.resolve();
    }

    pub fn reject<E: Into<UpstreamError>>(self, error: E) {
        self.commit_guard.reject(error);
    }
}

impl<Key: Send + 'static, Data: ServiceData> OwnedDataCommitRequest<Key, Data> {
    pub fn spawn<
        R: Future<Output = Result<(), UpstreamError>> + Send + 'static,
        F: FnOnce(Data) -> R + Send + 'static,
    >(
        self,
        func: F,
    ) {
        tokio::spawn(async move {
            let drop_guard = self.commit_guard;
            let fut = (func)(self.data);
            match fut.await {
                Ok(()) => drop_guard.resolve(),
                Err(err) => drop_guard.reject(err),
            };
        });
    }
}

pub struct ProcessingDataCommitRequest<Key, Data> {
    drop_guard: DataCommitGuard<Key, Data>,
}

impl<Key, Data> ProcessingDataCommitRequest<Key, Data> {
    pub fn resolve(self) {
        self.drop_guard.resolve();
    }

    pub fn reject<E: Into<UpstreamError>>(self, error: E) {
        self.drop_guard.reject(error);
    }
}

impl<Key: Send + 'static, Data: ServiceData> ProcessingDataCommitRequest<Key, Data> {
    pub fn spawn<F: Future<Output = Result<(), UpstreamError>> + Send + 'static>(self, fut: F) {
        tokio::spawn(async move {
            match fut.await {
                Ok(()) => self.resolve(),
                Err(err) => self.reject(err),
            };
        });
    }
}

impl<Key, Data> DataCommitGuard<Key, Data> {
    fn key(&self) -> &Key {
        self.key.as_ref().unwrap()
    }

    fn resolve(mut self) {
        let key = self.key.take();
        // todo: send back internal message.
    }

    fn reject<E: Into<UpstreamError>>(self, error: E) {
        // todo: send back internal message if commit failed.
        panic!(format!("unhandled failure - failed {:?}", error.into()));
    }
}

impl<Key, Data> Drop for DataCommitGuard<Key, Data> {
    fn drop(&mut self) {
        match self.key.take() {
            None => {}
            Some(_key) => {
                todo!("send back internal message if commit guard is dropped without key being taken.");
            }
        }
    }
}
