rest/
response.rs

1use actix_web::{
2    http::{header, StatusCode},
3    web::Bytes,
4    HttpRequest, HttpResponse, ResponseError,
5};
6use futures::Stream;
7use serde::Serialize;
8use serde_json::{json, Value};
9use std::{fmt, sync::Arc};
10use tonic::{Code, Status};
11
12type SuccessMapper = Arc<dyn Fn(&HttpRequest, Value) -> Value + Send + Sync>;
13type ErrorMapper =
14    Arc<dyn Fn(&HttpRequest, StatusCode, &str, &str, Option<Value>) -> Value + Send + Sync>;
15
16/// An application error with a stable machine-readable code and HTTP status.
17#[derive(Debug, Clone)]
18pub struct ApiError {
19    status: StatusCode,
20    code: String,
21    message: String,
22    details: Option<Value>,
23}
24
25impl ApiError {
26    pub fn new(status: StatusCode, code: impl Into<String>, message: impl Into<String>) -> Self {
27        Self {
28            status,
29            code: code.into(),
30            message: message.into(),
31            details: None,
32        }
33    }
34
35    pub fn with_details(mut self, details: impl Serialize) -> Result<Self, serde_json::Error> {
36        self.details = Some(serde_json::to_value(details)?);
37        Ok(self)
38    }
39
40    pub fn status(&self) -> StatusCode {
41        self.status
42    }
43
44    pub fn code(&self) -> &str {
45        &self.code
46    }
47
48    pub fn message(&self) -> &str {
49        &self.message
50    }
51
52    pub fn details(&self) -> Option<&Value> {
53        self.details.as_ref()
54    }
55}
56
57impl fmt::Display for ApiError {
58    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
59        formatter.write_str(&self.message)
60    }
61}
62
63impl std::error::Error for ApiError {}
64
65impl ResponseError for ApiError {
66    fn status_code(&self) -> StatusCode {
67        self.status
68    }
69
70    fn error_response(&self) -> HttpResponse {
71        json_response(
72            self.status,
73            json!({
74                "code": self.code,
75                "message": self.message,
76                "details": self.details,
77            }),
78        )
79    }
80}
81
82/// Per-application JSON success and error policy.
83///
84/// Mappers receive the current request, allowing envelopes to include request IDs or other
85/// context without process-global handlers. Values are fully serialized before headers are
86/// committed, so serialization failures become a deterministic HTTP 500 response.
87#[derive(Clone)]
88pub struct ResponsePolicy {
89    success_mapper: SuccessMapper,
90    error_mapper: ErrorMapper,
91}
92
93impl Default for ResponsePolicy {
94    fn default() -> Self {
95        Self {
96            success_mapper: Arc::new(|_, value| value),
97            error_mapper: Arc::new(|_, _, code, message, details| {
98                json!({
99                    "code": code,
100                    "message": message,
101                    "details": details,
102                })
103            }),
104        }
105    }
106}
107
108impl ResponsePolicy {
109    pub fn new() -> Self {
110        Self::default()
111    }
112
113    pub fn with_success_mapper<F>(mut self, mapper: F) -> Self
114    where
115        F: Fn(&HttpRequest, Value) -> Value + Send + Sync + 'static,
116    {
117        self.success_mapper = Arc::new(mapper);
118        self
119    }
120
121    pub fn with_error_mapper<F>(mut self, mapper: F) -> Self
122    where
123        F: Fn(&HttpRequest, StatusCode, &str, &str, Option<Value>) -> Value + Send + Sync + 'static,
124    {
125        self.error_mapper = Arc::new(mapper);
126        self
127    }
128
129    pub fn ok<T>(&self, request: &HttpRequest, value: T) -> HttpResponse
130    where
131        T: Serialize,
132    {
133        match serde_json::to_value(value) {
134            Ok(value) => json_response(StatusCode::OK, (self.success_mapper)(request, value)),
135            Err(error) => {
136                tracing::error!(error = %error, "failed to serialize HTTP response");
137                self.mapped_error(
138                    request,
139                    StatusCode::INTERNAL_SERVER_ERROR,
140                    "response_serialization_failed",
141                    "failed to serialize response",
142                    None,
143                )
144            }
145        }
146    }
147
148    pub fn error(&self, request: &HttpRequest, error: &ApiError) -> HttpResponse {
149        self.mapped_error(
150            request,
151            error.status,
152            &error.code,
153            &error.message,
154            error.details.clone(),
155        )
156    }
157
158    pub fn respond<T>(&self, request: &HttpRequest, result: Result<T, ApiError>) -> HttpResponse
159    where
160        T: Serialize,
161    {
162        match result {
163            Ok(value) => self.ok(request, value),
164            Err(error) => self.error(request, &error),
165        }
166    }
167
168    /// Converts a Tonic status using the same status families as go-zero's REST bridge.
169    pub fn grpc_error(&self, request: &HttpRequest, status: &Status) -> HttpResponse {
170        let http_status = grpc_status_to_http(status.code());
171        self.mapped_error(
172            request,
173            http_status,
174            grpc_code(status.code()),
175            status.message(),
176            None,
177        )
178    }
179
180    fn mapped_error(
181        &self,
182        request: &HttpRequest,
183        status: StatusCode,
184        code: &str,
185        message: &str,
186        details: Option<Value>,
187    ) -> HttpResponse {
188        json_response(
189            status,
190            (self.error_mapper)(request, status, code, message, details),
191        )
192    }
193}
194
195/// Maps a gRPC status code to its conventional HTTP equivalent.
196pub fn grpc_status_to_http(code: Code) -> StatusCode {
197    match code {
198        Code::Ok => StatusCode::OK,
199        Code::InvalidArgument | Code::FailedPrecondition | Code::OutOfRange => {
200            StatusCode::BAD_REQUEST
201        }
202        Code::Unauthenticated => StatusCode::UNAUTHORIZED,
203        Code::PermissionDenied => StatusCode::FORBIDDEN,
204        Code::NotFound => StatusCode::NOT_FOUND,
205        Code::Cancelled => StatusCode::REQUEST_TIMEOUT,
206        Code::AlreadyExists | Code::Aborted => StatusCode::CONFLICT,
207        Code::ResourceExhausted => StatusCode::TOO_MANY_REQUESTS,
208        Code::Internal | Code::DataLoss | Code::Unknown => StatusCode::INTERNAL_SERVER_ERROR,
209        Code::Unimplemented => StatusCode::NOT_IMPLEMENTED,
210        Code::Unavailable => StatusCode::SERVICE_UNAVAILABLE,
211        Code::DeadlineExceeded => StatusCode::GATEWAY_TIMEOUT,
212    }
213}
214
215/// Builds a chunked response whose stream items are explicit flush opportunities.
216///
217/// Actix forwards each yielded `Bytes` value as a body chunk. The anti-buffering headers keep
218/// common reverse proxies from coalescing those chunks indefinitely.
219pub fn streaming_response<S, E>(
220    status: StatusCode,
221    content_type: impl Into<String>,
222    stream: S,
223) -> HttpResponse
224where
225    S: Stream<Item = Result<Bytes, E>> + 'static,
226    E: std::error::Error + 'static,
227{
228    HttpResponse::build(status)
229        .insert_header((header::CONTENT_TYPE, content_type.into()))
230        .insert_header((header::CACHE_CONTROL, "no-cache, no-transform"))
231        .insert_header(("x-accel-buffering", "no"))
232        .streaming(stream)
233}
234
235fn json_response(status: StatusCode, value: Value) -> HttpResponse {
236    // Value serialization is infallible for valid serde_json values. Serializing before building
237    // the response guarantees no success status is committed if that invariant ever changes.
238    match serde_json::to_vec(&value) {
239        Ok(body) => HttpResponse::build(status)
240            .insert_header((header::CONTENT_TYPE, "application/json"))
241            .body(body),
242        Err(error) => {
243            tracing::error!(error = %error, "failed to serialize mapped HTTP response");
244            HttpResponse::InternalServerError()
245                .insert_header((header::CONTENT_TYPE, "application/json"))
246                .body(r#"{"code":"response_serialization_failed","message":"failed to serialize response"}"#)
247        }
248    }
249}
250
251fn grpc_code(code: Code) -> &'static str {
252    match code {
253        Code::Ok => "ok",
254        Code::Cancelled => "cancelled",
255        Code::Unknown => "unknown",
256        Code::InvalidArgument => "invalid_argument",
257        Code::DeadlineExceeded => "deadline_exceeded",
258        Code::NotFound => "not_found",
259        Code::AlreadyExists => "already_exists",
260        Code::PermissionDenied => "permission_denied",
261        Code::ResourceExhausted => "resource_exhausted",
262        Code::FailedPrecondition => "failed_precondition",
263        Code::Aborted => "aborted",
264        Code::OutOfRange => "out_of_range",
265        Code::Unimplemented => "unimplemented",
266        Code::Internal => "internal",
267        Code::Unavailable => "unavailable",
268        Code::DataLoss => "data_loss",
269        Code::Unauthenticated => "unauthenticated",
270    }
271}
272
273#[cfg(test)]
274mod tests {
275    use super::*;
276    use actix_web::{body::to_bytes, test};
277    use futures::stream;
278    use serde::ser::{Error as _, Serializer};
279
280    struct Unserializable;
281
282    impl Serialize for Unserializable {
283        fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
284        where
285            S: Serializer,
286        {
287            Err(S::Error::custom("no representation"))
288        }
289    }
290
291    #[actix_rt::test]
292    async fn applies_context_aware_success_and_error_envelopes() {
293        let policy = ResponsePolicy::new()
294            .with_success_mapper(|request, data| {
295                json!({
296                    "request_id": request.headers().get("x-request-id").unwrap().to_str().unwrap(),
297                    "data": data,
298                })
299            })
300            .with_error_mapper(|request, _, code, message, details| {
301                json!({
302                    "request_id": request.headers().get("x-request-id").unwrap().to_str().unwrap(),
303                    "error": {"code": code, "message": message, "details": details},
304                })
305            });
306        let request = test::TestRequest::default()
307            .insert_header(("x-request-id", "req-42"))
308            .to_http_request();
309
310        let response = policy.ok(&request, json!({"name": "Ada"}));
311        assert_eq!(response.status(), StatusCode::OK);
312        let body: Value =
313            serde_json::from_slice(&to_bytes(response.into_body()).await.unwrap()).unwrap();
314        assert_eq!(
315            body,
316            json!({
317                "data": {"name": "Ada"},
318                "request_id": "req-42",
319            })
320        );
321
322        let error = ApiError::new(StatusCode::CONFLICT, "duplicate", "already exists")
323            .with_details(json!({"field": "email"}))
324            .unwrap();
325        let response = policy.error(&request, &error);
326        assert_eq!(response.status(), StatusCode::CONFLICT);
327        let body: Value =
328            serde_json::from_slice(&to_bytes(response.into_body()).await.unwrap()).unwrap();
329        assert_eq!(body["request_id"], "req-42");
330        assert_eq!(body["error"]["code"], "duplicate");
331        assert_eq!(body["error"]["details"]["field"], "email");
332    }
333
334    #[actix_rt::test]
335    async fn converts_serialization_failures_before_committing_success() {
336        let request = test::TestRequest::default().to_http_request();
337        let response = ResponsePolicy::new().ok(&request, Unserializable);
338
339        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
340        let body: Value =
341            serde_json::from_slice(&to_bytes(response.into_body()).await.unwrap()).unwrap();
342        assert_eq!(body["code"], "response_serialization_failed");
343    }
344
345    #[actix_rt::test]
346    async fn translates_grpc_statuses_through_the_error_policy() {
347        let request = test::TestRequest::default().to_http_request();
348        let response = ResponsePolicy::new().grpc_error(
349            &request,
350            &Status::new(Code::Unavailable, "users backend is unavailable"),
351        );
352
353        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
354        let body: Value =
355            serde_json::from_slice(&to_bytes(response.into_body()).await.unwrap()).unwrap();
356        assert_eq!(body["code"], "unavailable");
357        assert_eq!(body["message"], "users backend is unavailable");
358        assert_eq!(
359            grpc_status_to_http(Code::Cancelled),
360            StatusCode::REQUEST_TIMEOUT
361        );
362        assert_eq!(
363            grpc_status_to_http(Code::DeadlineExceeded),
364            StatusCode::GATEWAY_TIMEOUT
365        );
366    }
367
368    #[actix_rt::test]
369    async fn streams_each_flush_chunk_with_anti_buffering_headers() {
370        let response = streaming_response(
371            StatusCode::OK,
372            "application/x-ndjson",
373            stream::iter([
374                Ok::<_, actix_web::Error>(Bytes::from_static(b"{\"id\":1}\n")),
375                Ok::<_, actix_web::Error>(Bytes::from_static(b"{\"id\":2}\n")),
376            ]),
377        );
378
379        assert_eq!(
380            response.headers().get(header::CONTENT_TYPE).unwrap(),
381            "application/x-ndjson"
382        );
383        assert_eq!(response.headers().get("x-accel-buffering").unwrap(), "no");
384        assert_eq!(
385            to_bytes(response.into_body()).await.unwrap(),
386            Bytes::from_static(b"{\"id\":1}\n{\"id\":2}\n")
387        );
388    }
389}
rest/response.rs

rest/
response.rs
