tina-core 0.0.2

//! tonic request

use std::{
    ops::{Deref, DerefMut},
    str::FromStr,
};

use std::fmt::Debug;

use bytes::Buf;
use http::request::Parts;
use http_body::Body;
use prost::Message;
use tonic::{
    metadata::{MetadataKey, MetadataMap, MetadataValue},
    Extensions, Status,
};

use crate::{
    app_error_from, app_system_error,
    tina::{
        data::{app_error::AppError, grpc::request_data::GrpcReqData, AppResult},
        grpc::{FromGrpcRequest, IntoGrpcRequest, IntoGrpcResponse},
        server::{application::Application, session::Session},
    },
};

/// Request Body
pub type BoxBody = http_body::combinators::UnsyncBoxBody<bytes::Bytes, tonic::Status>;
/// Request
pub struct Request {
    pub(crate) inner: http::Request<hyper::Body>,
}

impl Request {
    /// 构建
    pub fn new(inner: http::Request<hyper::Body>) -> Self {
        Self {
            inner,
        }
    }
    /// 提取
    pub fn into_inner(self) -> http::Request<hyper::Body> {
        self.inner
    }
    /// 转换成tonic request
    pub async fn into_tonic_request(self) -> AppResult<tonic::Request<bytes::Bytes>> {
        let (parts, mut body) = self.inner.into_parts();
        match body.data().await {
            Some(r) => {
                let data = r.map_err(app_error_from!())?;
                let req = http::Request::from_parts(parts, data);
                Ok(tonic::Request::from_http(req))
            }
            None => {
                let req = http::Request::from_parts(parts, bytes::Bytes::new());
                Ok(tonic::Request::from_http(req))
            }
        }
    }
    /// box unsync
    pub fn boxed_unsync(self) -> http::Request<BoxBody> {
        let (parts, body) = self.inner.into_parts();
        let body = body.map_err(|err| Status::from_error(Box::new(err))).boxed_unsync();
        http::Request::from_parts(parts, body)
    }
    /// box
    pub fn boxed(self) -> http::Request<http_body::combinators::BoxBody<bytes::Bytes, tonic::Status>> {
        let (parts, body) = self.inner.into_parts();
        let body = body.map_err(|err| Status::from_error(Box::new(err))).boxed();
        http::Request::from_parts(parts, body)
    }
}

impl From<http::Request<hyper::Body>> for Request {
    fn from(value: http::Request<hyper::Body>) -> Self {
        Self {
            inner: value,
        }
    }
}

impl Debug for Request {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Request").field("inner", &self.inner).finish()
    }
}

impl Deref for Request {
    type Target = http::Request<hyper::Body>;

    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

impl DerefMut for Request {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.inner
    }
}

/// 从Parts转换
#[async_trait]
pub trait FromGrpcRequestParts: Sized {
    /// If the extractor fails it'll use this "rejection" type. A rejection is
    /// a kind of error that can be converted into a response.
    type Rejection: IntoGrpcResponse;

    /// Perform the extraction.
    async fn from_request_parts(parts: &mut Parts) -> Result<Self, Self::Rejection>;
}

/// 转换Parts
#[async_trait]
pub trait ToGrpcRequestParts {
    /// 转换
    async fn to_request_parts(&self, metadata: &mut MetadataMap, extensions: &mut Extensions);
}

#[async_trait]
impl FromGrpcRequestParts for Application {
    type Rejection = AppError;

    async fn from_request_parts(parts: &mut Parts) -> Result<Self, Self::Rejection> {
        match parts.extensions.get::<Application>() {
            Some(v) => Ok(v.clone()),
            None => Err(app_system_error!("No Application found from request extensions")),
        }
    }
}

#[async_trait]
impl ToGrpcRequestParts for Application {
    async fn to_request_parts(&self, _metadata: &mut MetadataMap, _extensions: &mut Extensions) {}
}

#[async_trait]
impl FromGrpcRequestParts for Session {
    type Rejection = AppError;

    async fn from_request_parts(parts: &mut Parts) -> Result<Self, Self::Rejection> {
        match parts.extensions.get::<Session>() {
            Some(v) => Ok(v.clone()),
            None => Err(app_system_error!("No Session found from request extensions")),
        }
    }
}

#[async_trait]
impl ToGrpcRequestParts for Session {
    async fn to_request_parts(&self, metadata: &mut MetadataMap, _extensions: &mut Extensions) {
        if let Some(token_value) = self.get_token() {
            let application = self.get_application();
            let security_config = match application.get_security_config() {
                Ok(v) => v,
                Err(err) => {
                    tracing::error!("{err:?}");
                    return;
                }
            };
            let token_key = security_config.token_header_name.as_str();
            let token_header_name = match MetadataKey::from_str(token_key) {
                Ok(v) => v,
                Err(err) => {
                    tracing::error!("parse token to header name failed: reason: {err:?}, token_key: {token_key}");
                    return;
                }
            };
            let token_header_value = match MetadataValue::try_from(token_value.as_ref()) {
                Ok(v) => v,
                Err(err) => {
                    tracing::error!("parse token to header value failed: reason: {err:?}, token: {token_value}");
                    return;
                }
            };
            metadata.insert(token_header_name, token_header_value);
        }
    }
}

const HEADER_SIZE: usize = std::mem::size_of::<u8>() + std::mem::size_of::<u32>();

#[async_trait]
impl<D> FromGrpcRequest for GrpcReqData<D>
where
    D: Message + Default + Debug + Send + Sync + 'static,
{
    type Request = Request;

    type Rejection = AppError;

    async fn from_grpc_request(mut req: Self::Request) -> Result<Self, Self::Rejection>
    where
        Self: Sized,
    {
        let data = req.body_mut().data().await;
        match data {
            Some(r) => match r {
                Ok(mut v) => {
                    {
                        let bytes = v.as_ref();
                        tracing::trace!("GrpcReqData receive: {bytes:?}");
                    }
                    let (compress, _) = match v.len() >= HEADER_SIZE {
                        true => {
                            let n1 = v.get_u8();
                            let n2 = v.get_u32();
                            (n1, n2)
                        }
                        false => {
                            let v = v.as_ref();
                            tracing::error!("Invalid data format: {v:?}");
                            return Err(app_system_error!("Invalid data format: {v:?}"));
                        }
                    };
                    if compress > 0 {
                        let v = v.as_ref();
                        tracing::error!("Invalid data format, first bytes must be 0: {v:?}");
                        return Err(app_system_error!("Invalid data format, first bytes must be 0: {v:?}"));
                    }
                    let d = match <D as Message>::decode(&mut v) {
                        Ok(v1) => v1,
                        Err(err) => {
                            let v = v.as_ref();
                            tracing::error!("Decode prost data failed, reason: {err:?}, data: {v:?}");
                            return Err(app_system_error!("Decode prost data failed, reason: {err:?}, data: {v:?}"));
                        }
                    };
                    let mut data = GrpcReqData::new(d);
                    {
                        let headers = req.inner.headers_mut();
                        std::mem::swap(&mut data.metadata, headers);
                    }
                    {
                        let extensions = req.inner.extensions_mut();
                        std::mem::swap(&mut data.extensions, extensions);
                    }
                    Ok(data)
                }
                Err(err) => Err(app_system_error!("Take body data from request failed, reason: {err:?}")),
            },
            None => Err(app_system_error!("No body data found from request")),
        }
    }
}

#[async_trait]
impl<D> IntoGrpcRequest for GrpcReqData<D>
where
    D: Message + Debug + Send + Sync + 'static,
{
    type Request = tonic::Request<D>;

    async fn into_grpc_request(mut self) -> Self::Request {
        let headers = self.metadata;
        let metadata = MetadataMap::from_headers(headers);
        tonic::Request::from_parts(metadata, Extensions::default(), self.data)
    }
}

#[async_trait]
impl<D, T1, T2> IntoGrpcRequest for (T1, T2)
where
    D: Message + Debug + Send + Sync + 'static,
    T1: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T2: IntoGrpcRequest<Request = tonic::Request<D>> + Debug + Send + Sync + 'static,
{
    type Request = tonic::Request<D>;

    async fn into_grpc_request(self) -> Self::Request {
        let req = self.1.into_grpc_request().await;
        let (mut metadata, mut extensions, message) = req.into_parts();

        self.0.to_request_parts(&mut metadata, &mut extensions).await;
        tonic::Request::from_parts(metadata, extensions, message)
    }
}

#[async_trait]
impl<D, T1, T2, T3> IntoGrpcRequest for (T1, T2, T3)
where
    D: Message + Debug + Send + Sync + 'static,
    T1: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T2: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T3: IntoGrpcRequest<Request = tonic::Request<D>> + Debug + Send + Sync + 'static,
{
    type Request = tonic::Request<D>;

    async fn into_grpc_request(self) -> Self::Request {
        let req = self.2.into_grpc_request().await;
        let (mut metadata, mut extensions, message) = req.into_parts();

        self.0.to_request_parts(&mut metadata, &mut extensions).await;
        self.1.to_request_parts(&mut metadata, &mut extensions).await;
        tonic::Request::from_parts(metadata, extensions, message)
    }
}

#[async_trait]
impl<D, T1, T2, T3, T4> IntoGrpcRequest for (T1, T2, T3, T4)
where
    D: Message + Debug + Send + Sync + 'static,
    T1: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T2: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T3: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T4: IntoGrpcRequest<Request = tonic::Request<D>> + Debug + Send + Sync + 'static,
{
    type Request = tonic::Request<D>;

    async fn into_grpc_request(self) -> Self::Request {
        let req = self.3.into_grpc_request().await;
        let (mut metadata, mut extensions, message) = req.into_parts();

        self.0.to_request_parts(&mut metadata, &mut extensions).await;
        self.1.to_request_parts(&mut metadata, &mut extensions).await;
        self.2.to_request_parts(&mut metadata, &mut extensions).await;
        tonic::Request::from_parts(metadata, extensions, message)
    }
}

#[async_trait]
impl<D, T1, T2, T3, T4, T5> IntoGrpcRequest for (T1, T2, T3, T4, T5)
where
    D: Message + Debug + Send + Sync + 'static,
    T1: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T2: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T3: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T4: ToGrpcRequestParts + Debug + Send + Sync + 'static,
    T5: IntoGrpcRequest<Request = tonic::Request<D>> + Debug + Send + Sync + 'static,
{
    type Request = tonic::Request<D>;

    async fn into_grpc_request(self) -> Self::Request {
        let req = self.4.into_grpc_request().await;
        let (mut metadata, mut extensions, message) = req.into_parts();

        self.0.to_request_parts(&mut metadata, &mut extensions).await;
        self.1.to_request_parts(&mut metadata, &mut extensions).await;
        self.2.to_request_parts(&mut metadata, &mut extensions).await;
        self.3.to_request_parts(&mut metadata, &mut extensions).await;
        tonic::Request::from_parts(metadata, extensions, message)
    }
}

#[async_trait]
impl<T1, T2> FromGrpcRequest for (T1, T2)
where
    T1: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T2: FromGrpcRequest<Request = Request, Rejection = AppError> + Debug + Send + Sync + 'static,
{
    type Request = Request;

    type Rejection = AppError;

    async fn from_grpc_request(req: Self::Request) -> Result<Self, Self::Rejection>
    where
        Self: Sized,
    {
        let (mut parts, body) = req.inner.into_parts();
        let t1 = T1::from_request_parts(&mut parts).await?;
        let req = Request {
            inner: http::Request::from_parts(parts, body),
        };
        let t2 = T2::from_grpc_request(req).await?;
        Ok((t1, t2))
    }
}

#[async_trait]
impl<T1, T2, T3> FromGrpcRequest for (T1, T2, T3)
where
    T1: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T2: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T3: FromGrpcRequest<Request = Request, Rejection = AppError> + Debug + Send + Sync + 'static,
{
    type Request = Request;

    type Rejection = AppError;

    async fn from_grpc_request(req: Self::Request) -> Result<Self, Self::Rejection>
    where
        Self: Sized,
    {
        let (mut parts, body) = req.inner.into_parts();
        let t1 = T1::from_request_parts(&mut parts).await?;
        let t2 = T2::from_request_parts(&mut parts).await?;
        let req = Request {
            inner: http::Request::from_parts(parts, body),
        };
        let t3 = T3::from_grpc_request(req).await?;
        Ok((t1, t2, t3))
    }
}

#[async_trait]
impl<T1, T2, T3, T4> FromGrpcRequest for (T1, T2, T3, T4)
where
    T1: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T2: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T3: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T4: FromGrpcRequest<Request = Request, Rejection = AppError> + Debug + Send + Sync + 'static,
{
    type Request = Request;

    type Rejection = AppError;

    async fn from_grpc_request(req: Self::Request) -> Result<Self, Self::Rejection>
    where
        Self: Sized,
    {
        let (mut parts, body) = req.inner.into_parts();
        let t1 = T1::from_request_parts(&mut parts).await?;
        let t2 = T2::from_request_parts(&mut parts).await?;
        let t3 = T3::from_request_parts(&mut parts).await?;
        let req = Request {
            inner: http::Request::from_parts(parts, body),
        };
        let t4 = T4::from_grpc_request(req).await?;
        Ok((t1, t2, t3, t4))
    }
}

#[async_trait]
impl<T1, T2, T3, T4, T5> FromGrpcRequest for (T1, T2, T3, T4, T5)
where
    T1: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T2: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T3: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T4: FromGrpcRequestParts<Rejection = AppError> + Debug + Send + Sync + 'static,
    T5: FromGrpcRequest<Request = Request, Rejection = AppError> + Debug + Send + Sync + 'static,
{
    type Request = Request;

    type Rejection = AppError;

    async fn from_grpc_request(req: Self::Request) -> Result<Self, Self::Rejection>
    where
        Self: Sized,
    {
        let (mut parts, body) = req.inner.into_parts();
        let t1 = T1::from_request_parts(&mut parts).await?;
        let t2 = T2::from_request_parts(&mut parts).await?;
        let t3 = T3::from_request_parts(&mut parts).await?;
        let t4 = T4::from_request_parts(&mut parts).await?;
        let req = Request {
            inner: http::Request::from_parts(parts, body),
        };
        let t5 = T5::from_grpc_request(req).await?;
        Ok((t1, t2, t3, t4, t5))
    }
}