Kordophone/kordophone/src/api/http_client.rs

extern crate hyper;
extern crate serde;

use std::{path::PathBuf, str};

use crate::api::{TokenStore, InMemoryTokenStore};
use hyper::{Body, Client, Method, Request, Uri};

use async_trait::async_trait;
use serde::{de::DeserializeOwned, Deserialize, Serialize};

use crate::{
    model::{Conversation, ConversationID, JwtToken, Message}, 
    APIInterface
};

type HttpClient = Client<hyper::client::HttpConnector>;

pub struct HTTPAPIClient<K: TokenStore + Send + Sync> {
    pub base_url: Uri,
    pub token_store: K,
    credentials: Option<Credentials>,
    client: HttpClient,
}

#[derive(Clone, Serialize, Deserialize, Debug)]
pub struct Credentials {
    pub username: String,
    pub password: String,
}

#[derive(Debug)]
pub enum Error {
    ClientError(String),
    HTTPError(hyper::Error),
    SerdeError(serde_json::Error),
    DecodeError,
}

impl std::error::Error for Error {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Error::HTTPError(ref err) => Some(err),
            _ => None,
        }
    }
}

impl std::fmt::Display for Error {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:?}", self)
    }
}

impl From <hyper::Error> for Error {
    fn from(err: hyper::Error) -> Error {
        Error::HTTPError(err)
    }
}

impl From <serde_json::Error> for Error {
    fn from(err: serde_json::Error) -> Error {
        Error::SerdeError(err)
    }
}

trait AuthBuilder {
    fn with_auth(self, token: &Option<JwtToken>) -> Self;
}

impl AuthBuilder for hyper::http::request::Builder {
    fn with_auth(self, token: &Option<JwtToken>) -> Self {
        if let Some(token) = &token {
            self.header("Authorization", token.to_header_value())
        } else { self }
    }
}

#[cfg(test)]
#[allow(dead_code)]
trait AuthSetting {
    fn authenticate(&mut self, token: &Option<JwtToken>);
}

#[cfg(test)]
impl<B> AuthSetting for hyper::http::Request<B> {
    fn authenticate(&mut self, token: &Option<JwtToken>) {
        if let Some(token) = &token {
            self.headers_mut().insert("Authorization", token.to_header_value());
        }
    }
}

#[async_trait]
impl<K: TokenStore + Send + Sync> APIInterface for HTTPAPIClient<K> {
    type Error = Error;

    async fn get_version(&mut self) -> Result<String, Self::Error> {
        let version: String = self.request("version", Method::GET).await?;
        Ok(version)
    }

    async fn get_conversations(&mut self) -> Result<Vec<Conversation>, Self::Error> {
        let conversations: Vec<Conversation> = self.request("conversations", Method::GET).await?;
        Ok(conversations)
    }

    async fn authenticate(&mut self, credentials: Credentials) -> Result<JwtToken, Self::Error> {
        #[derive(Deserialize, Debug)]
        struct AuthResponse {
            jwt: String,
        }

        let body = || -> Body { serde_json::to_string(&credentials).unwrap().into() };
        let token: AuthResponse = self.request_with_body_retry("authenticate", Method::POST, body, false).await?;
        let token = JwtToken::new(&token.jwt).map_err(|_| Error::DecodeError)?;
        self.token_store.set_token(token.clone()).await;
        Ok(token)
    }

    async fn get_messages(&mut self, conversation_id: &ConversationID) -> Result<Vec<Message>, Self::Error> {
        let endpoint = format!("messages?guid={}", conversation_id);
        let messages: Vec<Message> = self.request(&endpoint, Method::GET).await?;
        Ok(messages)
    }
}

impl<K: TokenStore + Send + Sync> HTTPAPIClient<K> {
    pub fn new(base_url: Uri, credentials: Option<Credentials>, token_store: K) -> HTTPAPIClient<K> {
        HTTPAPIClient { 
            base_url,
            credentials,
            token_store,
            client: Client::new(),
        }
    }

    fn uri_for_endpoint(&self, endpoint: &str) -> Uri {
        let mut parts = self.base_url.clone().into_parts();
        let root_path: PathBuf = parts.path_and_query.unwrap().path().into();
        let path = root_path.join(endpoint);
        parts.path_and_query = Some(path.to_str().unwrap().parse().unwrap());

        Uri::try_from(parts).unwrap()
    }

    async fn request<T: DeserializeOwned>(&mut self, endpoint: &str, method: Method) -> Result<T, Error> {
        self.request_with_body(endpoint, method, || { Body::empty() }).await
    }

    async fn request_with_body<T, B>(&mut self, endpoint: &str, method: Method, body_fn: B) -> Result<T, Error> 
        where T: DeserializeOwned, B: Fn() -> Body 
    {
        self.request_with_body_retry(endpoint, method, body_fn, true).await
    }

    async fn request_with_body_retry<T, B>(
        &mut self, 
        endpoint: &str, 
        method: Method, 
        body_fn: B, 
        retry_auth: bool) -> Result<T, Error>
    where 
        T: DeserializeOwned, 
        B: Fn() -> Body
    {
        use hyper::StatusCode;

        let uri = self.uri_for_endpoint(endpoint);
        log::debug!("Requesting {:?} {:?}", method, uri);

        let build_request = move |auth: &Option<JwtToken>| {
            let body = body_fn();
            Request::builder()
                .method(&method)
                .uri(&uri)
                .with_auth(auth)
                .body(body)
                .expect("Unable to build request")
        };

        let token = self.token_store.get_token().await;
        let request = build_request(&token);
        let mut response = self.client.request(request).await?;

        log::debug!("-> Response: {:}", response.status());

        match response.status() {
            StatusCode::OK => { /* cool */ },

            // 401: Unauthorized. Token may have expired or is invalid. Attempt to renew. 
            StatusCode::UNAUTHORIZED => {
                if !retry_auth {
                    return Err(Error::ClientError("Unauthorized".into()));
                }

                if let Some(credentials) = &self.credentials {
                    self.authenticate(credentials.clone()).await?;

                    let request = build_request(&token);
                    response = self.client.request(request).await?;
                }
            },

            // Other errors: bubble up. 
            _ => {
                let message = format!("Request failed ({:})", response.status());
                return Err(Error::ClientError(message));    
            }
        }

        // Read and parse response body
        let body = hyper::body::to_bytes(response.into_body()).await?;
        let parsed: T = match serde_json::from_slice(&body) {
            Ok(result) => Ok(result),
            Err(json_err) => {
                // If JSON deserialization fails, try to interpret it as plain text
                // Unfortunately the server does return things like this...
                let s = str::from_utf8(&body).map_err(|_| Error::DecodeError)?;
                serde_plain::from_str(s).map_err(|_| json_err)
            }
        }?;

        Ok(parsed)
    }
}

#[cfg(test)]
mod test {
    use super::*;

    #[cfg(test)]
    fn local_mock_client() -> HTTPAPIClient<InMemoryTokenStore> {
        let base_url = "http://localhost:5738".parse().unwrap();
        let credentials = Credentials { 
            username: "test".to_string(),
            password: "test".to_string(),
        };

        HTTPAPIClient::new(base_url, credentials.into(), InMemoryTokenStore::new())
    }

    #[cfg(test)]
    async fn mock_client_is_reachable() -> bool {
        let mut client = local_mock_client();
        let version = client.get_version().await;

        match version {
            Ok(_) => true,
            Err(e) => {
                log::error!("Mock client error: {:?}", e);
                false 
            }
        }
    }

    #[tokio::test]
    async fn test_version() {
        if !mock_client_is_reachable().await {
            log::warn!("Skipping http_client tests (mock server not reachable)");
            return;
        }

        let mut client = local_mock_client();
        let version = client.get_version().await.unwrap();
        assert!(version.starts_with("KordophoneMock-"));
    }

    #[tokio::test]
    async fn test_conversations() {
        if !mock_client_is_reachable().await {
            log::warn!("Skipping http_client tests (mock server not reachable)");
            return;
        }

        let mut client = local_mock_client();
        let conversations = client.get_conversations().await.unwrap();
        assert!(!conversations.is_empty());
    }

    #[tokio::test]
    async fn test_messages() {
        if !mock_client_is_reachable().await {
            log::warn!("Skipping http_client tests (mock server not reachable)");
            return;
        }

        let mut client = local_mock_client();
        let conversations = client.get_conversations().await.unwrap();
        let conversation = conversations.first().unwrap();
        let messages = client.get_messages(&conversation.guid).await.unwrap();
        assert!(!messages.is_empty());
    }
}