cratestack-macros 0.7.2

Rust-native schema-first framework for typed HTTP APIs, generated clients, and backend services.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185

//! The two per-procedure `match` arms `service::build_service` splices
//! into `ApiServer::call` (ticket #208) — split out of `service.rs` to
//! stay under this repo's 200-LoC file convention. See `service.rs`'s
//! module doc for what "server-streaming" means for the streaming arm,
//! and its `build_create_arm` for the CRUD sibling this unary shape
//! mirrors almost exactly (decode -> encode -> dispatch -> bridge).

use cratestack_core::Procedure;
use quote::quote;

use crate::shared::{ident, to_snake_case};

use super::service::{request_prelude, status_from_bridge_error};

fn procedure_method_path(package: &str, procedure: &Procedure) -> String {
    let op_id = format!("procedure.{}", procedure.name);
    format!(
        "/{package}.Api/{}",
        cratestack_proto::op_id_to_method_name(&op_id)
    )
}

/// Unary procedure arm: decode the pb `<Base>Input`, convert it to
/// `procedures::<name>::Args` via the `TryFrom` impl
/// [`super::procedures::render_procedure_input`] generates, re-encode
/// through the schema's own wire codec, and call the exact same
/// `super::axum::handle_<name>_dispatch` fn REST/RPC already call.
pub(super) fn build_procedure_unary_arm(
    package: &str,
    procedure: &Procedure,
) -> proc_macro2::TokenStream {
    let path = procedure_method_path(package, procedure);
    let dispatch_ident = ident(&format!(
        "handle_{}_dispatch",
        to_snake_case(&procedure.name)
    ));
    let module_ident = ident(&to_snake_case(&procedure.name));
    let base = cratestack_proto::to_pascal_case(&procedure.name);
    let request_ty = ident(&format!("{base}Input"));
    let response_ty = ident(&format!("{base}Output"));
    let svc_ident = ident(&format!("Grpc{base}Svc"));
    let prelude = request_prelude(&path);
    let status = status_from_bridge_error(quote! { code }, quote! { message });

    quote! {
        #path => {
            struct #svc_ident<R, C, Auth>(super::axum::ProcedureRouterState<R, C, Auth>);
            impl<R, C, Auth> ::cratestack::grpc::tonic::server::UnaryService<pb::#request_ty> for #svc_ident<R, C, Auth>
            where
                R: super::procedures::ProcedureRegistry,
                C: ::cratestack::HttpTransport + Send + Sync + 'static,
                Auth: ::cratestack::AuthProvider + Send + Sync + 'static,
            {
                type Response = pb::#response_ty;
                type Future = ::cratestack::grpc::tonic::codegen::BoxFuture<
                    ::cratestack::grpc::tonic::Response<Self::Response>,
                    ::cratestack::grpc::tonic::Status,
                >;
                fn call(&mut self, request: ::cratestack::grpc::tonic::Request<pb::#request_ty>) -> Self::Future {
                    let state = self.0.clone();
                    Box::pin(async move {
                        #prelude
                        let args: ::core::result::Result<super::procedures::#module_ident::Args, ::cratestack::CoolError> =
                            super::procedures::#module_ident::Args::try_from(message);
                        let args = match args {
                            Ok(value) => value,
                            Err(error) => {
                                return Err(::cratestack::grpc::tonic::Status::new(
                                    ::cratestack::grpc::cool_error_code_to_tonic_code(error.code()),
                                    error.public_message().into_owned(),
                                ));
                            }
                        };
                        let body_bytes = match ::cratestack::__private::encode_rpc_value(&state.codec, &headers, &args).await {
                            Ok(bytes) => ::cratestack::axum::body::Bytes::from(bytes),
                            Err(error) => {
                                return Err(::cratestack::grpc::tonic::Status::new(
                                    ::cratestack::grpc::cool_error_code_to_tonic_code(error.code()),
                                    error.public_message().into_owned(),
                                ));
                            }
                        };
                        let response = super::axum::#dispatch_ident(state.clone(), canonical, headers.clone(), body_bytes).await;
                        let domain: super::procedures::#module_ident::Output = match ::cratestack::__private::bridge_grpc_response(response, &state.codec, &headers).await {
                            Ok(value) => value,
                            Err((code, message)) => return Err(#status),
                        };
                        Ok(::cratestack::grpc::tonic::Response::new(pb::#response_ty::from(&domain)))
                    })
                }
            }
            let svc = #svc_ident(state);
            let codec = ::cratestack::grpc::tonic::codec::ProstCodec::default();
            let mut grpc = ::cratestack::grpc::tonic::server::Grpc::new(codec);
            Box::pin(async move { Ok(grpc.unary(svc, req).await) })
        }
    }
}

/// Server-streaming procedure arm (`OpKind::Sequence` — `List`-arity
/// return): same decode/dispatch/bridge shape as
/// [`build_procedure_unary_arm`], but the bridged domain value is a
/// `Vec<Item>` (a `List`-arity procedure's `Output` type alias — see
/// `crate::procedure::types::procedure_type_tokens` — already resolves
/// to `Vec<Item>` directly, not a wrapper struct) and the one resulting
/// `<Base>Output { repeated result }` message travels back as a
/// single-item `tonic::server::ServerStreamingService` stream rather
/// than a unary response — see `service.rs`'s module doc for exactly
/// what that does and does not mean.
pub(super) fn build_procedure_stream_arm(
    package: &str,
    procedure: &Procedure,
) -> proc_macro2::TokenStream {
    let path = procedure_method_path(package, procedure);
    let dispatch_ident = ident(&format!(
        "handle_{}_dispatch",
        to_snake_case(&procedure.name)
    ));
    let module_ident = ident(&to_snake_case(&procedure.name));
    let base = cratestack_proto::to_pascal_case(&procedure.name);
    let request_ty = ident(&format!("{base}Input"));
    let response_ty = ident(&format!("{base}Output"));
    let svc_ident = ident(&format!("Grpc{base}StreamSvc"));
    let prelude = request_prelude(&path);
    let status = status_from_bridge_error(quote! { code }, quote! { message });

    quote! {
        #path => {
            struct #svc_ident<R, C, Auth>(super::axum::ProcedureRouterState<R, C, Auth>);
            impl<R, C, Auth> ::cratestack::grpc::tonic::server::ServerStreamingService<pb::#request_ty> for #svc_ident<R, C, Auth>
            where
                R: super::procedures::ProcedureRegistry,
                C: ::cratestack::HttpTransport + Send + Sync + 'static,
                Auth: ::cratestack::AuthProvider + Send + Sync + 'static,
            {
                type Response = pb::#response_ty;
                type ResponseStream = ::cratestack::grpc::tonic::codegen::BoxStream<Self::Response>;
                type Future = ::cratestack::grpc::tonic::codegen::BoxFuture<
                    ::cratestack::grpc::tonic::Response<Self::ResponseStream>,
                    ::cratestack::grpc::tonic::Status,
                >;
                fn call(&mut self, request: ::cratestack::grpc::tonic::Request<pb::#request_ty>) -> Self::Future {
                    let state = self.0.clone();
                    Box::pin(async move {
                        #prelude
                        let args: ::core::result::Result<super::procedures::#module_ident::Args, ::cratestack::CoolError> =
                            super::procedures::#module_ident::Args::try_from(message);
                        let args = match args {
                            Ok(value) => value,
                            Err(error) => {
                                return Err(::cratestack::grpc::tonic::Status::new(
                                    ::cratestack::grpc::cool_error_code_to_tonic_code(error.code()),
                                    error.public_message().into_owned(),
                                ));
                            }
                        };
                        let body_bytes = match ::cratestack::__private::encode_rpc_value(&state.codec, &headers, &args).await {
                            Ok(bytes) => ::cratestack::axum::body::Bytes::from(bytes),
                            Err(error) => {
                                return Err(::cratestack::grpc::tonic::Status::new(
                                    ::cratestack::grpc::cool_error_code_to_tonic_code(error.code()),
                                    error.public_message().into_owned(),
                                ));
                            }
                        };
                        let response = super::axum::#dispatch_ident(state.clone(), canonical, headers.clone(), body_bytes).await;
                        let items: super::procedures::#module_ident::Output = match ::cratestack::__private::bridge_grpc_response(response, &state.codec, &headers).await {
                            Ok(value) => value,
                            Err((code, message)) => return Err(#status),
                        };
                        let wire = pb::#response_ty::from(&items);
                        let stream: Self::ResponseStream = Box::pin(
                            ::cratestack::grpc::tonic::codegen::tokio_stream::once(Ok(wire)),
                        );
                        Ok(::cratestack::grpc::tonic::Response::new(stream))
                    })
                }
            }
            let svc = #svc_ident(state);
            let codec = ::cratestack::grpc::tonic::codec::ProstCodec::default();
            let mut grpc = ::cratestack::grpc::tonic::server::Grpc::new(codec);
            Box::pin(async move { Ok(grpc.server_streaming(svc, req).await) })
        }
    }
}