use std::sync::Arc;

use futures::{stream, Stream, StreamExt, TryFutureExt};

use crate::{
    datasets::DatasetView,
    graphs::{GraphError, GraphView},
    impl_default_cbd,
    prelude::DatasetError,
    quads::{filter::TripleToQuadFiter, Quad},
    terms::Term,
    triples::Triple,
};

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct DefaultAssociatedGraphView<D>(pub D);

fn map_to_graph_err(term: Result<Arc<Term>, DatasetError>) -> Result<Arc<Term>, GraphError> {
    term.map_err(|_| GraphError::UnsupportedOp)
}

type MapToGraphErr<S> =
    stream::Map<S, fn(Result<Arc<Term>, DatasetError>) -> Result<Arc<Term>, GraphError>>;

#[macro_export]
macro_rules! impl_default_dataset_associated_view {
    () => {
        type AssociatedGraphView =
            $crate::datasets::default::reduction::DefaultAssociatedGraphView<Self>;

        async fn to_graph_view(self) -> Self::AssociatedGraphView {
            $crate::datasets::default::reduction::DefaultAssociatedGraphView(self)
        }
    };
}

impl<D> GraphView for DefaultAssociatedGraphView<D>
where
    D: DatasetView,
{
    type Stream = DefaultAssociatedStream<D::Stream>;

    async fn stream(self) -> Result<Self::Stream, crate::prelude::GraphError> {
        Ok(DefaultAssociatedStream(
            self.0
                .stream()
                .await
                .map_err(|_| GraphError::UnsupportedOp)?,
        ))
    }

    type Filter<T>
        = DefaultAssociatedGraphView<D::Filter<TripleToQuadFiter<T>>>
    where
        T: crate::prelude::TripleFilter + Unpin;

    async fn filter<F>(self, filter: F) -> Result<Self::Filter<F>, GraphError>
    where
        F: crate::prelude::TripleFilter + Unpin,
    {
        Ok(DefaultAssociatedGraphView(
            self.0
                .filter(TripleToQuadFiter(filter))
                .await
                .map_err(|_| GraphError::UnsupportedOp)?,
        ))
    }

    type Matches<M>
        = DefaultAssociatedGraphView<D::Matches<M>>
    where
        M: crate::graphs::Matches + Unpin;

    async fn matches<M>(
        self,
        matches: M,
        limit: Option<usize>,
    ) -> Result<Self::Matches<M>, GraphError>
    where
        M: crate::graphs::Matches + Unpin,
    {
        Ok(DefaultAssociatedGraphView(
            self.0
                .matches(matches, limit)
                .await
                .map_err(|_| GraphError::UnsupportedOp)?,
        ))
    }

    impl_default_cbd!();

    type SubjectStream<'g> = MapToGraphErr<D::SubjectStream<'g>>;

    async fn subjects<'g>(
        self,
        predicate: Option<&'g crate::prelude::Term>,
        object: Option<&'g crate::prelude::Term>,
        limit: Option<usize>,
    ) -> Result<Self::SubjectStream<'g>, crate::prelude::GraphError> {
        self.0
            .subjects(predicate, object, None, limit)
            .await
            .map_err(|_| GraphError::UnsupportedOp)
            .map(|s| s.map(map_to_graph_err as _))
    }

    type SubjectStreamDedup<'g> = MapToGraphErr<D::SubjectStreamDedup<'g>>;

    async fn subjects_dedup<'g>(
        self,
        predicate: Option<&'g Term>,
        object: Option<&'g Term>,
        limit: Option<usize>,
    ) -> Result<Self::SubjectStreamDedup<'g>, GraphError> {
        self.0
            .subjects_dedup(predicate, object, None, limit)
            .await
            .map_err(|_| GraphError::UnsupportedOp)
            .map(|s| s.map(map_to_graph_err as _))
    }

    type PredicateStream<'g> = MapToGraphErr<D::PredicateStream<'g>>;

    async fn predicates<'g>(
        self,
        subject: Option<&'g crate::prelude::Term>,
        object: Option<&'g crate::prelude::Term>,
        limit: Option<usize>,
    ) -> Result<Self::PredicateStream<'g>, crate::prelude::GraphError> {
        self.0
            .predicates(subject, object, None, limit)
            .await
            .map_err(|_| GraphError::UnsupportedOp)
            .map(|s| s.map(map_to_graph_err as _))
    }

    type PredicateStreamDedup<'g> = MapToGraphErr<D::PredicateStreamDedup<'g>>;

    async fn predicates_dedup<'g>(
        self,
        subject: Option<&'g Term>,
        object: Option<&'g Term>,
        limit: Option<usize>,
    ) -> Result<Self::PredicateStreamDedup<'g>, GraphError> {
        self.0
            .predicates_dedup(subject, object, None, limit)
            .await
            .map_err(|_| GraphError::UnsupportedOp)
            .map(|s| s.map(map_to_graph_err as _))
    }

    type ObjectStream<'g> = MapToGraphErr<D::ObjectStream<'g>>;

    async fn objects<'g>(
        self,
        subject: Option<&'g crate::prelude::Term>,
        predicate: Option<&'g crate::prelude::Term>,
        limit: Option<usize>,
    ) -> Result<Self::ObjectStream<'g>, crate::prelude::GraphError> {
        self.0
            .objects(subject, predicate, None, limit)
            .await
            .map_err(|_| GraphError::UnsupportedOp)
            .map(|s| s.map(map_to_graph_err as _))
    }

    type ObjectStreamDedup<'g> = MapToGraphErr<D::ObjectStreamDedup<'g>>;

    async fn objects_dedup<'g>(
        self,
        subject: Option<&'g Term>,
        predicate: Option<&'g Term>,
        limit: Option<usize>,
    ) -> Result<Self::ObjectStreamDedup<'g>, GraphError> {
        self.0
            .objects_dedup(subject, predicate, None, limit)
            .await
            .map_err(|_| GraphError::UnsupportedOp)
            .map(|s| s.map(map_to_graph_err as _))
    }

    async fn subject(
        self,
        predicate: Option<&Term>,
        object: Option<&Term>,
    ) -> Result<Option<Arc<Term>>, GraphError> {
        self.0
            .subject(predicate, object, None)
            .await
            .map_err(|_| GraphError::UnsupportedOp)
    }

    async fn predicate(
        self,
        subject: Option<&Term>,
        object: Option<&Term>,
    ) -> Result<Option<Arc<Term>>, GraphError> {
        self.0
            .predicate(subject, object, None)
            .await
            .map_err(|_| GraphError::UnsupportedOp)
    }

    async fn object(
        self,
        subject: Option<&Term>,
        predicate: Option<&Term>,
    ) -> Result<Option<Arc<Term>>, GraphError> {
        self.0
            .object(subject, predicate, None)
            .await
            .map_err(|_| GraphError::UnsupportedOp)
    }
}

pub struct DefaultAssociatedStream<S>(S);
impl<S> Stream for DefaultAssociatedStream<S>
where
    S: Stream<Item = Result<Quad, DatasetError>> + Unpin,
{
    type Item = Result<Arc<Triple>, GraphError>;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        match self.0.poll_next_unpin(cx) {
            core::task::Poll::Ready(Some(Ok(quad))) => {
                core::task::Poll::Ready(Some(Ok(quad.triple_arc())))
            }
            core::task::Poll::Ready(Some(Err(_))) => {
                core::task::Poll::Ready(Some(Err(GraphError::UnsupportedOp)))
            } // FIXME: better error type
            core::task::Poll::Ready(None) => core::task::Poll::Ready(None),
            core::task::Poll::Pending => core::task::Poll::Pending,
        }
    }
}
