use std::sync::Arc;

use futures::{Stream, StreamExt};

use crate::{
    default_graph_view,
    graphs::{GraphError, GraphView, TripleFilter},
    triples::Triple,
};

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct DefaultFilter<G, T> {
    pub(crate) view: G,
    pub(crate) filter: T,
}

impl<G, T> DefaultFilter<G, T> {
    pub fn new(view: G, filter: T) -> Self {
        Self { view, filter }
    }
}

#[macro_export]
macro_rules! impl_default_filter {
    () => {
        type Filter<FilterType>
            = $crate::graphs::default::DefaultFilter<Self, FilterType>
        where
            FilterType: $crate::triples::filters::TripleFilter + Send + Sync + Unpin + Clone;

        async fn filter<FilterType>(
            self,
            filter: FilterType,
        ) -> Result<Self::Filter<FilterType>, $crate::prelude::GraphError>
        where
            FilterType: $crate::triples::filters::TripleFilter + Send + Sync + Unpin + Clone,
        {
            Ok($crate::graphs::default::DefaultFilter::new(self, filter))
        }
    };
}

impl<G, T> GraphView for DefaultFilter<G, T>
where
    G: GraphView,
    T: TripleFilter + Send + Sync + Unpin + Clone,
{
    type Stream = DefaultFilterStream<G::Stream, T>;
    async fn stream(self) -> Result<Self::Stream, crate::prelude::GraphError> {
        Ok(Self::Stream {
            stream: self.view.stream().await?,
            filter: self.filter,
        })
    }

    default_graph_view!();
}

pub struct DefaultFilterStream<S, T> {
    stream: S,
    filter: T,
}

impl<S, T> Stream for DefaultFilterStream<S, T>
where
    S: Stream<Item = Result<Arc<Triple>, GraphError>> + Unpin,
    T: TripleFilter + 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>> {
        loop {
            match self.stream.poll_next_unpin(cx) {
                core::task::Poll::Pending => break core::task::Poll::Pending,
                core::task::Poll::Ready(None) => break core::task::Poll::Ready(None),
                core::task::Poll::Ready(Some(t)) => match t {
                    Ok(t) => {
                        if self.filter.filter(t.as_triple_ref()) {
                            break core::task::Poll::Ready(Some(Ok(t)));
                        }
                    }
                    Err(e) => break core::task::Poll::Ready(Some(Err(e))),
                },
            }
        }
    }
    fn size_hint(&self) -> (usize, Option<usize>) {
        self.stream.size_hint()
    }
}
