gapirs-common 0.0.1

use crate::api::client::media::multipart::{MultipartBody, MultipartState};
use crate::api::client::media::{MediaUpload, MediaUploadStream};
use crate::errors::ApiCallError;
use crate::ConnectRequirements;
use bytes::{Buf, Bytes};
use derive_more::Debug;
use futures_core::Stream;
use futures_util::AsyncRead;
use http_body_util::combinators::BoxBody;
use http_body_util::{BodyExt, Full, StreamBody};
use hyper::body::{Body, Frame};
use media::{AsyncMediaUpload, AsyncMediaUploadStream};
use serde::{Deserialize, Serialize};
use std::error::Error;
use std::io::{Read, Seek};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use url::Url;

#[derive(Default, Debug)]
pub enum OutgoingBodyContent<const N: usize = { 1024 * 10 }> {
    #[default]
    Empty,
    Json(String),
    Stream(MediaUpload<dyn MediaUploadStream>),
    AsyncStream(AsyncMediaUpload<dyn AsyncMediaUploadStream>),
    Multipart(MultipartBody),
}
impl OutgoingBodyContent {
    pub fn from_body_and_media(
        body: Option<impl Into<String>>,
        media: Option<MediaUpload<dyn MediaUploadStream>>,
    ) -> Self {
        match (body, media) {
            (Some(body), Some(media)) => {
                OutgoingBodyContent::Multipart(MultipartBody::new(media, body))
            }
            (None, Some(media)) => OutgoingBodyContent::Stream(media),
            (Some(body), None) => OutgoingBodyContent::Json(body.into()),
            (None, None) => OutgoingBodyContent::Empty,
        }
    }
    pub fn get_content_type(&self) -> Option<String> {
        Some(match self {
            OutgoingBodyContent::Empty => return None,
            OutgoingBodyContent::Json(_) => "application/json".into(),
            OutgoingBodyContent::Stream(x) => x.mime_type.as_str().into(),
            OutgoingBodyContent::AsyncStream(x) => x.mime_type.as_str().into(),
            OutgoingBodyContent::Multipart(multipart) => {
                format!("multipart/related; boundary={}", multipart.boundary)
            }
        })
    }

    pub fn get_length(&self) -> u64 {
        match self {
            OutgoingBodyContent::Empty => 0,
            OutgoingBodyContent::Json(string) => Bytes::from(string.to_string()).len() as u64,
            OutgoingBodyContent::Stream(x) => x.length,
            OutgoingBodyContent::AsyncStream(x) => x.length,
            OutgoingBodyContent::Multipart(x) => {
                x.get_body_before_media().len() as u64
                    + x.media_body.length
                    + x.get_body_after_media().len() as u64
            }
        }
    }

    fn poll_stream(
        cx: &mut Context,
        stream: &mut MediaUpload<dyn MediaUploadStream>,
    ) -> Poll<Option<Result<Bytes, Box<dyn Error + Send + Sync + 'static>>>> {
        stream.body.as_mut().poll_next(cx)
    }
}
impl<const ASYNC_STREAM_BUF_SIZE: usize> OutgoingBodyContent<ASYNC_STREAM_BUF_SIZE> {
    fn poll_stream_async(
        cx: &mut Context,
        stream: &mut AsyncMediaUpload<dyn AsyncMediaUploadStream>,
    ) -> Poll<Option<Result<Bytes, Box<dyn Error + Send + Sync + 'static>>>> {
        let mut buf = [0u8; ASYNC_STREAM_BUF_SIZE];
        let x = Pin::new(&mut stream.body).poll_read(cx, &mut buf)?;
        match x {
            Poll::Ready(x) => {
                let x = Bytes::copy_from_slice(&buf[0..x]);
                Poll::Ready(Some(Ok(x)))
            }
            Poll::Pending => Poll::Pending,
        }
    }
}

type PollResult<T, E> = Poll<Option<Result<T, Box<E>>>>;
fn bytes_to_frame<E: ?Sized>(data: PollResult<Bytes, E>) -> PollResult<Frame<Bytes>, E> {
    // println!("returning bytes: {data:?}");
    match data {
        Poll::Ready(Some(Ok(data))) => {
            if data.is_empty() {
                Poll::Ready(None)
            } else {
                Poll::Ready(Some(Ok(Frame::data(data))))
            }
        }
        Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(err))),
        Poll::Ready(None) => Poll::Ready(None),
        Poll::Pending => Poll::Pending,
    }
}
pub mod media;

impl Body for OutgoingBodyContent {
    type Data = Bytes;
    type Error = Box<dyn Error + Send + Sync + 'static>;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        match &*self {
            OutgoingBodyContent::Empty => Poll::Ready(None),
            OutgoingBodyContent::Json(value) => {
                let json_str = value.to_string();
                *self.get_mut() = OutgoingBodyContent::Empty;
                Poll::Ready(Some(Ok(Frame::data(Bytes::from(json_str)))))
            }
            OutgoingBodyContent::Multipart(_) => {
                use media::multipart::MultipartState;
                let multipart = self.get_mut();
                let multipart = match multipart {
                    OutgoingBodyContent::Multipart(multipart) => multipart,
                    _ => unreachable!(),
                };
                // dbg!("Polling multipart", &multipart);
                match multipart.state {
                    MultipartState::NotStarted => {
                        multipart.state = MultipartState::Polling;
                        let body_string = multipart.get_body_before_media();
                        Poll::Ready(Some(Ok(Frame::data(Bytes::from(body_string)))))
                    }
                    MultipartState::Polling => {
                        let poll = Self::poll_stream(cx, &mut multipart.media_body);
                        match poll {
                            Poll::Ready(None) => {
                                multipart.state = MultipartState::Done;
                                let string = multipart.get_body_after_media();
                                Poll::Ready(Some(Ok(Frame::data(Bytes::from(string)))))
                            }
                            Poll::Ready(Some(Ok(x))) => Poll::Ready(Some(Ok(Frame::data(x)))),
                            _ => bytes_to_frame(poll),
                        }
                    }
                    MultipartState::Done => Poll::Ready(None),
                }
            }
            OutgoingBodyContent::Stream(_) => {
                let stream = self.get_mut();
                let stream = match stream {
                    OutgoingBodyContent::Stream(stream) => stream,
                    _ => unreachable!(),
                };

                bytes_to_frame(Self::poll_stream(cx, stream))
            }
            OutgoingBodyContent::AsyncStream(_) => {
                let stream = self.get_mut();
                let stream = match stream {
                    OutgoingBodyContent::AsyncStream(stream) => stream,
                    _ => unreachable!(),
                };

                bytes_to_frame(Self::poll_stream_async(cx, stream))
            }
        }
    }

    fn is_end_stream(&self) -> bool {
        match self {
            OutgoingBodyContent::Empty => true,
            OutgoingBodyContent::Json(_) => false, // we will turn it into empty after first poll
            OutgoingBodyContent::Stream(_) => false,
            OutgoingBodyContent::AsyncStream(_) => false,
            OutgoingBodyContent::Multipart(_) => false,
        }
    }

    fn size_hint(&self) -> http_body::SizeHint {
        http_body::SizeHint::default()
    }
}