#[cfg(target_arch = "wasm32")]
use crate::graphs::browser::FakeSend;

use crate::USER_AGENT;

use crate::prelude::*;

use std::sync::Mutex;

use crate::graphs::{ConfigurableGraph, GraphConfig};
use iref::InvalidIri;
use taganak_core::default_graph_view;
use taganak_core::graphs::GraphView;
use taganak_core::sources::parsers::TripleParser;
use taganak_core::sources::serializers::{GraphSerializer, TripleSerializer};

use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, CONTENT_LENGTH, CONTENT_TYPE};
use reqwest::{Body, Client};

use mediatype::MediaType;

use futures::{stream, try_join, Stream, StreamExt, TryFutureExt, TryStreamExt};
use std::iter;
use taganak_core::terms::TermError;
use taganak_core::triples::TripleRef;
use taganak_core::USizeAlias;
use taganak_orm::GraphORM;
use tokio::io::duplex;
use tokio::io::BufReader;
use tokio_util::compat::TokioAsyncReadCompatExt;
use tokio_util::io::{ReaderStream, StreamReader};

use iref::{IriBuf, IriRefBuf};

use tracing::{debug, error, info};

#[derive(Debug)]
pub struct GraphStoreHttpSourceGraph {
    client: Client,
    config: GraphStoreHttpSourceGraphConfig,
    state: Arc<Mutex<BasicProfile>>,
}

#[derive(Debug, Clone, PartialEq, Eq, GraphORM, Hash)]
#[rdf(
    prefix(ns = "sdk", iri = "http://sdk.taganak.net/vocab/taganak#"),
    rdf_type = "sdk:HttpSourceGraphConfig"
)]
pub struct GraphStoreHttpSourceGraphConfig {
    #[rdf(predicate = "sdk:sourceIri")]
    source_iri: IriRefBuf,
    #[rdf(predicate = "sdk:mediaType")]
    media_type: Option<String>,
    #[rdf(predicate = "sdk:tripleSizeHint")]
    triple_size_hint: Option<USizeAlias>,
    #[rdf(predicate = "sdk:bufSizeHint")]
    buf_size_hint: Option<USizeAlias>,
    #[rdf(predicate = "sdk:parser-base-iri")]
    base_iri: Option<IriRefBuf>,
}

#[derive(Debug, thiserror::Error)]
pub enum FromGraphError {
    #[error("the provided subject is not of type <http://sdk.taganak.net/vocab/taganak#GraphStoreHttpSourceGraph>")]
    NotHttpSourceGraphConfig,
    #[error(
        "the requested subject has no <http://sdk.taganak.net/vocab/taganak#SourceIri> predicate"
    )]
    NoSourceIri,
    #[error("the source iri is invalid: `{_0}`")]
    InvalidSourceIri(
        #[source]
        #[from]
        InvalidIri<String>,
    ),
    #[error("the source iri of the provided subject is no literal")]
    SourceIriNotLiteral,
    #[error("the provided graph config does not have the <http://sdk.taganak.net/vocab/taganak#graphConfig> attribute")]
    NoGraphConfig,
    #[error("the provided graph config is not of type <http://sdk.taganak.net/vocab/taganak#GraphStoreHttpSourceGraphConfig>")]
    NotGraphConfig,
    #[error("the provided HttpSourceGraphConfig is missing the mediatype attribute")]
    NoMediaType,
    #[error("the provided media type is not a literal")]
    MediaTypeNotLiteral,
    #[error("invalid media type {0:?}")]
    InvalidMediaType(
        #[source]
        #[from]
        TermError,
    ),
    #[error(transparent)]
    GraphError(#[from] GraphError),
}

impl GraphStoreHttpSourceGraphConfig {
    pub fn new(source_iri: IriBuf) -> Self {
        Self {
            base_iri: Some(source_iri.as_iri_ref().to_owned()),
            source_iri: source_iri.as_iri_ref().to_owned(),
            media_type: None,
            triple_size_hint: None,
            buf_size_hint: None,
        }
    }
}

impl GraphConfig for GraphStoreHttpSourceGraphConfig {
    type Graph = GraphStoreHttpSourceGraph;
}

impl ConfigurableGraph for GraphStoreHttpSourceGraph {
    type Config = GraphStoreHttpSourceGraphConfig;

    fn from_config(config: Self::Config) -> Result<Self, SourceError> {
        let client = Client::builder()
            .user_agent(USER_AGENT)
            .default_headers(Self::default_headers())
            .build()
            .map_err(|e| SourceError::UnsupportedSourceConfig)?;

        let mut profile = BasicProfile::new();
        if let Some(base) =  &config.base_iri {
            // FIXME reconsider IriRef vs. Iri everywhere here
            profile.set_base_iri(Some(base.as_iri().unwrap().to_owned()));
        }
        let state = Arc::new(profile.into());
        Ok(Self {
            client,
            config,
            state,
        })
    }
}

impl<'g> GraphView for &'g GraphStoreHttpSourceGraph {
    #[cfg(not(target_arch = "wasm32"))]
    type Stream = Box<dyn Stream<Item = Result<Arc<Triple>, GraphError>> + 'g + Send + Unpin>;
    #[cfg(target_arch = "wasm32")]
    type Stream =
        core::pin::Pin<Box<dyn Stream<Item = Result<Arc<Triple>, GraphError>> + 'g + Send>>;

    fn stream(self) -> impl core::future::Future<Output = Result<Self::Stream, GraphError>> {
        let fut = async move {
            let uri = self
                .config
                .source_iri
                .clone()
                .try_into_uri()
                .map_err(|e| SourceError::UnsupportedSourceIri)?;

            info!("Requesting Graph Store HTTP data from {}", uri.as_str());

            let res = self.client.get(uri.as_str()).send().await;
            debug!(" -> {:?}", res);
            match res {
                Ok(response) => {
                    let headers = response.headers().clone();
                    let content_type = headers
                        .get(CONTENT_TYPE)
                        .ok_or(SourceError::UnsupportedSourceConfig)?
                        .to_str()
                        .map_err(|e| SourceError::UnsupportedSourceConfig)?;
                    match headers.get(CONTENT_LENGTH) {
                        Some(h) => debug!(
                            "Received {} bytes of type {}",
                            h.to_str()
                                .map_err(|e| SourceError::UnsupportedSourceConfig)?,
                            content_type
                        ),
                        None => {
                            debug!("Received data of type {} with unknown length", content_type)
                        }
                    };

                    let media_type = MediaType::parse(content_type)
                        .map_err(|e| SourceError::UnsupportedSourceConfig)?
                        .essence()
                        .to_string();

                    let stream_reader = StreamReader::new(
                        response
                            .bytes_stream()
                            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)),
                    );

                    let parser = TripleParser::from_media_type(
                        &media_type,
                        BufReader::with_capacity(self.buf_size_hint(), stream_reader),
                        self.state.clone(),
                    )
                    .map_err(GraphError::Source)
                    .map(|s| s.map(|s| s.map_err(GraphError::Source)));

                    #[cfg(not(target_arch = "wasm32"))]
                    {
                        parser.map(Box::new).map(|e| e as _)
                    }
                    #[cfg(target_arch = "wasm32")]
                    {
                        parser.map(|e| Box::pin(FakeSend::new(e)) as _)
                    }
                }
                Err(error) => {
                    error!(
                        "Failed to load Graph Store HTTP data from {}: {}",
                        uri.as_str(),
                        error.to_string()
                    );
                    Err(GraphError::Source(SourceError::UnsupportedSourceConfig))
                }
            }
        };

        #[cfg(target_arch = "wasm32")]
        {
            FakeSend::new(fut)
        }

        #[cfg(not(target_arch = "wasm32"))]
        {
            fut
        }
    }

    default_graph_view!();

    #[inline(always)]
    fn triple_size_hint(&self) -> usize {
        match self.config.triple_size_hint {
            None => Self::default_triple_size_hint(),
            Some(size) => size as usize,
        }
    }

    #[inline(always)]
    fn buf_size_hint(&self) -> usize {
        match self.config.buf_size_hint {
            None => Self::default_buf_size_hint(),
            Some(size) => size as usize,
        }
    }
}

impl Graph for GraphStoreHttpSourceGraph {
    type View<'g> = &'g Self;

    fn view(&self) -> impl core::future::Future<Output = Self::View<'_>> + Send {
        core::future::ready(self)
    }

    async fn add(
        &mut self,
        triple: Arc<Triple>,
    ) -> Result<(), GraphError> {
        self.merge(&triple.as_triple_ref()).await
    }

    fn merge<G>(
        &mut self,
        source_graph: G,
    ) -> impl core::future::Future<Output = Result<(), GraphError>> + Send
    where
        G: GraphView,
    {
        let fut = async move {
            let uri = self
                .config
                .source_iri
                .clone()
                .try_into_uri()
                .map_err(|e| SourceError::UnsupportedSourceIri)?;
            let content_type = self.content_type();

            let size_hint =
                <&Self as GraphView>::triple_size_hint(&(self as &GraphStoreHttpSourceGraph));
            let (mut writer, reader) = duplex(size_hint);

            let serializer =
                TripleSerializer::from_media_type(&content_type, None).map_err(GraphError::Source)?;
            let write_fut = serializer.write_all(source_graph, &mut writer);

            #[cfg(not(target_arch = "wasm32"))]
            {
                let body = Body::wrap_stream(ReaderStream::new(reader));

                let read_fut = self
                    .client
                    .post(uri.as_str())
                    .header("Content-Type", content_type)
                    .body(body)
                    .send();

                // FIXME use better error type
                let (res, _) = try_join![
                    read_fut.map_err(|_| GraphError::Source(SourceError::UnsupportedSourceConfig)),
                    write_fut
                ]?;
            }

            #[cfg(target_arch = "wasm32")]
            {
                let body = wasm_streams::ReadableStream::from_async_read(reader.compat(), 1024);
                let window = web_sys::window().expect("this needs to be run in a browser");
                let mut opts = web_sys::RequestInit::new();
                opts.set_method("POST");
                let mut headers = web_sys::Headers::new().unwrap();
                headers.set("User-Agent", USER_AGENT);
                headers.set("Content-Type", &content_type);
                headers.set("Accept", Self::accept().to_str().unwrap());
                opts.set_headers(&headers);
                let body = body.into_raw();
                let body: &web_sys::wasm_bindgen::JsValue = body.as_ref();
                opts.set_body(body);
                let request = web_sys::Request::new_with_str_and_init(&uri, &opts).unwrap();
                let future =
                    wasm_bindgen_futures::JsFuture::from(window.fetch_with_request(&request));
                let (res, _) = try_join![
                    future.map_err(|_| GraphError::Source(SourceError::UnsupportedSourceConfig)),
                    write_fut
                ]?;
            }
            // FIXME handle response
            Ok(())
        };
        #[cfg(target_arch = "wasm32")]
        {
            FakeSend::new(fut)
        }
        #[cfg(not(target_arch = "wasm32"))]
        {
            fut
        }
    }
}

impl GraphStoreHttpSourceGraph {
    fn accept() -> HeaderValue {
        "application/n-triples".parse().unwrap()
    }

    fn default_headers() -> HeaderMap {
        let mut headers = HeaderMap::new();

        headers.insert(ACCEPT, Self::accept());

        headers
    }
}

impl GraphStoreHttpSourceGraph {
    fn content_type(&self) -> String {
        // FIXME consider determining the default media type somehow, while we don't have an Accept-Post header yet
        self.config
            .media_type
            .clone()
            .unwrap_or("application/n-triples".to_string())
    }
}
