use std::{
    collections::HashSet,
    sync::{Arc, Mutex},
};

use futures::TryStreamExt;
use winnow::stream::ContainsToken;

use crate::{
    environment::profile::ProfileError,
    graphs::{GraphError, GraphView},
    prelude::Profile,
    sources::{
        parsers::turtle::{PN_CHARS, PN_CHARS_U},
        SourceError,
    },
    terms::{Iri, Term},
    triples::Triple,
};

use super::GraphSerializer;

#[derive(Debug, Clone)]
pub struct TurtleSerializer<P> {
    profile: Arc<Mutex<P>>,
    assumed_base: Option<Iri>,
}

fn append_escaped(dest: &mut Option<String>, src: &String, last: &mut usize, cur: usize, c: char) {
    let mut temp = dest
        .take()
        .or(Some(String::with_capacity(src.len() + 32)))
        .unwrap();

    temp.push_str(&src[*last..cur]);
    temp.push('\\');
    temp.push(c);

    *last = cur;
    let _ = dest.insert(temp);
}

/// Escape the local name part of a prefixed name for use in Turtle
///
/// # Examples
///
/// ```
/// use taganak_core::sources::serializers::turtle::escape_pn_local;
///
/// assert_eq!(escape_pn_local("ex:testverbatim".into()).unwrap(), "ex:testverbatim");
/// assert_eq!(escape_pn_local("ex:test.with.dots".into()).unwrap(), "ex:test.with.dots");
/// assert_eq!(escape_pn_local("ex:test.with.dots.end.".into()).unwrap(), "ex:test.with.dots.end\\.");
/// assert_eq!(escape_pn_local("ex:.test.with.dots.begin".into()).unwrap(), "ex:\\.test.with.dots.begin");
/// assert_eq!(escape_pn_local("ex:#frag".into()).unwrap(), "ex:\\#frag");
/// assert_eq!(escape_pn_local("ex:test:with:colon".into()).unwrap(), "ex:test:with:colon");
/// assert_eq!(escape_pn_local("ex:test%nonhex".into()).unwrap(), "ex:test\\%nonhex");
/// assert_eq!(escape_pn_local("ex:test%3fhex".into()).unwrap(), "ex:test%3fhex");
/// assert_eq!(escape_pn_local("ex:test%3f%c4hex".into()).unwrap(), "ex:test%3f%c4hex");
/// assert_eq!(escape_pn_local("ex:🐢".into()).unwrap(), "ex:🐢");
/// ```
pub fn escape_pn_local(shrunk: String) -> Result<String, GraphError> {
    let mut escaped = None as Option<String>;
    let mut last_pos = 0usize;

    let mut chars = shrunk
        .char_indices()
        .skip_while(|(_, c)| c != &':')
        .peekable();
    chars.next();

    let mut first = true;
    while let Some((p, c)) = chars.next() {
        if first && (PN_CHARS_U.contains_token(c) || ('0'..='9').contains_token(c) || c == ':') {
            ()
        } else if !first
            && (PN_CHARS.contains_token(c) || (c == '.' && chars.peek().is_some()) || c == ':')
        {
            ()
        } else if c == '%' {
            let a = shrunk.get(p + 1..=p + 1);
            let b = shrunk.get(p + 2..=p + 2);

            if let (Some(a), Some(b)) = (a, b) {
                if ('0'..='9', 'A'..='F', 'a'..='f').contains_token(a.chars().next().unwrap())
                    && ('0'..='9', 'A'..='F', 'a'..='f').contains_token(b.chars().next().unwrap())
                {
                    chars.next();
                    chars.next();
                } else {
                    append_escaped(&mut escaped, &shrunk, &mut last_pos, p, c);
                }
            } else {
                append_escaped(&mut escaped, &shrunk, &mut last_pos, p, c);
            }
        } else if (
            '_', '~', '.', '-', '!', '$', '&', '\'', '(', ')', '*', '+', ',', ';', '=', '/', '?',
            '#', '@', '%',
        )
            .contains_token(c)
        {
            append_escaped(&mut escaped, &shrunk, &mut last_pos, p, c);
        } else {
            return Err(GraphError::Source(SourceError::Syntax(format!(
                "unserializable char {} in {}",
                c, shrunk
            ))));
        }

        first = false;
    }

    if let Some(mut escaped) = escaped.take() {
        escaped.push_str(&shrunk[last_pos + 1..]);
        Ok(escaped)
    } else {
        Ok(shrunk)
    }
}

impl<P> Default for TurtleSerializer<P>
where
    P: Profile,
{
    fn default() -> Self {
        Self::new(Default::default(), None)
    }
}

/// # Example
///
/// ```
/// use taganak_core::{sources::parsers::turtle::TurtleParser, environment::profile::BasicProfile};
/// use taganak_core::{graphs::Graph, sources::serializers::turtle::TurtleSerializer, sources::serializers::GraphSerializer};
/// use taganak_framework::graphs::local::memory::simple::SimpleGraph;
/// use std::sync::{Arc, Mutex};
/// use futures::{Stream, StreamExt};
///
/// const TRIPLES: &str = "
/// @prefix foo: <https://example.com/foo#> .
/// @prefix rdfs: <https://www.w3.org/2000/01/rdf-schema#> .
///
/// foo:Emil a foo:Turtle ;
/// \tfoo:age 12 ;
/// \tfoo:shell [ foo:color \"#00ff00\" ;
/// \t\tfoo:tiles 15 ] .
///
/// foo:Turtle rdfs:label \"Schildkröte\"@de ,
/// \t\t\"tortoise\"@fr .
/// ";
///
/// let mut profile = Arc::new(Mutex::new(BasicProfile::new()));
/// let mut parser = TurtleParser::new(TRIPLES.as_bytes(), profile.clone());
///
/// let serializer = TurtleSerializer::new(profile.clone(), None);
///
/// let mut temp_graph = SimpleGraph::default();
/// let mut res = Vec::with_capacity(TRIPLES.as_bytes().len());
///
/// tokio_test::block_on(async {
///   temp_graph.merge_stream(parser).await;
///   serializer.write_all(temp_graph.view().await, &mut res).await;
///
///   assert_eq!(TRIPLES.trim(), String::from_utf8(res).unwrap().trim());
/// });
/// ```
impl<P> GraphSerializer for TurtleSerializer<P>
where
    P: Profile + Send + Sync,
{
    async fn write_all<G, W>(
        &self,
        graph: G,
        writer: &mut W,
    ) -> Result<(), crate::prelude::GraphError>
    where
        G: crate::prelude::GraphView,
        W: tokio::io::AsyncWriteExt + Unpin + Send,
    {
        self.profile_serialize(writer).await?;

        let mut visited_subjects = HashSet::new();
        let mut subjects = graph.clone().subjects(None, None, None).await?;
        while let Some(subject) = subjects.try_next().await? {
            let subject: Arc<Term> = subject;
            if visited_subjects.contains(&subject) {
                continue;
            }
            visited_subjects.insert(subject.clone());
            if subject.is_blank_node() && self.term_needs_resolving(&subject, graph.clone()).await?
            {
                continue;
            }
            writer
                .write_all(format!("{} ", self.to_turtle_term(subject.clone())?).as_bytes())
                .await
                .map_err(SourceError::Io)?;
            self.subject_serialize(subject.clone(), graph.clone(), writer, 1)
                .await?;
            writer
                .write_all(".\n\n".as_bytes())
                .await
                .map_err(SourceError::Io)?;
        }
        Ok(())
    }

    // async fn is_part_of_acyclic_unique_collection<G>(node: &Term, graph: G) -> Result<bool, GraphError>
    // where
    //     G: GraphView,
    // {
    //     graph.matches(node, Some(()), object, limit)
    // }
}

impl<P> TurtleSerializer<P>
where
    P: Profile,
{
    pub fn new(profile: Arc<Mutex<P>>, assumed_base: Option<Iri>) -> Self {
        Self {
            profile,
            assumed_base,
        }
    }

    pub fn set_prefix(&mut self, prefix: String, iri: String) -> Result<(), ProfileError> {
        self.profile.lock().unwrap().set_prefix(prefix, iri)
    }

    pub fn set_base(&mut self, iri: crate::terms::Iri) -> Result<(), ProfileError> {
        self.profile.lock().unwrap().set_base_iri(Some(iri));
        Ok(())
    }

    async fn profile_serialize<W>(&self, writer: &mut W) -> Result<(), GraphError>
    where
        W: tokio::io::AsyncWriteExt + Unpin,
    {
        let mut written = false;
        let (base_iri, mut prefices) = {
            let profile = self.profile.lock().unwrap();
            (
                profile.base_iri().cloned(),
                profile
                    .iter_prefices()
                    .map(|(p, i)| (p.clone(), i.clone()))
                    .collect::<Vec<(String, String)>>(),
            )
        };
        prefices.sort();

        if self.assumed_base.is_none() {
            if let Some(base_iri) = base_iri {
                writer
                    .write_all(format!("@base <{}> .\n", base_iri).as_bytes())
                    .await
                    .map_err(SourceError::Io)?;
                written = true;
            }
        }

        for (prefix, iri) in prefices {
            writer
                .write_all(format!("@prefix {}: <{}> .\n", prefix, iri).as_bytes())
                .await
                .map_err(SourceError::Io)?;
            written = true;
        }

        if written {
            writer.write_all(b"\n").await.map_err(SourceError::Io)?;
        }

        Ok(())
    }

    async fn term_needs_resolving<G>(&self, term: &Term, graph: G) -> Result<bool, GraphError>
    where
        G: GraphView,
    {
        if !term.is_blank_node() {
            return Ok(false);
        }
        let object_occurs: Vec<_> = graph
            .matches((None, None, Some(term)), Some(2))
            .await?
            .stream()
            .await?
            .try_collect()
            .await?;
        Ok(object_occurs.len() == 1)
    }

    async fn term_is_list<G>(
        &self,
        term: Arc<Term>,
        graph: G,
    ) -> Result<Option<Vec<Arc<Term>>>, GraphError>
    where
        G: GraphView,
    {
        let mut to_visit = vec![term.clone()];
        let mut list_contents = vec![];
        while let Some(term) = to_visit.pop() {
            if !self.term_needs_resolving(&term, graph.clone()).await? {
                return Ok(None);
            }

            let subject_occurances: Vec<_> = graph
                .clone()
                .matches((Some(term.as_ref()), None, None), Some(3))
                .await?
                .stream()
                .await?
                .try_collect()
                .await?;
            let position_first = subject_occurances.iter().position(|triple| {
                triple.predicate()
                    == &"<http://www.w3.org/1999/02/22-rdf-syntax-ns#first>"
                        .try_into()
                        .unwrap()
            });
            let position_rest = subject_occurances.iter().position(|triple| {
                triple.predicate()
                    == &"<http://www.w3.org/1999/02/22-rdf-syntax-ns#rest>"
                        .try_into()
                        .unwrap()
            });

            match (subject_occurances.len(), position_first, position_rest) {
                (2, Some(position_first), Some(position_rest)) => {
                    list_contents.push(subject_occurances[position_first].object_arc());
                    if subject_occurances[position_rest].object()
                        != &TryInto::<Term>::try_into(
                            "<http://www.w3.org/1999/02/22-rdf-syntax-ns#nil>",
                        )
                        .unwrap()
                    {
                        to_visit.push(subject_occurances[position_rest].object_arc());
                    }
                }
                _ => return Ok(None),
            }
        }

        Ok(Some(list_contents))
    }

    async fn write_object<G, W>(
        &self,
        object: Arc<Term>,
        graph: G,
        writer: &mut W,
        indent_level: usize,
    ) -> Result<(), GraphError>
    where
        G: GraphView,
        W: tokio::io::AsyncWriteExt + Unpin,
    {
        let object_needs_resolving = self.term_needs_resolving(&object, graph.clone()).await?;
        if object_needs_resolving {
            let object_list = self.term_is_list(object.clone(), graph.clone()).await?;
            match object_list {
                Some(list) => {
                    writer
                        .write_all("( ".as_bytes())
                        .await
                        .map_err(SourceError::Io)?;
                    for term in list {
                        Box::pin(self.write_object(term, graph.clone(), writer, indent_level))
                            .await?;
                    }
                    writer
                        .write_all(") ".as_bytes())
                        .await
                        .map_err(SourceError::Io)?;
                }
                None => {
                    writer
                        .write_all("[ ".as_bytes())
                        .await
                        .map_err(SourceError::Io)?;
                    Box::pin(self.subject_serialize(
                        object,
                        graph.clone(),
                        writer,
                        indent_level + 1,
                    ))
                    .await?;
                    writer
                        .write_all("] ".as_bytes())
                        .await
                        .map_err(SourceError::Io)?;
                }
            }
        } else {
            writer
                .write_all(format!("{} ", self.to_turtle_term(object)?).as_bytes())
                .await
                .map_err(SourceError::Io)?;
        }
        Ok(())
    }

    fn to_turtle_term(&self, term: Arc<Term>) -> Result<String, GraphError> {
        if term.is_named_node() {
            let profile = self.profile.lock().unwrap();
            let node = term.to_iri().unwrap();
            if self.assumed_base.is_none() {
                if let Some(base) = profile.base_iri() {
                    if base == node {
                        return Ok("<>".to_string());
                    }
                }
            }
            if let Some(base) = &self.assumed_base {
                if base == node {
                    return dbg!(Ok("<>".to_string()));
                }
            }
            Ok(match profile.shrink(term.clone()) {
                Ok(s) => escape_pn_local(s)?,
                Err(_) => {
                    if self.assumed_base.is_none() {
                        if let Some(base) = profile.base_iri() {
                            let relative = node.relative_to(base);
                            return Ok(format!("<{}>", relative));
                        }
                    }
                    if let Some(base) = &self.assumed_base {
                        let relative = node.relative_to(base);
                        return Ok(format!("<{}>", relative));
                    }
                    term.to_string()
                }
            })
        } else if term.is_literal() {
            let literal = term.to_literal().unwrap();
            match literal.datatype() {
                "http://www.w3.org/2001/XMLSchema#integer"
                | "http://www.w3.org/2001/XMLSchema#decimal"
                | "http://www.w3.org/2001/XMLSchema#double"
                | "http://www.w3.org/2001/XMLSchema#boolean" => {
                    if literal.valid() {
                        Ok(literal.lexical().to_string())
                    } else {
                        Ok(term.to_string())
                    }
                }
                _ => Ok(term.to_string()),
            }
        } else {
            Ok(term.to_string())
        }
    }

    async fn subject_serialize<G, W>(
        &self,
        subject: Arc<Term>,
        graph: G,
        writer: &mut W,
        indent_level: usize,
    ) -> Result<(), GraphError>
    where
        G: GraphView,
        W: tokio::io::AsyncWriteExt + Unpin,
    {
        let mut description: Vec<Arc<Triple>> = graph
            .clone()
            .describe(&subject, None)
            .await?
            .stream()
            .await?
            .try_collect()
            .await?;
        description.sort();
        let mut prev_predicate: Option<Arc<Term>> = None;
        for triple in description {
            match &mut prev_predicate {
                Some(pred) => {
                    if pred.as_ref() == triple.predicate() {
                        writer
                            .write_all(
                                format!(",\n{:\t>width$}", "", width = indent_level + 1).as_bytes(),
                            )
                            .await
                            .map_err(SourceError::Io)?;
                        self.write_object(triple.object_arc(), graph.clone(), writer, indent_level)
                            .await?;
                    } else {
                        *pred = triple.predicate_arc();
                        writer
                            .write_all(
                                format!(
                                    ";\n{:\t>width$}{} ",
                                    "",
                                    if triple.predicate()
                                        == &"<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>"
                                            .try_into()
                                            .unwrap()
                                    {
                                        "a".to_string()
                                    } else {
                                        self.to_turtle_term(triple.predicate_arc())?
                                    },
                                    width = indent_level
                                )
                                .as_bytes(),
                            )
                            .await
                            .map_err(SourceError::Io)?;
                        self.write_object(triple.object_arc(), graph.clone(), writer, indent_level)
                            .await?;
                    }
                }
                none => {
                    *none = Some(triple.predicate_arc());
                    writer
                        .write_all(
                            format!(
                                "{} ",
                                if triple.predicate()
                                    == &"<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>"
                                        .try_into()
                                        .unwrap()
                                {
                                    "a".to_string()
                                } else {
                                    self.to_turtle_term(triple.predicate_arc())?
                                }
                            )
                            .as_bytes(),
                        )
                        .await
                        .map_err(SourceError::Io)?;
                    self.write_object(triple.object_arc(), graph.clone(), writer, indent_level)
                        .await?;
                }
            }
        }
        Ok(())
    }
}
