sprinklers_rs/sprinklers_mqtt/src/request/mod.rs

use sprinklers_actors::{ProgramRunner, SectionRunner, StateManager};
use sprinklers_core::model::Sections;

use futures_util::{ready, FutureExt};
use num_derive::FromPrimitive;
use serde::{Deserialize, Serialize};
use std::{fmt, future::Future, pin::Pin, task::Poll};

mod programs;
mod sections;

pub struct RequestContext {
    pub sections: Sections,
    pub section_runner: SectionRunner,
    pub program_runner: ProgramRunner,
    pub state_manager: StateManager,
}

type BoxFuture<Output> = Pin<Box<dyn Future<Output = Output>>>;

#[derive(Copy, Clone, Debug, PartialEq, Eq, FromPrimitive)]
#[repr(u16)]
pub enum ErrorCode {
    BadRequest = 100,
    NotSpecified = 101,
    Parse = 102,
    Range = 103,
    InvalidData = 104,
    BadToken = 105,
    Unauthorized = 106,
    NoPermission = 107,
    NotFound = 109,
    // NotUnique = 110,
    NoSuchSection = 120,
    NoSuchSectionRun = 121,
    NoSuchProgram = 122,
    Internal = 200,
    NotImplemented = 201,
    Timeout = 300,
    // ServerDisconnected = 301,
    // BrokerDisconnected = 302,
}

mod ser {
    use super::ErrorCode;
    use num_traits::FromPrimitive;
    use serde::{Deserialize, Deserializer, Serialize, Serializer};

    impl Serialize for ErrorCode {
        fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
        where
            S: Serializer,
        {
            serializer.serialize_u16(*self as u16)
        }
    }

    impl<'de> Deserialize<'de> for ErrorCode {
        fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
        where
            D: Deserializer<'de>,
        {
            let prim = u16::deserialize(deserializer)?;
            ErrorCode::from_u16(prim)
                .ok_or_else(|| <D::Error as serde::de::Error>::custom("invalid ErrorCode"))
        }
    }
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", tag = "result")]
pub struct RequestError {
    code: ErrorCode,
    message: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    name: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    cause: Option<String>,
}

impl fmt::Display for RequestError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "request error (code {:?}", self.code)?;
        if let Some(name) = &self.name {
            write!(f, "on {}", name)?;
        }
        write!(f, "): {}", self.message)?;
        if let Some(cause) = &self.cause {
            write!(f, ", caused by {}", cause)?;
        }
        Ok(())
    }
}

impl std::error::Error for RequestError {}

impl From<eyre::Report> for RequestError {
    fn from(report: eyre::Report) -> Self {
        let mut chain = report.chain();
        let message = match chain.next() {
            Some(a) => a.to_string(),
            None => "unknown error".to_string(),
        };
        let cause = chain.fold(None, |cause, err| match cause {
            Some(cause) => Some(format!("{}: {}", cause, err)),
            None => Some(err.to_string()),
        });
        RequestError::new(ErrorCode::Internal, message, None, cause)
    }
}

#[allow(dead_code)]
impl RequestError {
    pub fn new<M>(code: ErrorCode, message: M, name: Option<String>, cause: Option<String>) -> Self
    where
        M: ToString,
    {
        Self {
            code,
            message: message.to_string(),
            name,
            cause,
        }
    }

    pub fn simple<M>(code: ErrorCode, message: M) -> Self
    where
        M: ToString,
    {
        Self::new(code, message, None, None)
    }

    pub fn with_name<M, N>(code: ErrorCode, message: M, name: N) -> Self
    where
        M: ToString,
        N: ToString,
    {
        Self::new(code, message, Some(name.to_string()), None)
    }

    pub fn with_cause<M, C>(code: ErrorCode, message: M, cause: C) -> Self
    where
        M: ToString,
        C: ToString,
    {
        Self::new(code, message, None, Some(cause.to_string()))
    }

    pub fn with_name_and_cause<M, N, C>(code: ErrorCode, message: M, name: N, cause: C) -> Self
    where
        M: ToString,
        N: ToString,
        C: ToString,
    {
        Self::new(
            code,
            message,
            Some(name.to_string()),
            Some(cause.to_string()),
        )
    }
}

#[derive(Debug, Serialize, Deserialize)]
struct ResponseMessage {
    message: String,
}

impl ResponseMessage {
    fn new<M>(message: M) -> Self
    where
        M: ToString,
    {
        ResponseMessage {
            message: message.to_string(),
        }
    }
}

impl From<String> for ResponseMessage {
    fn from(message: String) -> Self {
        ResponseMessage { message }
    }
}

pub type ResponseValue = serde_json::Value;

type RequestResult<Ok = ResponseValue> = Result<Ok, RequestError>;
type RequestFuture<Ok = ResponseValue> = BoxFuture<RequestResult<Ok>>;

trait IRequest {
    type Response: Serialize;

    fn exec(self, ctx: &mut RequestContext) -> RequestFuture<Self::Response>;

    fn exec_erased(self, ctx: &mut RequestContext) -> RequestFuture
    where
        Self::Response: 'static,
        Self: Sized,
    {
        // TODO: figure out how to get rid of this nested box
        Box::pin(ErasedRequestFuture(self.exec(ctx)))
    }
}

struct ErasedRequestFuture<Response>(RequestFuture<Response>)
where
    Response: Serialize;

impl<Response> Future for ErasedRequestFuture<Response>
where
    Response: Serialize,
{
    type Output = RequestResult;

    fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
        use eyre::WrapErr;
        let response = ready!(self.as_mut().0.poll_unpin(cx));
        Poll::Ready(response.and_then(|res| {
            serde_json::to_value(res)
                .wrap_err("could not serialize response")
                .map_err(RequestError::from)
        }))
    }
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", tag = "result")]
pub enum Response {
    Success(ResponseValue),
    Error(RequestError),
}

impl From<RequestResult> for Response {
    fn from(res: RequestResult) -> Self {
        match res {
            Ok(value) => Response::Success(value),
            Err(error) => Response::Error(error),
        }
    }
}

impl From<RequestError> for Response {
    fn from(error: RequestError) -> Self {
        Response::Error(error)
    }
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct WithRequestId<T> {
    pub rid: i32,
    #[serde(flatten)]
    pub rest: T,
}

pub type ResponseWithId = WithRequestId<Response>;

#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", tag = "type")]
pub enum Request {
    RunSection(sections::RunSectionRequest),
    CancelSection(sections::CancelSectionRequest),
    CancelSectionRunId(sections::CancelSectionRunIdRequest),
    PauseSectionRunner(sections::PauseSectionRunnerRequest),
    RunProgram(programs::RunProgramRequest),
    CancelProgram(programs::CancelProgramRequest),
    UpdateProgram(programs::UpdateProgramRequest),
}

impl IRequest for Request {
    type Response = ResponseValue;

    fn exec(self, ctx: &mut RequestContext) -> RequestFuture {
        match self {
            Request::RunSection(req) => req.exec_erased(ctx),
            Request::CancelSection(req) => req.exec_erased(ctx),
            Request::CancelSectionRunId(req) => req.exec_erased(ctx),
            Request::PauseSectionRunner(req) => req.exec_erased(ctx),
            Request::RunProgram(req) => req.exec_erased(ctx),
            Request::CancelProgram(req) => req.exec_erased(ctx),
            Request::UpdateProgram(req) => req.exec_erased(ctx),
        }
    }
}

impl Request {
    pub fn execute(self, ctx: &mut RequestContext) -> impl Future<Output = Response> {
        self.exec(ctx).map(Response::from)
    }
}