// FIXME re-introduce actions
//pub mod actions;

use crate::{prelude::*, triples::TripleRef};

pub use crate::triples::filters::*;

use core::ops::Deref;
use futures::{Stream, StreamExt, TryStreamExt};
use std::{
    collections::{BTreeSet, HashMap, HashSet},
    future,
    process::Output,
};

pub mod bijection;
pub mod default;
mod matches;
pub use matches::{
    FinalMatches, JoinedMatches, Matches, MatchesFilter, MatchesTuple, SimpleMatches, SubjectMatch,
    YesMatches as YesMatch,
};
pub mod either_graph;
pub mod union_graph;
use thiserror::Error;

#[derive(Debug, Error)]
pub enum GraphError {
    #[error("The operation is not supported for this graph type")]
    UnsupportedOp,
    #[error("Ambiguous results while expecting a single term")]
    Ambiguous,
    #[error("An underlying graph source failed")]
    Source(#[from] SourceError),
}

macro_rules! single_term_search {
    ($name:ident with [ $arg1:ident, $arg2:ident] on $self:ident) => {
        async move {
            let terms = $self
                .$name($arg1, $arg2, Some(2))
                .await?
                .try_collect::<Vec<Arc<Term>>>()
                .await?;

            match terms.len() {
                0 => Ok(None),
                1 => Ok(Some(terms[0].clone())),
                2 => Err(GraphError::Ambiguous),
                _ => unreachable!("how did we get here? Is limit on matches working?"),
            }
        }
    };
}

pub trait GraphView: Sized + Send + Clone {
    type Stream: Send + Unpin + Stream<Item = Result<Arc<Triple>, GraphError>>;
    fn stream(self) -> impl core::future::Future<Output = Result<Self::Stream, GraphError>> + Send;

    type Filter<T>: GraphView
    where
        T: TripleFilter + Unpin;

    fn filter<F>(
        self,
        filter: F,
    ) -> impl core::future::Future<Output = Result<Self::Filter<F>, GraphError>> + Send
    where
        F: TripleFilter + Unpin;

    type Matches<M>: GraphView
    where
        M: Matches + Unpin;
    fn matches<M>(
        self,
        matches: M,
        limit: Option<usize>,
    ) -> impl core::future::Future<Output = Result<Self::Matches<M>, GraphError>> + Send
    where
        M: Matches + Unpin;

    type CBD<T>: GraphView
    where
        T: AsRef<Term> + Clone + Send;
    /// # Example
    ///
    /// ```
    /// use taganak_core::prelude::Term;
    /// use taganak_framework::lazy_graph;
    /// use taganak_framework::prelude::{Graph, GraphView};
    /// use futures::StreamExt;
    ///
    /// lazy_graph!(graph, r##"
    /// @prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
    /// @prefix foo: <https://example.com/foo#> .
    /// @prefix rdfs: <https://www.w3.org/2000/01/rdf-schema#> .
    ///
    /// foo:Emil a foo:Turtle ;
    ///   foo:age 12 ;
    ///   foo:shell [ foo:color "#00ff00" ; foo:tiles 15 ] .
    ///
    /// [ rdf:subject foo:Emil ; rdf:predicate foo:age ; rdf:object 12 ; foo:saidBy foo:Emil ] .
    ///
    /// foo:Turtle rdfs:label "Schildkröte"@de , "tortoise"@fr .
    /// "##
    /// );
    ///
    /// tokio_test::block_on(async {
    ///     let mut cbd_stream = graph.view().await.cbd::<Term>("<https://example.com/foo#Emil>".try_into().unwrap()).await.unwrap().stream().await.unwrap();
    ///     assert_eq!(cbd_stream.count().await, 9);
    /// });
    /// ```
    fn cbd<T>(
        self,
        subject: T,
    ) -> impl core::future::Future<Output = Result<Self::CBD<T>, GraphError>> + Send
    where
        T: AsRef<Term> + Clone + Send;

    fn describe(
        self,
        subject: &Term,
        limit: Option<usize>,
    ) -> impl core::future::Future<Output = Result<Self::Matches<SubjectMatch<'_>>, GraphError>> + Send
    {
        self.matches(SubjectMatch::new(subject), limit)
    }

    type SubjectStream<'g>: Send + Unpin + Stream<Item = Result<Arc<Term>, GraphError>>;
    fn subjects<'g>(
        self,
        predicate: Option<&'g Term>,
        object: Option<&'g Term>,
        limit: Option<usize>,
    ) -> impl core::future::Future<Output = Result<Self::SubjectStream<'g>, GraphError>> + Send;

    type SubjectStreamDedup<'g>: Send + Unpin + Stream<Item = Result<Arc<Term>, GraphError>>;

    fn subjects_dedup<'g>(
        self,
        predicate: Option<&'g Term>,
        object: Option<&'g Term>,
        limit: Option<usize>,
    ) -> impl core::future::Future<Output = Result<Self::SubjectStreamDedup<'g>, GraphError>> + Send;

    type PredicateStreamDedup<'g>: Send + Unpin + Stream<Item = Result<Arc<Term>, GraphError>>;

    fn predicates_dedup<'g>(
        self,
        subject: Option<&'g Term>,
        object: Option<&'g Term>,
        limit: Option<usize>,
    ) -> impl core::future::Future<Output = Result<Self::PredicateStreamDedup<'g>, GraphError>> + Send;

    type ObjectStreamDedup<'g>: Send + Unpin + Stream<Item = Result<Arc<Term>, GraphError>>;

    fn objects_dedup<'g>(
        self,
        subject: Option<&'g Term>,
        predicate: Option<&'g Term>,
        limit: Option<usize>,
    ) -> impl core::future::Future<Output = Result<Self::ObjectStreamDedup<'g>, GraphError>> + Send;

    type PredicateStream<'g>: Send + Unpin + Stream<Item = Result<Arc<Term>, GraphError>>;
    fn predicates<'g>(
        self,
        subject: Option<&'g Term>,
        object: Option<&'g Term>,
        limit: Option<usize>,
    ) -> impl core::future::Future<Output = Result<Self::PredicateStream<'g>, GraphError>> + Send;

    type ObjectStream<'g>: Send + Unpin + Stream<Item = Result<Arc<Term>, GraphError>>;
    fn objects<'g>(
        self,
        subject: Option<&'g Term>,
        predicate: Option<&'g Term>,
        limit: Option<usize>,
    ) -> impl core::future::Future<Output = Result<Self::ObjectStream<'g>, GraphError>> + Send;

    fn subject(
        self,
        predicate: Option<&Term>,
        object: Option<&Term>,
    ) -> impl core::future::Future<Output = Result<Option<Arc<Term>>, GraphError>> + Send {
        single_term_search!(subjects with [predicate, object] on self)
    }
    fn predicate(
        self,
        subject: Option<&Term>,
        object: Option<&Term>,
    ) -> impl core::future::Future<Output = Result<Option<Arc<Term>>, GraphError>> + Send {
        single_term_search!(predicates with [subject, object] on self)
    }

    fn object(
        self,
        subject: Option<&Term>,
        predicate: Option<&Term>,
    ) -> impl core::future::Future<Output = Result<Option<Arc<Term>>, GraphError>> + Send {
        single_term_search!(objects with [subject, predicate] on self)
    }

    fn contains(
        self,
        triple: TripleRef<'_>,
    ) -> impl core::future::Future<Output = Result<bool, GraphError>> + Send {
        async move {
            let mt = (
                Some(triple.subject()),
                Some(triple.predicate()),
                Some(triple.object()),
            );
            let terms = self
                .matches(mt, Some(1))
                .await?
                .stream()
                .await?
                .count()
                .await;
            Ok(terms > 0)
        }
    }

    ///
    ///
    /// ```
    /// use taganak_core::prelude::Term;
    /// use taganak_framework::lazy_graph;
    /// use taganak_framework::prelude::{Graph, GraphView};
    /// use futures::StreamExt;
    ///
    /// lazy_graph!(graph1, r##"
    /// @prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
    /// @prefix foo: <https://example.com/foo#> .
    /// @prefix rdfs: <https://www.w3.org/2000/01/rdf-schema#> .
    ///
    /// foo:Emil a foo:Turtle ;
    ///   foo:age 12 ;
    ///   foo:shell [ foo:color "#00ff00" ; foo:tiles 15 ] .
    ///
    /// [ rdf:subject foo:Emil ; rdf:predicate foo:age ; rdf:object 12 ; foo:saidBy foo:Emil ] .
    ///
    /// foo:Turtle rdfs:label "Schildkröte"@de , "tortoise"@fr .
    /// "##
    /// );
    ///
    /// lazy_graph!(graph2, r##"
    /// @prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
    /// @prefix foo: <https://example.com/foo#> .
    /// @prefix rdfs: <https://www.w3.org/2000/01/rdf-schema#> .
    ///
    ///
    /// foo:Emil a foo:Turtle ;
    ///   foo:age 12 ;
    ///   foo:shell [ foo:color "#00ff00" ; foo:tiles 15 ] .
    ///
    /// [ rdf:subject foo:Emil ; rdf:predicate foo:age ; rdf:object 12 ; foo:saidBy foo:Emil ] .
    ///
    /// foo:Turtle rdfs:label "Schildkröte"@de , "tortoise"@fr .
    /// "##
    /// );
    ///
    /// # tokio_test::block_on(async {
    ///     assert!(graph1.view().await.isomorphic(graph2.view().await).await.unwrap());
    ///     let emil: Term = "<https://example.com/foo#Emil>".try_into().unwrap();
    ///     assert!(!graph1.view().await.isomorphic(graph2.view().await.describe(&emil,
    ///     None).await.unwrap()).await.unwrap());
    /// # });
    ///
    /// ```
    fn isomorphic<G2>(
        self,
        other: G2,
    ) -> impl core::future::Future<Output = Result<bool, GraphError>> + Send
    where
        G2: GraphView,
    {
        async move {
            let mut subjects = self.clone().subjects_dedup(None, None, None).await?;
            let mut blank_node_map: HashMap<Arc<Term>, Arc<Term>> = HashMap::new();

            let mut subj_count = 0;

            while let Some(subject) = subjects.next().await {
                subj_count += 1;
                let subject: Arc<Term> = subject?;
                let mut predicates = self
                    .clone()
                    .predicates_dedup(Some(&subject), None, None)
                    .await?;

                while let Some(predicate) = predicates.next().await {
                    let predicate: Arc<Term> = predicate?;

                    let mut objects_1: BTreeSet<_> = self
                        .clone()
                        .objects_dedup(Some(&subject), Some(&predicate), None)
                        .await?
                        .try_collect()
                        .await?;

                    let mut objects_2: BTreeSet<_> = self
                        .clone()
                        .objects_dedup(Some(&subject), Some(&predicate), None)
                        .await?
                        .try_collect()
                        .await?;

                    if objects_1.len() != objects_2.len() {
                        return Ok(false);
                    }

                    let mut to_remove = Vec::new();
                    for v in objects_1
                        .iter()
                        .filter(|v| v.is_literal() || v.is_named_node())
                    {
                        if !objects_2.contains(v) {
                            return Ok(false);
                        }
                        objects_2.remove(v);
                        to_remove.push(v.clone());
                    }
                    to_remove.into_iter().for_each(|v| {
                        objects_1.remove(&v);
                    });

                    if objects_1.len() != objects_2.len() {
                        return Ok(false);
                    }
                    let mut to_remove = Vec::new();
                    for v in objects_1
                        .iter()
                        .filter(|&term| blank_node_map.contains_key(term))
                    {
                        if !objects_2.contains(blank_node_map.get(v).unwrap()) {
                            return Ok(false);
                        }
                        objects_2.remove(blank_node_map.get(v).unwrap());
                        to_remove.push(v.clone());
                    }

                    to_remove.into_iter().for_each(|v| {
                        objects_1.remove(&v);
                    });

                    for (v1, v2) in objects_1.into_iter().zip(objects_2) {
                        blank_node_map.insert(v1, v2);
                    }
                }
            }

            let subj_count_2: usize = other
                .subjects_dedup(None, None, None)
                .await?
                .filter(|v| {
                    let result = v.is_ok();
                    async move { result }
                })
                .count()
                .await;
            Ok(subj_count == subj_count_2)
        }
    }

    /// Checks whether graph is *ground* as defined by
    /// [RDF Concepts and Abstract Syntax](https://www.w3.org/TR/rdf12-concepts/#section-terms)
    ///
    /// A graph is *ground* if all triples in it are *ground* (cf. [super::triples::Triple::ground]).
    ///
    /// # Example
    ///
    /// ```
    /// use taganak_framework::lazy_graph;
    /// use taganak_core::prelude::Term;
    /// use taganak_framework::prelude::{Graph, GraphView};
    /// use futures::StreamExt;
    ///
    /// lazy_graph!(GRAPH, r##"
    /// @base <https://example.com/> .
    ///
    /// <t1> a <Foo> ;
    ///   <sampleNumber> 12 .
    ///
    /// <t2> a <Foo> ;
    ///   <sampleList> ( 12 15 ) .
    /// "##);
    ///
    /// tokio_test::block_on(async {
    ///     assert_eq!(GRAPH.view().await.ground().await.unwrap(), false);
    ///     assert_eq!(GRAPH.view().await
    ///         .cbd::<Term>("<https://example.com/t1>".try_into().unwrap()).await.unwrap()
    ///         .ground().await.unwrap(),
    ///     true);
    /// });
    /// ```
    fn ground(self) -> impl core::future::Future<Output = Result<bool, GraphError>> + Send {
        async move {
            let mut stream = self.stream().await?;
            while let Some(t) = stream.next().await {
                match t {
                    Ok(t) if t.ground() => continue,
                    Ok(t) => return Ok(false),
                    Err(e) => return Err(e),
                };
            }

            return Ok(true);
        }
    }

    #[inline(always)]
    fn default_triple_size_hint() -> usize {
        crate::DEFAULT_TRIPLE_SIZE_HINT
    }

    #[inline(always)]
    fn default_buf_size_hint() -> usize {
        crate::DEFAULT_BUF_SIZE_HINT
    }

    #[inline(always)]
    fn triple_size_hint(&self) -> usize {
        Self::default_triple_size_hint()
    }

    #[inline(always)]
    fn buf_size_hint(&self) -> usize {
        Self::default_buf_size_hint()
    }
}

pub trait Graph: Send {
    type View<'g>: GraphView
    where
        Self: 'g;
    fn view(&self) -> impl core::future::Future<Output = Self::View<'_>> + Send;

    fn add(
        &mut self,
        triple: Arc<Triple>,
    ) -> impl core::future::Future<Output = Result<(), GraphError>> + Send {
        future::ready(Err(GraphError::UnsupportedOp))
    }
    fn remove(
        &mut self,
        triple: TripleRef<'_>,
    ) -> impl core::future::Future<Output = Result<(), GraphError>> + Send {
        future::ready(Err(GraphError::UnsupportedOp))
    }

    fn clear(&mut self) -> impl core::future::Future<Output = Result<(), GraphError>> + Send {
        future::ready(Err(GraphError::UnsupportedOp))
    }

    fn merge<G>(
        &mut self,
        source_graph: G,
    ) -> impl core::future::Future<Output = Result<(), GraphError>> + Send
    where
        G: GraphView,
    {
        async move {
            let stream = source_graph.stream().await?;
            self.merge_stream(stream).await
        }
    }

    fn merge_stream<S, E>(
        &mut self,
        mut stream: S,
    ) -> impl core::future::Future<Output = Result<(), GraphError>> + Send
    where
        S: Stream<Item = Result<Arc<Triple>, E>> + Unpin + Send,
        GraphError: From<E>,
        E: Send,
    {
        async move {
            while let Some(triple) = stream.next().await {
                self.add(triple?).await?
            }
            Ok(())
        }
    }
}

impl GraphView for () {
    type Stream = futures::stream::Empty<Result<Arc<Triple>, GraphError>>;
    async fn stream(self) -> Result<Self::Stream, GraphError> {
        Ok(futures::stream::empty())
    }

    type Filter<T>
        = ()
    where
        T: TripleFilter + Send + Sync + Unpin + Clone;

    async fn filter<F>(self, _filter: F) -> Result<Self::Filter<F>, GraphError>
    where
        F: TripleFilter + Send + Sync + Unpin + Clone,
    {
        Ok(())
    }

    type Matches<M>
        = ()
    where
        M: Matches + Unpin;
    async fn matches<M>(
        self,
        _matches: M,
        _limit: Option<usize>,
    ) -> Result<Self::Matches<M>, GraphError>
    where
        M: Matches + Unpin,
    {
        Ok(())
    }

    type CBD<T>
        = ()
    where
        T: AsRef<Term> + Clone + Send;
    async fn cbd<T>(self, _subject: T) -> Result<Self::CBD<T>, GraphError>
    where
        T: AsRef<Term> + Clone + Send,
    {
        Ok(())
    }

    type SubjectStream<'g> = futures::stream::Empty<Result<Arc<Term>, GraphError>>;
    async fn subjects<'g>(
        self,
        _predicate: Option<&'g Term>,
        _object: Option<&'g Term>,
        _limit: Option<usize>,
    ) -> Result<Self::SubjectStream<'g>, GraphError> {
        Ok(futures::stream::empty())
    }

    type SubjectStreamDedup<'g> = futures::stream::Empty<Result<Arc<Term>, GraphError>>;

    async fn subjects_dedup<'g>(
        self,
        predicate: Option<&'g Term>,
        object: Option<&'g Term>,
        limit: Option<usize>,
    ) -> Result<Self::SubjectStreamDedup<'g>, GraphError> {
        Ok(futures::stream::empty())
    }

    type PredicateStream<'g> = futures::stream::Empty<Result<Arc<Term>, GraphError>>;
    async fn predicates<'g>(
        self,
        _subject: Option<&'g Term>,
        _object: Option<&'g Term>,
        _limit: Option<usize>,
    ) -> Result<Self::PredicateStream<'g>, GraphError> {
        Ok(futures::stream::empty())
    }

    type PredicateStreamDedup<'g> = futures::stream::Empty<Result<Arc<Term>, GraphError>>;

    async fn predicates_dedup<'g>(
        self,
        subject: Option<&'g Term>,
        object: Option<&'g Term>,
        limit: Option<usize>,
    ) -> Result<Self::PredicateStreamDedup<'g>, GraphError> {
        Ok(futures::stream::empty())
    }

    type ObjectStream<'g> = futures::stream::Empty<Result<Arc<Term>, GraphError>>;
    async fn objects<'g>(
        self,
        _subject: Option<&'g Term>,
        _predicate: Option<&'g Term>,
        _limit: Option<usize>,
    ) -> Result<Self::ObjectStream<'g>, GraphError> {
        Ok(futures::stream::empty())
    }

    type ObjectStreamDedup<'g> = futures::stream::Empty<Result<Arc<Term>, GraphError>>;

    async fn objects_dedup<'g>(
        self,
        subject: Option<&'g Term>,
        predicate: Option<&'g Term>,
        limit: Option<usize>,
    ) -> Result<Self::PredicateStreamDedup<'g>, GraphError> {
        Ok(futures::stream::empty())
    }

    async fn subject(
        self,
        _predicate: Option<&Term>,
        _object: Option<&Term>,
    ) -> Result<Option<Arc<Term>>, GraphError> {
        Ok(None)
    }

    async fn predicate(
        self,
        _subject: Option<&Term>,
        _object: Option<&Term>,
    ) -> Result<Option<Arc<Term>>, GraphError> {
        Ok(None)
    }

    async fn object(
        self,
        _subject: Option<&Term>,
        _predicate: Option<&Term>,
    ) -> Result<Option<Arc<Term>>, GraphError> {
        Ok(None)
    }

    async fn contains(self, _triple: TripleRef<'_>) -> Result<bool, GraphError> {
        Ok(false)
    }

    async fn ground(self) -> Result<bool, GraphError> {
        Ok(true)
    }
    fn default_triple_size_hint() -> usize {
        0
    }

    fn default_buf_size_hint() -> usize {
        0
    }

    fn triple_size_hint(&self) -> usize {
        0
    }

    fn buf_size_hint(&self) -> usize {
        0
    }
}

impl Graph for () {
    type View<'g> = ();

    async fn view(&self) -> Self::View<'_> {
        // intentionally left empty
    }

    async fn add(&mut self, _triple: Arc<Triple>) -> Result<(), GraphError> {
        Err(GraphError::UnsupportedOp)
    }

    async fn remove(&mut self, _triple: TripleRef<'_>) -> Result<(), GraphError> {
        Err(GraphError::UnsupportedOp)
    }

    async fn merge<G>(&mut self, _source_graph: G) -> Result<(), GraphError>
    where
        G: GraphView,
    {
        Err(GraphError::UnsupportedOp)
    }

    async fn merge_stream<S, E>(&mut self, _stream: S) -> Result<(), GraphError>
    where
        S: Stream<Item = Result<Arc<Triple>, E>> + Unpin + Send,
        GraphError: From<E>,
        E: Send,
    {
        Err(GraphError::UnsupportedOp)
    }
}
