polyester/
transport.rs

1//! HTTP/Connect transport factory and API-key request signing.
2
3use crate::auth::{self, Credentials};
4use crate::errors::{Error, Result, map_connect_error};
5use crate::user_agent::user_agent;
6use buffa::Message;
7use connectrpc::ConnectError;
8use connectrpc::client::{CallOptions, ClientConfig, HttpClient};
9use connectrpc::rustls;
10use http::{HeaderValue, Uri, header::USER_AGENT};
11use serde::Serialize;
12use std::sync::Arc;
13use std::time::Duration;
14
15pub const DEFAULT_API_URL: &str = "https://api-devnet.polyester.ai";
16pub const DEFAULT_WS_URL: &str = "wss://api-devnet.polyester.ai";
17/// Maximum decompressed ConnectRPC response message accepted by the SDK.
18///
19/// This is set explicitly instead of relying on the transport dependency's
20/// default so catalog and other unary responses remain allocation-bounded
21/// across dependency upgrades.
22pub const MAX_CONNECT_RESPONSE_BYTES: usize = 4 * 1024 * 1024;
23
24/// Wire encoding for Connect unary calls.
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
26pub enum WireFormat {
27    #[default]
28    Binary,
29    Json,
30}
31
32impl WireFormat {
33    pub fn parse(value: &str) -> Self {
34        match value.trim().to_ascii_lowercase().as_str() {
35            "json" => Self::Json,
36            _ => Self::Binary,
37        }
38    }
39}
40
41/// Transport configuration.
42#[derive(Debug, Clone)]
43pub struct Config {
44    pub api_url: String,
45    pub ws_url: String,
46    pub timeout: Duration,
47    pub wire_format: WireFormat,
48}
49
50impl Default for Config {
51    fn default() -> Self {
52        Self {
53            api_url: DEFAULT_API_URL.to_owned(),
54            ws_url: DEFAULT_WS_URL.to_owned(),
55            timeout: Duration::from_secs(10),
56            wire_format: WireFormat::Binary,
57        }
58    }
59}
60
61/// Shared transport handle used by all generated Connect clients.
62pub type SharedTransport = HttpClient;
63
64/// Owns HTTP client, Connect config, and optional credentials.
65#[derive(Clone)]
66pub struct Factory {
67    pub config: Config,
68    pub credentials: Option<Credentials>,
69    transport: SharedTransport,
70    connect_config: ClientConfig,
71}
72
73impl Factory {
74    pub fn new(config: Config, credentials: Option<Credentials>) -> Result<Self> {
75        let uri: Uri = config
76            .api_url
77            .parse()
78            .map_err(|e| Error::validation(format!("invalid api_url: {e}")))?;
79
80        let transport = build_http_client(&config.api_url)?;
81
82        let ua = HeaderValue::from_str(&user_agent())
83            .map_err(|e| Error::validation(format!("invalid User-Agent header value: {e}")))?;
84        let mut connect_config = ClientConfig::new(uri)
85            .with_default_timeout(config.timeout)
86            .with_default_max_message_size(MAX_CONNECT_RESPONSE_BYTES)
87            .with_default_header(USER_AGENT, ua);
88
89        if config.wire_format == WireFormat::Json {
90            connect_config = connect_config.json();
91        }
92
93        Ok(Self {
94            config,
95            credentials,
96            transport,
97            connect_config,
98        })
99    }
100
101    pub(crate) fn transport(&self) -> SharedTransport {
102        self.transport.clone()
103    }
104
105    pub(crate) fn connect_config(&self) -> ClientConfig {
106        self.connect_config.clone()
107    }
108
109    pub fn require_credentials(&self) -> Result<&Credentials> {
110        self.credentials
111            .as_ref()
112            .ok_or_else(|| Error::auth("This endpoint requires Polyester API-key credentials"))
113    }
114
115    pub fn map_error(err: ConnectError) -> Error {
116        map_connect_error(err)
117    }
118
119    /// Build `CallOptions` with API-key signatures over the exact bytes that
120    /// Connect will send for the configured wire format.
121    pub fn sign_options<M: Message + Serialize>(
122        &self,
123        procedure: &str,
124        request: &M,
125    ) -> Result<CallOptions> {
126        let creds = self.require_credentials()?;
127        let body = match self.config.wire_format {
128            WireFormat::Binary => request.encode_to_bytes(),
129            WireFormat::Json => connectrpc::JsonCodec::encode(request).map_err(Self::map_error)?,
130        };
131        let sign_url = auth::request_url(&self.config.api_url, procedure);
132        let headers = creds.sign_request("POST", &sign_url, &body, None)?;
133        let mut opts = CallOptions::default().with_header(USER_AGENT, user_agent());
134        for (k, v) in headers {
135            opts = opts.with_header(k, v);
136        }
137        Ok(opts)
138    }
139
140    /// Async variant used by SDK network calls so timestamp-capacity
141    /// backpressure never blocks a Tokio worker thread.
142    pub async fn sign_options_async<M: Message + Serialize>(
143        &self,
144        procedure: &str,
145        request: &M,
146    ) -> Result<CallOptions> {
147        let creds = self.require_credentials()?;
148        let body = match self.config.wire_format {
149            WireFormat::Binary => request.encode_to_bytes(),
150            WireFormat::Json => connectrpc::JsonCodec::encode(request).map_err(Self::map_error)?,
151        };
152        let sign_url = auth::request_url(&self.config.api_url, procedure);
153        let headers = creds
154            .sign_request_async("POST", &sign_url, &body, None)
155            .await?;
156        let mut opts = CallOptions::default().with_header(USER_AGENT, user_agent());
157        for (k, v) in headers {
158            opts = opts.with_header(k, v);
159        }
160        Ok(opts)
161    }
162}
163
164fn build_http_client(api_url: &str) -> Result<HttpClient> {
165    static INIT: std::sync::Once = std::sync::Once::new();
166    INIT.call_once(|| {
167        let _ = rustls::crypto::ring::default_provider().install_default();
168    });
169
170    if api_url.starts_with("https://") {
171        let mut roots = rustls::RootCertStore::empty();
172        roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
173        let tls = Arc::new(
174            rustls::ClientConfig::builder()
175                .with_root_certificates(roots)
176                .with_no_client_auth(),
177        );
178        Ok(HttpClient::with_tls(tls))
179    } else if api_url.starts_with("http://") {
180        Ok(HttpClient::plaintext())
181    } else {
182        Err(Error::validation(
183            "api_url must start with http:// or https://",
184        ))
185    }
186}
187
188#[cfg(test)]
189mod tests {
190    use super::*;
191    use crate::user_agent::user_agent;
192
193    #[test]
194    fn connect_config_sets_polyester_user_agent() {
195        let factory = Factory::new(
196            Config {
197                api_url: "http://127.0.0.1:9".into(),
198                ..Default::default()
199            },
200            None,
201        )
202        .expect("factory");
203        let config = factory.connect_config();
204        let ua = config
205            .default_headers()
206            .get(USER_AGENT)
207            .expect("User-Agent default header")
208            .to_str()
209            .expect("ascii");
210        assert_eq!(ua, user_agent());
211    }
212}
polyester/transport.rs

polyester/
transport.rs
