use core::{future::Future, marker::PhantomData, pin::Pin, task::Poll};
use std::{
    collections::{HashSet, VecDeque},
    process::Output,
    sync::Arc,
};

use futures::{FutureExt, Stream, StreamExt};
use iref::IriBuf;

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

use super::util::PhantonRefStream;

#[derive(Clone, Debug)]
pub struct CBD<G, T> {
    view: G,
    initial_term: T,
    // _marker: PhantomData<&'g ()>,
}

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

#[macro_export]
macro_rules! impl_default_cbd {
    () => {
        type CBD<TermType>
            = $crate::graphs::default::CBD<Self, TermType>
        where
            TermType: ::core::convert::AsRef<$crate::terms::Term>
                + ::core::clone::Clone
                + ::core::marker::Send;
        fn cbd<TermType>(
            self,
            subject: TermType,
        ) -> impl ::core::future::Future<
            Output = ::core::result::Result<Self::CBD<TermType>, $crate::graphs::GraphError>,
        >
        where
            TermType: ::core::convert::AsRef<$crate::terms::Term>
                + ::core::clone::Clone
                + ::core::marker::Send,
        {
            ::core::future::ready(::core::result::Result::Ok(
                $crate::graphs::default::CBD::new(self, subject),
            ))
        }
    };
}

impl<G, T> GraphView for CBD<G, T>
where
    G: GraphView + Clone,
    T: AsRef<Term> + Clone + Send,
{
    type Stream = PhantonRefStream<
        Pin<Box<dyn Stream<Item = Result<Arc<Triple>, GraphError>> + Send>>,
        (G, T),
    >;

    async fn stream(self) -> Result<Self::Stream, GraphError> {
        let rdf_subject = Term::NamedNode(
            IriBuf::new("http://www.w3.org/1999/02/22-rdf-syntax-ns#subject".to_string())
                .expect("static IRI"),
        );
        let rdf_predicate = Arc::new(Term::NamedNode(
            IriBuf::new("http://www.w3.org/1999/02/22-rdf-syntax-ns#predicate".to_string())
                .expect("static IRI"),
        ));
        let rdf_object = Arc::new(Term::NamedNode(
            IriBuf::new("http://www.w3.org/1999/02/22-rdf-syntax-ns#object".to_string())
                .expect("static IRI"),
        ));

        let stream = Box::pin(async_stream::stream! {
                let term = Arc::new(self.initial_term.as_ref().clone());
                let mut terms = VecDeque::from([term]);
                let mut viewed = HashSet::new();
                while let Some(term) = terms.pop_front() {
                    let mut description = match self.view.clone().describe(&term, None).await {
                        Ok(d) => match d.stream().await {
                            Ok(d) => d,
                            Err(e) => {
                                yield Err(e);
                                continue
                            },
                        }
                        Err(e) => {
                            yield Err(e);
                            continue;
                        },
                    };
                    while let Some(node) = description.next().await {
                        let node: Arc<Triple> = match node {
                            Ok(node) => node,
                            Err(e) => {
                                yield Err(e);
                                continue;
                            }
                        };

                        // Include all statements about blank node objects
                        let node_clone = node.clone();
                        let object = node_clone.object_arc();
                        if let Term::BlankNode(_) = &*object {
                            if !viewed.contains(&object) {
                                viewed.insert(object.clone());
                                terms.push_back(object.clone());
                            }
                        }

                        // Include all reifications
                        let mut candidates = match self.view.clone().subjects(Some(&rdf_subject), Some(node_clone.subject()), None).await {
                            Ok(c) => c,
                            Err(e) => {
                                yield Err(e);
                                continue;
                            }
                        };
                        while let Some(candidate) = candidates.next().await {
                            if let Ok(candidate) = candidate {
                                let contains_p = match self.view.clone().contains(Triple::new(candidate.clone(), rdf_predicate.clone(), node_clone.predicate_arc()).expect("all terms are already validated").as_triple_ref()).await {
                                    Ok(t) => t,
                                    Err(e) => {
                                        yield Err(e);
                                        continue;
                                    }
                                };
                                let contains_o = match self.view.clone().contains(Triple::new(candidate.clone(), rdf_object.clone(), object.clone()).expect("all terms are already validated").as_triple_ref()).await {
                                    Ok(t) => t,
                                    Err(e) => {
                                        yield Err(e);
                                        continue;
                                    }
                                };

                                if contains_p && contains_o {
                                    if !viewed.contains(&candidate) {
                                        viewed.insert(candidate.clone());
                                        terms.push_back(candidate.clone());
                                    }
                                }
                            } else if let Err(e) = candidate {
                                yield Err(e);
                                continue;
                            }
                        }

                        yield Ok(node);
                    }
                }
        }) as Pin<Box<_>>;

        Ok(PhantonRefStream {
            _marker: PhantomData,
            stream: unsafe {
                core::mem::transmute::<
                    Pin<Box<dyn Stream<Item = Result<Arc<Triple>, GraphError>> + Send>>,
                    Pin<Box<dyn Stream<Item = Result<Arc<Triple>, GraphError>> + Send + 'static>>,
                >(stream)
            },
        })
    }
    default_graph_view!();
}
