ldp 0.1.0

use crate::container::ContainerType;
use crate::rdf_source::RdfSource;
use crate::{Preference, prefer, vocab};
use bytes::Bytes;
use futures::Stream;
use http::Method;
use oxigraph::io::{RdfFormat, RdfParser};
use oxigraph::model::{GraphNameRef, NamedNodeRef, Quad};
use reqwest_middleware::reqwest::{Client, Response, StatusCode, Url, header};
use reqwest_middleware::{ClientBuilder, ClientWithMiddleware, RequestBuilder};
use sfv::{Item, TokenRef};
use std::collections::BTreeSet;
use std::str::FromStr;
use tracing::error;

/// Builds an HTTP request for a [Resource](https://www.w3.org/TR/ldp/#ldpr).
///
/// # Example
/// ```rust
/// use ldp::ResourceRequestBuilder;
/// use ldp::oxigraph::model::Dataset;
/// use ldp::reqwest::{Client, Url};
/// use ldp::reqwest_middleware::ClientBuilder;
///
/// let url = Url::parse("http://server/resource")?;
/// let request = ResourceRequestBuilder::new(url)
///     .follow_described_by(true)
///     .accept_all_rdf_formats()
///     .build();
///
/// let resource = request.send().await?;
/// let rdf_source = resource.into_rdf_source::<Dataset>().await?;
/// for quad in rdf_source.dataset() {
///     println!("{:?}", quad);
/// }
/// ```
#[derive(Clone, Debug)]
pub struct ResourceRequestBuilder {
    client: ClientWithMiddleware,
    url: Url,
    follow_described_by: bool,
    validate_support: bool,
    formats: Vec<RdfFormat>,
    include_preferences: Vec<Preference>,
    omit_preferences: Vec<Preference>,
}

impl ResourceRequestBuilder {
    /// Creates a new request.
    ///
    /// A reqwest client will be created and managed by this library.
    pub fn new(url: Url) -> Self {
        Self::with_client_and_url(ClientBuilder::new(Client::new()).build(), url)
    }

    /// Creates a new request with the given HTTP client.
    #[must_use]
    pub fn with_client_and_url(client: ClientWithMiddleware, url: Url) -> Self {
        Self {
            client,
            url,
            follow_described_by: true,
            validate_support: true,
            formats: Vec::new(),
            include_preferences: Vec::new(),
            omit_preferences: Vec::new(),
        }
    }

    /// If the HTTP response includes a [describedby](https://www.w3.org/TR/ldp/#link-relation-describedby)
    /// link rel, then it will be followed. The original URL may be accessed using [`Resource::origin`],
    /// and the link rel may be accessed using [`Resource::described_by`].
    ///
    /// Note: If your intention is to fetch a NonRDFSource (e.g. a PDF, video, etc.), this value
    /// must be `false`.
    ///
    /// Default value: `true`.
    #[must_use]
    pub fn follow_described_by(mut self, value: bool) -> Self {
        self.follow_described_by = value;
        self
    }

    /// Require the presence of a `Link: <http://www.w3.org/ns/ldp#Resource>; rel="type"` response header.
    /// This is described in [section 4.2.1.4](https://www.w3.org/TR/ldp/#ldpr-resource) of the spec.
    ///
    /// Default value: `true`.
    #[must_use]
    pub fn validate_support(mut self, value: bool) -> Self {
        self.validate_support = value;
        self
    }

    /// Restrict the request to the given RDF format.
    ///
    /// Repeated calls are additive. By default, all formats — even non-RDF formats — are accepted.
    /// If your intention is to process non-RDF data, then this should not be called.
    #[must_use]
    pub fn accept_rdf_format(mut self, format: RdfFormat) -> Self {
        self.formats.push(format);
        self
    }

    /// Restrict the request to all the formats supported by the underlying parser.
    #[must_use]
    pub fn accept_all_rdf_formats(mut self) -> Self {
        self.formats = vec![
            RdfFormat::N3,
            RdfFormat::NQuads,
            RdfFormat::NTriples,
            RdfFormat::RdfXml,
            RdfFormat::TriG,
            RdfFormat::Turtle,
        ];
        self
    }

    /// Include a representation preference according to [section 7.2](https://www.w3.org/TR/ldp/#prefer-parameters).
    pub fn include_preference(mut self, preference: Preference) -> Self {
        self.include_preferences.push(preference);
        self
    }

    /// Omit a representation preference according to [section 7.2](https://www.w3.org/TR/ldp/#prefer-parameters).
    pub fn omit_preference(mut self, preference: Preference) -> Self {
        self.omit_preferences.push(preference);
        self
    }

    /// Build the request.
    pub fn build(self) -> ResourceRequest {
        ResourceRequest { builder: self }
    }
}

/// A request for a LDP [Resource](https://www.w3.org/TR/ldp/#ldpr).
#[derive(Clone, Debug)]
pub struct ResourceRequest {
    builder: ResourceRequestBuilder,
}

impl ResourceRequest {
    /// Described by example:
    /// Link: <http://fedora.quill.lan/rest/E2/fcr:metadata>; rel="describedby"
    fn extract_described_by(response: &Response) -> Option<Url> {
        if response.status() == StatusCode::OK {
            let headers = response.headers();
            for link in headers.get_all(header::LINK) {
                let link = link.to_str().unwrap_or("");
                match parse_link_header::parse(link) {
                    Ok(link_map) => {
                        if let Some(metadata_url) = link_map.get(&Some("describedby".to_string())) {
                            let raw_url = metadata_url.raw_uri.as_str();
                            return Url::parse(raw_url).ok();
                        }
                    }
                    Err(err) => error!(err = ?err, "Failed to parse Link header"),
                }
            }
        }
        None
    }

    fn ensure_ldp_support(response: &Response) -> crate::Result<()> {
        if response.status() == StatusCode::OK {
            let headers = response.headers();
            for link in headers.get_all(header::LINK) {
                let link = link.to_str().unwrap_or_default();
                match parse_link_header::parse(link) {
                    Ok(link_map) => {
                        if let Some(metadata_url) = link_map.get(&Some("type".to_string())) {
                            let raw_url = metadata_url.raw_uri.as_str();
                            if raw_url == vocab::ldp::RESOURCE {
                                return Ok(());
                            }
                        }
                    }
                    Err(err) => error!(err = ?err, "Failed to parse Link header"),
                }
            }
        }
        Err(crate::Error::LDPUnsupported)
    }

    fn resource_type(response: &Response) -> Option<ResourceType> {
        let headers = response.headers();
        let links = headers
            .get_all(header::LINK)
            .iter()
            .map(|hv| hv.to_str().unwrap_or_default())
            .filter_map(|link| parse_link_header::parse_with_rel(link).ok())
            .filter_map(|link_map| link_map.get("type").map(|item| item.raw_uri.clone()))
            .collect::<BTreeSet<_>>();

        // NB: The comparisons below must be done in this order, because multiple Link headers may
        // be present with differing degrees of specificity. E.g. BasicContainer is more specific
        // than RDFSource, but both are valid and may be present concomitantly.
        if links.contains(vocab::ldp::BASIC_CONTAINER.as_str()) {
            Some(ResourceType::RdfSource(Some(ContainerType::Basic)))
        } else if links.contains(vocab::ldp::DIRECT_CONTAINER.as_str()) {
            Some(ResourceType::RdfSource(Some(ContainerType::Direct)))
        } else if links.contains(vocab::ldp::INDIRECT_CONTAINER.as_str()) {
            Some(ResourceType::RdfSource(Some(ContainerType::Indirect)))
        } else if links.contains(vocab::ldp::RDF_SOURCE.as_str()) {
            Some(ResourceType::RdfSource(None))
        } else if links.contains(vocab::ldp::NON_RDF_SOURCE.as_str()) {
            Some(ResourceType::NonRdfSource)
        } else {
            None
        }
    }

    fn add_media_types(&self, mut request_builder: RequestBuilder) -> RequestBuilder {
        let media_types = self.builder.formats.iter().map(|f| f.media_type());
        for media_type in media_types {
            request_builder = request_builder.header(header::ACCEPT, media_type);
        }
        request_builder
    }

    fn is_method_allowed(response: &Response, method: Method) -> crate::Result<bool> {
        for header in response.headers().get_all(header::ALLOW) {
            for value in header.to_str()?.replace(" ", "").split(',') {
                if value == method {
                    return Ok(true);
                }
            }
        }
        Ok(false)
    }

    /// Send the request.
    ///
    /// There are two stages to this process. First, a `HEAD` request is made to determine whether
    /// the requested resource is described by an RDF graph at another location. If it is, and if
    /// the user [permits it](`ResourceRequestBuilder::follow_described_by`), then the URL in the
    /// [describedby](https://www.w3.org/TR/ldp/#link-relation-describedby) header is used.
    /// Otherwise, the original URL is used.
    ///
    /// During the second stage, a `GET` request is made. **The response body is not consumed yet.**
    /// To consume the response body, you must call either [`Resource::into_stream`] to receive the
    /// raw bytes or [`Resource::into_rdf_source`] to parse the content and receive a [`RdfSource`].
    ///
    /// If the `HEAD` method is not allowed on this URL (a violation of the LDP spec), the `Allow`
    /// header is inspected for `GET`. If absent, this method will return an error.
    pub async fn send(&self) -> crate::Result<Resource> {
        let request_builder = self.builder.client.head(self.builder.url.clone());
        let mut response = request_builder.send().await?;

        match response.error_for_status_ref() {
            Ok(response) => {
                if self.builder.validate_support {
                    Self::ensure_ldp_support(response)?;
                }
            }
            Err(err) if err.status() == Some(StatusCode::METHOD_NOT_ALLOWED) => {
                if !Self::is_method_allowed(&response, Method::GET)? {
                    return Err(err.into());
                }
            }
            err => {
                err?;
            }
        }

        let size = response
            .headers()
            .get(header::CONTENT_LENGTH)
            .and_then(|hv| hv.to_str().ok())
            .and_then(|et| usize::from_str(et).ok());

        let content_disposition = response
            .headers()
            .get(header::CONTENT_DISPOSITION)
            .and_then(|hv| hv.to_str().ok());

        let file_name = if let Some(content_disposition) = content_disposition {
            sfv::Parser::new(content_disposition)
                .parse::<Item>()
                .ok()
                .and_then(|item| {
                    if item.bare_item.as_token() == Some(TokenRef::constant("attachment")) {
                        item.params
                            .get("filename")
                            .and_then(|item| item.as_string().map(|s| s.to_string()))
                    } else {
                        None
                    }
                })
        } else {
            None
        };

        let resource_type = Self::resource_type(&response);

        let url_to_get;
        let described_by = Self::extract_described_by(&response);
        if let Some(new_url) = &described_by
            && self.builder.follow_described_by
        {
            url_to_get = new_url.clone();
        } else {
            url_to_get = self.builder.url.clone();
        }

        let mut request_builder = self.builder.client.get(url_to_get);
        request_builder = self.add_media_types(request_builder);

        if !self.builder.include_preferences.is_empty() || !self.builder.omit_preferences.is_empty()
        {
            let header = prefer::header_for_preferences(
                &self.builder.include_preferences,
                &self.builder.omit_preferences,
            )?;
            request_builder = request_builder.headers(header);
        }

        response = request_builder.send().await?.error_for_status()?;

        if self.builder.validate_support {
            Self::ensure_ldp_support(&response)?;
        }

        let state_token = response
            .headers()
            .get(crate::header::X_STATE_TOKEN)
            .and_then(|hv| hv.to_str().ok().map(|et| et.to_string()));

        let format = response
            .headers()
            .get(header::CONTENT_TYPE)
            .and_then(|hv| hv.to_str().ok())
            .map(|value| {
                if let Some(format) = RdfFormat::from_media_type(value) {
                    ResponseFormat::RdfFormat(format)
                } else {
                    ResponseFormat::Other(value.to_string())
                }
            })
            .unwrap_or(ResponseFormat::Unspecified);

        Ok(Resource {
            origin: self.builder.url.clone(),
            described_by,
            state_token,
            resource_type,
            format,
            file_name,
            size,
            response,
        })
    }
}

/// The type of [Resource](https://www.w3.org/TR/ldp/#ldpr-resource).
#[derive(Clone, Debug)]
pub enum ResourceType {
    NonRdfSource,
    RdfSource(Option<ContainerType>),
}

impl ResourceType {
    pub fn as_named_node_ref(&self) -> NamedNodeRef<'_> {
        match self {
            ResourceType::NonRdfSource => vocab::ldp::NON_RDF_SOURCE,
            ResourceType::RdfSource(None) => vocab::ldp::RDF_SOURCE,
            ResourceType::RdfSource(Some(container_type)) => container_type.as_named_node_ref(),
        }
    }

    pub fn from_named_node_ref(node: NamedNodeRef<'_>) -> Option<Self> {
        match node {
            vocab::ldp::NON_RDF_SOURCE => Some(ResourceType::NonRdfSource),
            vocab::ldp::RDF_SOURCE => Some(ResourceType::RdfSource(None)),
            _ => {
                let container_type = ContainerType::from_named_node_ref(node);
                if container_type.is_some() {
                    Some(ResourceType::RdfSource(container_type))
                } else {
                    None
                }
            }
        }
    }
}

/// The format of the response, as determined by the `Content-Type` HTTP header.
pub enum ResponseFormat {
    RdfFormat(RdfFormat),
    Other(String),
    Unspecified,
}

/// A LDP [Resource](https://www.w3.org/TR/ldp/#ldpr).
///
/// A Resource on its own is not very useful, as it is a very abstract concept. If your intention is
/// to treat the response as opaque data (such as a PDF or video file), call [`Resource::into_stream`].
/// In the spec, this is equivalent to a [NonRDFSource](https://www.w3.org/TR/ldp/#ldpnr).
///
/// If your intention is to treat the response as RDF data to be parsed, call
/// [`Resource::into_rdf_source`].
pub struct Resource {
    origin: Url,
    described_by: Option<Url>,
    state_token: Option<String>,
    resource_type: Option<ResourceType>,
    format: ResponseFormat,
    file_name: Option<String>,
    size: Option<usize>,
    response: Response,
}

impl Resource {
    /// The original URL used to make the request.
    pub fn origin(&self) -> &Url {
        &self.origin
    }

    /// The URL used to describe the Resource, which may be different from the Resource itself.
    ///
    /// This is useful because it allows RDF data to be attached to non-RDF data, such as a video.
    pub fn described_by(&self) -> Option<&Url> {
        self.described_by.as_ref()
    }

    /// The format of the response, as reported by the `Content-Type` HTTP header.
    pub fn format(&self) -> &ResponseFormat {
        &self.format
    }

    /// The size of the content, as reported by the `Content-Length` HTTP header.
    ///
    /// Note: This value is extracted from the response headers of the HEAD request for the origin
    /// URL.
    pub fn size(&self) -> Option<usize> {
        self.size
    }

    /// The file name of the content, as reported by the `Content-Disposition` HTTP header.
    /// Example:
    /// ```
    /// Content-Disposition: attachment; filename="new-king-james-version-en.pdf"
    /// ```
    ///
    /// Note: This value is extracted from the response headers of the HEAD request for the origin
    /// URL.
    pub fn file_name(&self) -> Option<&str> {
        self.file_name.as_deref()
    }

    /// The type of Resource, as reported by the `Link: <...> rel="type"` HTTP header.
    ///
    /// Note: This value is extracted from the response headers of the HEAD request for the origin
    /// URL.
    pub fn resource_type(&self) -> Option<&ResourceType> {
        self.resource_type.as_ref()
    }

    /// The state token, as extracted from the `X-State-Token` header.
    ///
    /// <div class="warning">
    /// This is a feature specific to <a href="https://fedora.info/2021/05/01/spec/#state-tokens">Fedora</a>.
    /// </div>
    pub fn state_token(&self) -> Option<&String> {
        self.state_token.as_ref()
    }

    /// Provide the response body as a stream of bytes.
    pub fn into_stream(self) -> impl Stream<Item = reqwest_middleware::reqwest::Result<Bytes>> {
        self.response.bytes_stream()
    }

    /// Parse the response.
    ///
    /// The `D` type parameter can be any type capable of storing [`Quad`]s. If you intend to query
    /// the dataset, use [`Dataset`](`oxigraph::model::Dataset`). If you want a stable ordering of the
    /// quads, a [`KeyedDataset`](`crate::model::KeyedDataset`), [`Vec`], or similar data structure
    /// may be used.
    pub async fn into_rdf_source<D>(self) -> crate::Result<RdfSource<D>>
    where
        D: FromIterator<Quad>,
    {
        if let ResponseFormat::RdfFormat(format) = self.format {
            let graph_url = self.described_by.as_ref().unwrap_or(&self.origin);
            let graph = GraphNameRef::NamedNode(NamedNodeRef::new_unchecked(graph_url.as_str()));
            let parser = RdfParser::from_format(format).with_default_graph(graph);
            let body = self.response.bytes().await?;
            let quads = parser.for_slice(&body);
            let dataset = quads.collect::<Result<_, _>>()?;
            Ok(RdfSource {
                origin: self.origin,
                origin_type: self.resource_type,
                state_token: self.state_token,
                described_by: self.described_by,
                file_name: self.file_name,
                size: self.size,
                dataset,
            })
        } else {
            Err(crate::Error::UnsupportedFormat)
        }
    }
}