use core::pin::Pin;
use std::{collections::HashMap, sync::Arc};

use futures::{FutureExt, Stream, StreamExt};
use taganak_core::{
    datasets::{Dataset, DatasetView},
    graphs::{either_graph::EitherGraph2, Graph, GraphError, GraphView},
    impl_default_dataset_view,
    quads::Quad,
};
use taganak_orm::re::{DatasetError, Term};

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SimpleDataset<G> {
    default_graph: Option<G>,
    graphs: HashMap<Arc<Term>, G>,
}

impl<G> Default for SimpleDataset<G>
where
    G: Graph + Sync,
{
    fn default() -> Self {
        Self {
            default_graph: None,
            graphs: HashMap::new(),
        }
    }
}

impl<'a, G> DatasetView for &'a SimpleDataset<G>
where
    G: Graph + Sync,
{
    type Stream = SimpleDatasetStream<'a, G>;

    async fn stream(self) -> Result<Self::Stream, DatasetError> {
        Ok(SimpleDatasetStream {
            dataset: self,
            state: SimpleDatasetStreamState::Start,
        })
    }

    impl_default_dataset_view!();
}

impl<G> SimpleDataset<G>
where
    G: Graph + Clone + Sync,
{
    pub fn new(default_graph: Option<G>) -> Self {
        Self {
            default_graph,
            ..Default::default()
        }
    }

    pub fn from_graphs<I>(default_graph: Option<G>, named_graphs: I) -> Self
    where
        I: IntoIterator<Item = (Arc<Term>, G)>,
    {
        Self {
            default_graph,
            graphs: named_graphs.into_iter().collect(),
        }
    }
}

impl<G> Dataset for SimpleDataset<G>
where
    G: Graph + Sync,
{
    type View<'a>
        = &'a Self
    where
        Self: 'a;

    type Graph = G;

    type DefaultGraph = Self;

    type DefaultView<'a>
        = <Self as Graph>::View<'a>
    where
        Self: 'a;

    fn view(&self) -> impl core::future::Future<Output = <Self as Dataset>::View<'_>> + Send {
        core::future::ready(self)
    }

    async fn default_view(&self) -> Self::DefaultView<'_> {
        <Self as Graph>::view(self).await
    }

    async fn named_view<'a>(&'a self, name: &Term) -> Option<<Self::Graph as Graph>::View<'a>> {
        Some(self.graphs.get(name)?.view().await)
    }

    async fn default_graph(&self) -> &Self::DefaultGraph {
        self
    }

    async fn default_graph_mut(&mut self) -> Option<&mut Self::DefaultGraph> {
        match self.default_graph {
            Some(_) => Some(self),
            None => None,
        }
    }

    async fn named_graph(&self, name: &Term) -> Option<&Self::Graph> {
        self.graphs.get(name)
    }

    async fn named_graph_mut(&mut self, name: &Term) -> Option<&mut Self::Graph> {
        self.graphs.get_mut(name)
    }

    async fn add_named_graph(
        &mut self,
        name: Arc<Term>,
        graph: Self::Graph,
    ) -> Result<(), taganak_core::prelude::DatasetError> {
        if self.graphs.contains_key(&name) {
            return Err(taganak_orm::re::DatasetError::ConflictingNamedGraph);
        }
        self.graphs.insert(name, graph);
        Ok(())
    }

    async fn remove_named_graph(
        &mut self,
        name: &Term,
    ) -> Result<Option<Self::Graph>, taganak_core::prelude::DatasetError> {
        Ok(self.graphs.remove(name))
    }
}

impl<G> Graph for SimpleDataset<G>
where
    G: Graph + Sync,
{
    type View<'g>
        = EitherGraph2<<&'g Self as DatasetView>::AssociatedGraphView, G::View<'g>>
    where
        Self: 'g;

    async fn view(&self) -> <Self as Graph>::View<'_> {
        match &self.default_graph {
            Some(graph) => EitherGraph2::N2(graph.view().await),
            None => EitherGraph2::N1(<Self as Dataset>::view(self).await.to_graph_view().await),
        }
    }

    async fn add(&mut self, triple: Arc<taganak_core::prelude::Triple>) -> Result<(), GraphError> {
        match &mut self.default_graph {
            Some(g) => g.add(triple).await,
            None => Err(GraphError::UnsupportedOp),
        }
    }

    async fn remove(
        &mut self,
        triple: taganak_core::prelude::TripleRef<'_>,
    ) -> Result<(), GraphError> {
        match &mut self.default_graph {
            Some(g) => g.remove(triple).await,
            None => Err(GraphError::UnsupportedOp),
        }
    }

    async fn clear(&mut self) -> Result<(), GraphError> {
        match &mut self.default_graph {
            Some(g) => g.clear().await,
            None => Ok(self.graphs.clear()),
        }
    }

    async fn merge<G1>(&mut self, source_graph: G1) -> Result<(), GraphError>
    where
        G1: GraphView,
    {
        match &mut self.default_graph {
            Some(g) => g.merge(source_graph).await,
            None => Err(GraphError::UnsupportedOp),
        }
    }

    async fn merge_stream<S, E>(
        &mut self,
        mut stream: S,
    ) -> Result<(), GraphError>
    where
        S: Stream<Item = Result<Arc<taganak_core::prelude::Triple>, E>> + Unpin + Send,
        GraphError: From<E>,
        E: Send,
    {
        match &mut self.default_graph {
            Some(g) => g.merge_stream(stream).await,
            None => Err(GraphError::UnsupportedOp),
        }
    }
}

pub struct SimpleDatasetStream<'a, G>
where
    G: Graph + 'a,
{
    dataset: &'a SimpleDataset<G>,
    state: SimpleDatasetStreamState<'a, G>,
}

pub enum SimpleDatasetStreamState<'a, G>
where
    G: Graph + 'a,
{
    Start,
    OpeningDefaultGraph(OpeningGraphState<'a, G>),
    ReadingDefaultGraph(<G::View<'a> as GraphView>::Stream),
    StartReadNamedGraphs,
    ReadingNamedGraphs {
        iter: std::collections::hash_map::Iter<'a, Arc<Term>, G>,
        state: NamedGraphReadingState<'a, G>,
    },
    End,
}

pub enum NamedGraphReadingState<'a, G>
where
    G: Graph + 'a,
{
    Start,
    OpeningGraph(&'a Arc<Term>, OpeningGraphState<'a, G>),
    ReadinGraph(&'a Arc<Term>, <G::View<'a> as GraphView>::Stream),
}

pub enum OpeningGraphState<'a, G>
where
    G: Graph + 'a,
{
    OpeningView(Pin<Box<dyn 'a + core::future::Future<Output = G::View<'a>> + Send>>),
    OpeningStream(
        Pin<
            Box<
                dyn 'a
                    + core::future::Future<
                        Output = Result<<G::View<'a> as GraphView>::Stream, GraphError>,
                    >
                    + Send,
            >,
        >,
    ),
}

impl<'a, G> Stream for SimpleDatasetStream<'a, G>
where
    G: Graph + Sync,
{
    type Item = Result<Quad, DatasetError>;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        let s = &mut *self;
        loop {
            match &mut s.state {
                SimpleDatasetStreamState::Start => {
                    if let Some(default_graph) = &s.dataset.default_graph {
                        s.state = SimpleDatasetStreamState::OpeningDefaultGraph(
                            OpeningGraphState::OpeningView(Box::pin(default_graph.view())),
                        )
                    } else {
                        s.state = SimpleDatasetStreamState::StartReadNamedGraphs;
                    }
                }
                SimpleDatasetStreamState::OpeningDefaultGraph(opening_state) => match opening_state
                {
                    OpeningGraphState::OpeningView(fut) => match fut.poll_unpin(cx) {
                        core::task::Poll::Ready(view) => {
                            *opening_state =
                                OpeningGraphState::OpeningStream(Box::pin(view.stream()))
                        }
                        core::task::Poll::Pending => break core::task::Poll::Pending,
                    },
                    OpeningGraphState::OpeningStream(fut) => match fut.poll_unpin(cx) {
                        core::task::Poll::Ready(Ok(stream)) => {
                            s.state = SimpleDatasetStreamState::ReadingDefaultGraph(stream)
                        }
                        core::task::Poll::Ready(Err(err)) => {
                            s.state = SimpleDatasetStreamState::StartReadNamedGraphs;
                            break core::task::Poll::Ready(Some(Err(DatasetError::GraphError(
                                err,
                            ))));
                        }
                        core::task::Poll::Pending => break core::task::Poll::Pending,
                    },
                },
                SimpleDatasetStreamState::ReadingDefaultGraph(stream) => {
                    match stream.poll_next_unpin(cx) {
                        core::task::Poll::Ready(Some(Ok(triple))) => {
                            break core::task::Poll::Ready(Some(Ok(
                                Quad::new(triple, None).unwrap()
                            )))
                        }
                        core::task::Poll::Ready(Some(Err(err))) => {
                            break core::task::Poll::Ready(Some(Err(DatasetError::GraphError(err))))
                        }
                        core::task::Poll::Ready(None) => {
                            s.state = SimpleDatasetStreamState::StartReadNamedGraphs;
                        }
                        core::task::Poll::Pending => break core::task::Poll::Pending,
                    }
                }
                SimpleDatasetStreamState::StartReadNamedGraphs => {
                    s.state = SimpleDatasetStreamState::ReadingNamedGraphs {
                        iter: s.dataset.graphs.iter(),
                        state: NamedGraphReadingState::Start,
                    }
                }
                SimpleDatasetStreamState::ReadingNamedGraphs { iter, state } => match state {
                    NamedGraphReadingState::Start => match iter.next() {
                        Some((name, graph)) => {
                            *state = NamedGraphReadingState::OpeningGraph(
                                name,
                                OpeningGraphState::OpeningView(Box::pin(graph.view())),
                            )
                        }
                        None => s.state = SimpleDatasetStreamState::End,
                    },
                    NamedGraphReadingState::OpeningGraph(name, opening_state) => {
                        match opening_state {
                            OpeningGraphState::OpeningView(fut) => match fut.poll_unpin(cx) {
                                core::task::Poll::Ready(view) => {
                                    *opening_state =
                                        OpeningGraphState::OpeningStream(Box::pin(view.stream()))
                                }
                                core::task::Poll::Pending => break core::task::Poll::Pending,
                            },
                            OpeningGraphState::OpeningStream(fut) => match fut.poll_unpin(cx) {
                                core::task::Poll::Ready(Ok(stream)) => {
                                    *state = NamedGraphReadingState::ReadinGraph(name, stream)
                                }
                                core::task::Poll::Ready(Err(err)) => {
                                    *state = NamedGraphReadingState::Start;
                                    break core::task::Poll::Ready(Some(Err(
                                        DatasetError::GraphError(err),
                                    )));
                                }
                                core::task::Poll::Pending => break core::task::Poll::Pending,
                            },
                        }
                    }
                    NamedGraphReadingState::ReadinGraph(name, stream) => {
                        match stream.poll_next_unpin(cx) {
                            core::task::Poll::Ready(Some(Ok(triple))) => {
                                break core::task::Poll::Ready(Some(Ok(Quad::new(
                                    triple,
                                    Some(name.clone()),
                                )
                                .unwrap())))
                            }
                            core::task::Poll::Ready(Some(Err(err))) => {
                                break core::task::Poll::Ready(Some(Err(DatasetError::GraphError(
                                    err,
                                ))))
                            }
                            core::task::Poll::Ready(None) => *state = NamedGraphReadingState::Start,
                            core::task::Poll::Pending => break core::task::Poll::Pending,
                        }
                    }
                },
                SimpleDatasetStreamState::End => break core::task::Poll::Ready(None),
            }
        }
    }
}
