gapirs-common 0.0.1

use super::*;
use crate::errors::ApiCallError::Call;
use crate::errors::CallError;
use crate::errors::GetUrlError;
use crate::errors::NonSuccessStatusCodeError;
use crate::errors::StdResult;
use crate::errors::{ApiCallError, ResumableUploadError};
use crate::media::resumable::{LimitedStream, PendingChunks};
use crate::media::{AsyncMediaUpload, MediaDownloadResponseStream};
use crate::utils::vec_helper::SortTupleIteratorBySortKeys;
pub use client::media;
pub use client::media::resumable::ResumableBody;
pub use client::media::resumable::ResumableMediaUpload;
use client::media::resumable::ResumableState;
pub use client::media::AsyncMediaUploadStream;
pub use client::media::MediaDownload;
pub use client::media::MediaDownloadStream;
pub use client::media::MediaUpload;
pub use client::media::MediaUploadProtocol;
pub use client::media::MediaUploadStream;
pub use client::OutgoingBodyContent;
use core::fmt::Debug;
use futures_util::lock::Mutex;
use futures_util::{AsyncSeekExt, StreamExt};
use http_body_util::{BodyExt, Full, StreamBody};
use hyper::body::Frame;
use hyper::body::{Bytes, Incoming};
use hyper::header::CONTENT_LENGTH;
use hyper::header::CONTENT_RANGE;
use hyper::header::CONTENT_TYPE;
use hyper::header::{AUTHORIZATION, RANGE};
use hyper::Request;
use hyper::Response;
use hyper::{Method, StatusCode};
use hyper_util::client::legacy::connect::Connect;
use hyper_util::client::legacy::Client;
use serde::de::DeserializeOwned;
use serde::Deserialize;
use std::collections::HashMap;
use std::io::{SeekFrom, Write};
use std::pin::Pin;
use std::str::FromStr;
use std::task::Context;
use std::task::Poll;
use url::Url;

mod client;

type Result<T> = StdResult<T, ApiCallError>;

async fn call_no_parse_response<'a, H: Hub, R2, A: ApiCall<'a, R2, H>>(
    call: &mut A,
) -> Result<Response<Incoming>> {
    let mut url: Url = call.get_url()?; //has to be called, before any call.take_*()-function from since it would change the url
    let media_upload = call.take_media_upload();
    let outgoing_body = call.take_body();
    let resumable_media = call.take_resumable_media_upload();
    let incoming_body = call.get_media_download().and_then(|x| x.get_range());
    let hub = call.get_hub()?;
    let mut outgoing_body = OutgoingBodyContent::from_body_and_media(outgoing_body, media_upload);

    let is_resuming_upload;
    if let Some(resumable) = &resumable_media {
        match resumable.state {
            ResumableState::Resuming => {
                if let Some(resume_url) = &resumable.url {
                    url = resume_url.clone();
                    is_resuming_upload = true;
                    outgoing_body = OutgoingBodyContent::Empty; //TODO: check if this is always empty when resuming (it should be right?)
                } else {
                    is_resuming_upload = false;
                }
            }
            _ => {
                is_resuming_upload = false;
            }
        }
    } else {
        is_resuming_upload = false;
    }

    let url = url.as_str();
    println!("url: '{}'", url);
    let body_len = outgoing_body.get_length();
    dbg!(&outgoing_body, &body_len, url);
    let mut req = hyper::Request::builder()
        .method(A::get_request_method())
        .uri(url)
        .header(CONTENT_LENGTH, body_len);

    if let Some(range) = incoming_body {
        dbg!("Setting range header", &range);
        req = req.header(RANGE, format!("bytes={}", range));
    }
    if is_resuming_upload {
        let total_range = resumable_media
            .as_ref()
            .and_then(|x| x.media_body.as_ref())
            .and_then(|x| x.length)
            .map(|x| x.to_string())
            .unwrap_or("*".to_string());
        req = req.header(CONTENT_RANGE, format!("bytes */{}", total_range));
    }
    if let Some(content_type) = outgoing_body.get_content_type() {
        req = req.header(CONTENT_TYPE, content_type);
    }
    dbg!(&req);
    let scopes = A::get_scopes();
    if !scopes.is_empty() {
        req = req.header(
            AUTHORIZATION,
            format!("Bearer {}", hub.get_token(&scopes).await?),
        );
    }

    let req = req.body(outgoing_body).map_err(CallError::InvalidRequest)?;
    // dbg!(&req);
    let client = hub.get_client();
    let mut response = client.request(req).await.map_err(CallError::Request)?;

    // dbg!(&response);
    let status_code = response.status();
    dbg!(&status_code);
    let resume_incompleted_upload = is_resuming_upload && status_code == 308;
    if status_code.is_success() || resume_incompleted_upload {
        if let Some(mut resumable) = resumable_media {
            dbg!("Resumable upload", &resumable, &response);
            if is_resuming_upload {
                handle_resumable_upload_resume_response(&response, &mut resumable)?;
            } else {
                handle_resumable_upload_init_response(&response, &mut resumable)?;
            }

            while let ResumableState::Sending(_) = &resumable.state {
                dbg!("Looping through resumable upload");
                response = resumable_upload_chunk(&mut resumable, hub).await?;
            }
            dbg!(&resumable, &response);
        }

        Ok(response)
        // let body = response.into_body();
        // Ok(body)
    } else {
        let body = response.into_body();
        let body = body.collect().await.map_err(CallError::GetBody)?.to_bytes();
        dbg!(&body);
        Err(CallError::NonSuccessStatusCode(
            NonSuccessStatusCodeError::from((
                status_code,
                String::from_utf8_lossy(&body).to_string(),
            )),
        ))?
    }
}

async fn call<'a, R, H: Hub, R2, A: ApiCall<'a, R2, H>>(call: &mut A) -> Result<R>
where
    R: Debug + DeserializeOwned,
{
    if call.is_media_download() {
        return Err(CallError::DownloadSetInNotDownloadCall.into());
    }
    let body = call_no_parse_response(call).await?;
    let body = body
        .into_body()
        .collect()
        .await
        .map_err(CallError::GetBody)?
        .to_bytes();
    dbg!(&body);
    let response: R = serde_json::from_slice(&body).map_err(CallError::ParseResponse)?;
    Ok(response)
}
async fn call_download<'a, H: Hub, R2, A: ApiCall<'a, R2, H>>(call: &mut A) -> Result<()> {
    if !call.is_media_download() {
        return Err(CallError::NoDownloadTarget.into());
    }
    let mut download_stream = call_get_download_response_stream(call).await?;

    let Some(mut writer) = call.take_media_download() else {
        return Err(CallError::NoDownloadTarget.into());
    };
    while let Some(frame) = download_stream.stream.next().await {
        let frame_data = match frame {
            Err(e) => {
                return Err(CallError::GetBody(e).into());
            }
            Ok(frame_data) => frame_data,
        };
        writer
            .stream
            .write_all(&frame_data)
            .map_err(CallError::WriteBodyToStream)?;
    }
    Ok(())
}

async fn call_get_download_response_stream<'a, H: Hub, R2, A: ApiCall<'a, R2, H>>(
    call: &mut A,
) -> Result<MediaDownloadResponseStream> {
    let response = call_no_parse_response(call).await?;
    let content_headers = response.headers().to_owned();
    let body = response.into_body();
    let stream = body.into_data_stream();
    let download_stream = MediaDownloadResponseStream {
        stream,
        headers: content_headers,
    };
    Ok(download_stream)
}

async fn resumable_upload_chunk(
    resumable: &mut ResumableBody,
    hub: &impl Hub,
) -> Result<Response<Incoming>> {
    println!("Uploading chunk");
    let single_chunk = resumable.is_single_chunk_upload().unwrap_or(true);
    let url = resumable.url.as_ref().ok_or(ResumableUploadError::NoUrl)?;

    let media = resumable
        .media_body
        .as_mut()
        .ok_or(ResumableUploadError::NoMediaBody)?;
    let total_length = media.length;
    if let ResumableState::Sending(ranges) = &resumable.state {
        if let Some(range) = ranges.ranges.first() {
            let range = range.resolve_range(total_length)?;
            println!("Chunk range {}/{}", range.start, range.end);
            {
                let mut stream = media.body.lock().await;
                stream
                    .seek(SeekFrom::Start(range.start))
                    .await
                    .map_err(ResumableUploadError::SeekPosition)?;
            }
            let body_len = range.end - range.start + 1; //ranges are inclusive
            let mut req_builder = hyper::Request::builder()
                .method(Method::PUT)
                .uri(url.as_str())
                .header(CONTENT_LENGTH, body_len);

            req_builder = req_builder.header(CONTENT_TYPE, &media.mime_type);
            let chunk_length = if single_chunk {
                total_length.unwrap()
            } else {
                body_len
            };
            dbg!(&chunk_length);
            let stream = LimitedStream::new(media.body.clone(), chunk_length)
                .await
                .unwrap();
            let body = OutgoingBodyContent::AsyncStream(AsyncMediaUpload::new(
                &media.mime_type,
                stream,
                chunk_length,
            ));

            let req = req_builder.body(body).unwrap();
            let client = hub.get_client();
            dbg!("sending request");
            let response = client.request(req).await.map_err(CallError::Request)?;
            dbg!(&response);
            handle_resumable_upload_resume_response(&response, resumable)?;
            dbg!("done with chunk");
            return Ok(response);
        } else {
            dbg!("missing ranges");
        }
    } else {
        dbg!("wrong state", &resumable.state);
    }

    todo!()
}

fn handle_resumable_upload_init_response(
    response: &Response<Incoming>,
    resumable: &mut ResumableBody,
) -> StdResult<(), ResumableUploadError> {
    let x = response
        .headers()
        .get("Location")
        .ok_or(ResumableUploadError::NoOrInvalidLocationHeader)?
        .to_str()
        .map_err(|_| ResumableUploadError::NoOrInvalidLocationHeader)?;
    let x = Url::from_str(x)?;
    resumable.url = Some(x.clone());
    resumable.call_save_url(x);
    resumable.state = ResumableState::Sending(PendingChunks::full(
        resumable
            .media_body
            .as_ref()
            .ok_or(ResumableUploadError::NoMediaBody)?
            .length
            .ok_or(ResumableUploadError::NoMediaBody)?,
    ));
    Ok(())
}

fn handle_resumable_upload_resume_response(
    response: &Response<Incoming>,
    resumable: &mut ResumableBody,
) -> StdResult<(), ResumableUploadError> {
    let status = response.status();
    if status.is_success() {
        resumable.state = ResumableState::Done;
    } else {
        let x = response
            .headers()
            .get("Range")
            .map(|x| x.to_str().unwrap_or_default().to_string());
        let range = if let Some(x) = x {
            PendingChunks::from_ranges(x)?
        } else {
            PendingChunks::full(
                resumable
                    .media_body
                    .as_ref()
                    .ok_or(ResumableUploadError::NoMediaBody)?
                    .length
                    .ok_or(ResumableUploadError::NoMediaBody)?,
            )
        };
        resumable.state = ResumableState::Sending(range);
    }
    Ok(())
}

pub trait ApiCall<'a, R, H: Hub>: ApiCallBase<'a, H, R> + Sized {
    fn call(&mut self) -> impl Future<Output = Result<R>>
    where
        R: Debug + DeserializeOwned,
    {
        call(self)
    }
    fn call_custom<R2>(&mut self) -> impl Future<Output = Result<R2>>
    where
        R2: Debug + DeserializeOwned,
    {
        call(self)
    }
    fn call_download(&mut self) -> impl Future<Output = Result<()>> {
        call_download(self)
    }
    fn get_url(&self) -> Result<Url> {
        let hub = self.get_hub()?;
        let base_url = &hub.get_base_url();
        let url = match self.get_protocol()? {
            MediaUploadProtocol::None => {
                let service_path = hub.get_service_path();
                self.get_normal_url(&format!("{}{}", base_url, service_path))?
            }
            MediaUploadProtocol::Simple => self.get_url_media_upload_simple(base_url)?,
            MediaUploadProtocol::Resumable => self.get_url_media_upload_resumable(base_url)?,
        };
        Ok(url)
    }
    fn get_protocol(&self) -> Result<MediaUploadProtocol> {
        Ok(
            match (self.has_media_upload(), self.has_resumable_media_upload()) {
                (true, false) => MediaUploadProtocol::Simple,
                (false, true) => MediaUploadProtocol::Resumable,
                (false, false) => MediaUploadProtocol::None,
                (true, true) => return Err(ApiCallError::MultipleMediaUploads),
            },
        )
    }
    fn get_url_media_upload_simple(&self, base_url: &str) -> StdResult<Url, GetUrlError> {
        let path = self
            .get_media_upload_path_simple()
            .ok_or(GetUrlError::NoMediaPath)?;
        self.get_url_with_path(base_url, path, Some("multipart"), false)
    }
    fn get_url_media_upload_resumable(&self, base_url: &str) -> StdResult<Url, GetUrlError> {
        let path = self
            .get_media_upload_path_resumable()
            .ok_or(GetUrlError::NoMediaPath)?;
        self.get_url_with_path(base_url, path, Some("resumable"), false)
    }
    fn get_normal_url(&self, base_url: &str) -> StdResult<Url, GetUrlError> {
        let path = self.get_method_path();
        let is_download = self.is_media_download();
        self.get_url_with_path(base_url, path, None, is_download)
    }

    fn get_url_with_path(
        &self,
        base_url: &str,
        path: impl Into<String>,
        upload_type_param: Option<&'static str>,
        download: bool,
    ) -> StdResult<Url, GetUrlError> {
        let mut path = path.into();
        let mut query_params = self.get_query_params();
        if download {
            query_params.push(("alt", "media".to_string()));
        }
        if let Some(upload_type_param) = upload_type_param {
            query_params.push(("uploadType", upload_type_param.to_string()));
        }
        let query_params = query_params;
        let mut remaining_parameters = vec![];
        //replace path parameters
        for (name, value) in query_params.into_iter() {
            if path.contains(&format!("{{{}}}", &name)) {
                path = path.replace(
                    &format!("{{{}}}", name),
                    &url::form_urlencoded::byte_serialize(value.as_bytes()).collect::<String>(),
                );
            } else {
                remaining_parameters.push((name, value));
            }
        }

        let query_param_order = self.get_query_param_order();
        let remaining_parameters = remaining_parameters.sort_by_other(&query_param_order);

        let path = path.trim_start_matches('/').trim_end_matches('/');
        let base_url = base_url.trim_end_matches('/');
        dbg!(&path, &base_url);
        let url = format!("{}/{}", base_url, path);
        let url = Url::parse_with_params(&url, remaining_parameters)?;

        Ok(url)
    }

    fn with_hub(&mut self, hub: &'a H) -> &mut Self {
        ApiCallBase::set_hub(self, hub);
        self
    }
}
impl<'a, T, R: DeserializeOwned + Debug, H: Hub> ApiCall<'a, R, H> for T where
    T: ApiCallBase<'a, H, R>
{
}
pub trait ApiCallBase<'a, HUB, RETURN> {
    fn get_method_path(&self) -> &str;

    fn get_media_upload_path_simple(&self) -> Option<&str>;
    fn get_media_upload_path_resumable(&self) -> Option<&str>;

    fn get_query_params(&self) -> Vec<(&str, String)>;
    fn get_query_param_order(&self) -> Vec<&str>;
    fn get_scopes() -> Vec<impl Scope>;
    fn get_hub(&self) -> Result<&impl Hub>;
    fn set_hub<'b>(&mut self, hub: &'b HUB)
    where
        'b: 'a;

    fn has_body(&self) -> bool;
    fn take_body(&mut self) -> Option<String>;
    fn has_media_upload(&self) -> bool;
    fn is_media_download(&self) -> bool;
    fn take_media_upload(&mut self) -> Option<MediaUpload<dyn MediaUploadStream>>;
    fn take_media_download(&mut self) -> Option<MediaDownload<dyn MediaDownloadStream>>;
    fn get_media_download(&mut self) -> Option<&MediaDownload<dyn MediaDownloadStream>>;
    fn has_resumable_media_upload(&self) -> bool;
    fn take_resumable_media_upload(&mut self) -> Option<ResumableBody>;
    fn get_request_method() -> Method;
}
pub trait ApiCallBuilder<'a, H: Hub> {
    fn new(hub: &'a H) -> Self;
}

pub trait ConnectRequirements: Connect + Clone + Send + Sync + Debug + 'static {}
impl<T> ConnectRequirements for T where T: Connect + Clone + Send + Sync + Debug + 'static {}

pub trait Hub: HubBase + Debug + Clone {
    fn call<'a, 'b, T>(
        &'a self,
        api_call: impl ApiCall<'b, T, Self> + 'b,
    ) -> impl Future<Output = Result<T>>
    where
        T: Debug + DeserializeOwned,
        Self: Sized,
        'a: 'b,
    {
        call_with_hub(api_call, self)
    }
    fn get_base_url(&self) -> &str {
        match self.get_region() {
            None => self.get_global_url(),
            Some(region) => self.get_url_for_region(region),
        }
    }
    fn get_url_for_region(&self, region: &str) -> &str {
        let endpoints = Self::get_endpoints();
        match endpoints.get(region) {
            None => self.get_global_url(),
            Some(url) => url,
        }
    }
}
async fn call_with_hub<'a, T: Debug + DeserializeOwned, H: Hub + 'a>(
    mut call: impl ApiCall<'a, T, H>,
    hub: &'a H,
) -> Result<T> {
    let x = {
        call.set_hub(hub);
        call.call().await?
    };
    Ok(x)
}
impl<T: HubBase + Debug + Clone> Hub for T {}
pub trait HubBase {
    type Connector: ConnectRequirements;
    type ConnectorAuth: ConnectRequirements;
    fn new(
        client: Client<Self::Connector, OutgoingBodyContent>,
        auth: Authenticator<Self::ConnectorAuth>,
    ) -> Self
    where
        Self: Sized;
    fn get_client(&self) -> &Client<Self::Connector, OutgoingBodyContent>;
    fn get_global_url(&self) -> &str;
    fn get_service_path(&self) -> &str;
    fn get_endpoints() -> &'static phf::Map<&'static str, &'static str>;
    fn get_token(&self, scopes: &[impl Scope]) -> impl Future<Output = Result<String>>;
    fn set_region(&mut self, region: impl Into<String>);
    fn get_region(&self) -> Option<&String>;
}

pub trait Scope: AsRef<str> {}
impl<T: AsRef<str>> Scope for T {}