use std::{collections::HashSet, sync::Arc, hash::Hash};

use futures::{Stream, StreamExt};

use crate::{graphs::GraphError, terms::Term, triples::Triple};

pub fn map_triple_to_subject(
    triple: Result<Arc<Triple>, GraphError>,
) -> Result<Arc<Term>, GraphError> {
    triple.map(|t| t.subject_arc())
}

pub fn map_triple_to_predicate(
    triple: Result<Arc<Triple>, GraphError>,
) -> Result<Arc<Term>, GraphError> {
    triple.map(|t| t.predicate_arc())
}

pub fn map_triple_to_object(
    triple: Result<Arc<Triple>, GraphError>,
) -> Result<Arc<Term>, GraphError> {
    triple.map(|t| t.object_arc())
}

#[macro_export]
macro_rules! impl_subject_stream {
    () => {
        type SubjectStream<'stream> = ::futures::stream::Map<<Self::Matches<$crate::graphs::MatchesTuple<'stream, 'stream, 'stream>> as $crate::graphs::GraphView>::Stream, fn(::std::result::Result<::std::sync::Arc<$crate::triples::Triple>, $crate::graphs::GraphError>) -> ::std::result::Result<::std::sync::Arc<$crate::terms::Term>, $crate::graphs::GraphError>>;


        async fn subjects<'stream>(
                self,
                arg1: Option<&'stream $crate::terms::Term>,
                arg2: Option<&'stream $crate::terms::Term>,
                limit: Option<usize>,
            ) -> Result<Self::SubjectStream<'stream>, $crate::prelude::GraphError> {
            use ::futures::stream::StreamExt;
            Ok(self.matches((None, arg1, arg2), limit).await?.stream().await?.map($crate::graphs::default::map_triple_to_subject))
        }
    };
}

#[macro_export]
macro_rules! impl_subject_stream_dedup {
    () => {
        type SubjectStreamDedup<'stream> =
            $crate::graphs::default::DedupStream<Self::SubjectStream<'stream>, ::std::sync::Arc<$crate::terms::Term>>;

        async fn subjects_dedup<'stream>(
            self,
            arg1: Option<&'stream $crate::terms::Term>,
            arg2: Option<&'stream $crate::terms::Term>,
            limit: Option<usize>,
        ) -> Result<Self::SubjectStreamDedup<'stream>, $crate::prelude::GraphError> {
            Ok($crate::graphs::default::DedupStream::new(
                self.subjects(arg1, arg2, limit).await?,
            ))
        }
    };
}

#[macro_export]
macro_rules! impl_object_stream {
    () => {
        type ObjectStream<'stream> = ::futures::stream::Map<<Self::Matches<$crate::graphs::MatchesTuple<'stream, 'stream, 'stream>> as $crate::graphs::GraphView>::Stream, fn(::std::result::Result<::std::sync::Arc<$crate::triples::Triple>, $crate::graphs::GraphError>) -> ::std::result::Result<::std::sync::Arc<$crate::terms::Term>, $crate::graphs::GraphError>>;

        async fn objects<'stream>(
                self,
                arg1: Option<&'stream $crate::terms::Term>,
                arg2: Option<&'stream $crate::terms::Term>,
                limit: Option<usize>,
            ) -> Result<Self::ObjectStream<'stream>, $crate::prelude::GraphError> {
            use ::futures::stream::StreamExt;
            Ok(self.matches((arg1, arg2, None), limit).await?.stream().await?.map($crate::graphs::default::map_triple_to_object))
        }
    };
}

#[macro_export]
macro_rules! impl_object_stream_dedup {
    () => {
        type ObjectStreamDedup<'stream> =
            $crate::graphs::default::DedupStream<Self::ObjectStream<'stream>, ::std::sync::Arc<$crate::terms::Term>>;

        async fn objects_dedup<'stream>(
            self,
            arg1: Option<&'stream $crate::terms::Term>,
            arg2: Option<&'stream $crate::terms::Term>,
            limit: Option<usize>,
        ) -> Result<Self::ObjectStreamDedup<'stream>, $crate::prelude::GraphError> {
            Ok($crate::graphs::default::DedupStream::new(
                self.objects(arg1, arg2, limit).await?,
            ))
        }
    };
}
#[macro_export]
macro_rules! impl_predicate_stream {
    () => {
        type PredicateStream<'stream> = ::futures::stream::Map<
            <Self::Matches<$crate::graphs::MatchesTuple<'stream, 'stream, 'stream>> as $crate::graphs::GraphView>::Stream,
            fn(
                ::std::result::Result<
                    ::std::sync::Arc<$crate::triples::Triple>,
                    $crate::graphs::GraphError,
                >,
            ) -> ::std::result::Result<
                ::std::sync::Arc<$crate::terms::Term>,
                $crate::graphs::GraphError,
            >,
        >;

        async fn predicates<'stream>(
            self,
            arg1: Option<&'stream $crate::terms::Term>,
            arg2: Option<&'stream $crate::terms::Term>,
            limit: Option<usize>,
        ) -> Result<Self::PredicateStream<'stream>, $crate::prelude::GraphError> {
            use ::futures::stream::StreamExt;
            Ok(self
                .matches((arg1, None, arg2), limit)
                .await?
                .stream()
                .await?
                .map($crate::graphs::default::map_triple_to_predicate))
        }
    };
}

#[macro_export]
macro_rules! impl_predicate_stream_dedup {
    () => {
        type PredicateStreamDedup<'stream> =
            $crate::graphs::default::DedupStream<Self::PredicateStream<'stream>, ::std::sync::Arc<$crate::terms::Term>>;

        async fn predicates_dedup<'stream>(
            self,
            arg1: Option<&'stream $crate::terms::Term>,
            arg2: Option<&'stream $crate::terms::Term>,
            limit: Option<usize>,
        ) -> Result<Self::PredicateStreamDedup<'stream>, $crate::prelude::GraphError> {
            Ok($crate::graphs::default::DedupStream::new(
                self.predicates(arg1, arg2, limit).await?,
            ))
        }
    };
}

#[macro_export]
macro_rules! impl_default_streams {
    () => {
        $crate::impl_subject_stream!();
        $crate::impl_predicate_stream!();
        $crate::impl_object_stream!();
    };
}

#[macro_export]
macro_rules! impl_default_streams_dedup {
    () => {
        $crate::impl_subject_stream_dedup!();
        $crate::impl_object_stream_dedup!();
        $crate::impl_predicate_stream_dedup!();
    };
}

pub struct DedupStream<S, I> {
    stream: S,
    visited: HashSet<I>,
}

impl<S, I> DedupStream<S, I> {
    pub fn new(stream: S) -> Self {
        Self {
            stream,
            visited: HashSet::new(),
        }
    }
}

impl<S, I, E> Stream for DedupStream<S, I>
where
    S: Stream<Item = Result<I, E>> + Unpin,
    I: Hash + Eq + Clone + Unpin
{
    type Item = Result<I, E>;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        let s = &mut *self;

        loop {
            match s.stream.poll_next_unpin(cx) {
                core::task::Poll::Ready(Some(Ok(term))) => {
                    if s.visited.contains(&term) {
                        continue;
                    }
                    s.visited.insert(term.clone());
                    break core::task::Poll::Ready(Some(Ok(term)));
                }
                core::task::Poll::Ready(Some(Err(err))) => {
                    break core::task::Poll::Ready(Some(Err(err)))
                }
                core::task::Poll::Ready(None) => break core::task::Poll::Ready(None),
                core::task::Poll::Pending => break core::task::Poll::Pending,
            }
        }
    }
}
