aion_client/
client.rs

1//! `Client` and `ClientBuilder` connection, auth, and TLS support.
2
3use std::collections::HashMap;
4use std::fmt;
5use std::sync::Arc;
6
7use tokio::sync::Mutex;
8
9use crate::error::ClientError;
10use crate::start::{CachedStart, StartFingerprint};
11use crate::transport::{GrpcWorkflowTransport, WorkflowTransport};
12
13/// Reusable caller-side SDK client for an `aion-server` deployment.
14///
15/// # Examples
16///
17/// ```no_run
18/// # async fn connect() -> Result<(), aion_client::ClientError> {
19/// use aion_client::{ClientAuth, ClientBuilder};
20///
21/// let client = ClientBuilder::new("https://aion.example.com")
22///     .with_auth(ClientAuth::bearer("secret-token"))
23///     .with_namespace("tenant-a")
24///     .build()
25///     .await?;
26///
27/// let shared = client.clone();
28/// # let _ = shared;
29/// # Ok(())
30/// # }
31/// ```
32#[derive(Clone)]
33pub struct Client {
34    pub(crate) transport: Arc<dyn WorkflowTransport>,
35    pub(crate) config: ClientConfig,
36    idempotent_starts: Arc<Mutex<HashMap<String, CachedStart>>>,
37}
38
39impl Client {
40    /// Creates a builder for an `aion-server` endpoint.
41    #[must_use]
42    pub fn builder(endpoint: impl Into<String>) -> ClientBuilder {
43        ClientBuilder::new(endpoint)
44    }
45
46    pub(crate) fn from_transport(
47        config: ClientConfig,
48        transport: Arc<dyn WorkflowTransport>,
49    ) -> Self {
50        Self {
51            transport,
52            config,
53            idempotent_starts: Arc::new(Mutex::new(HashMap::new())),
54        }
55    }
56
57    #[cfg(feature = "embedded")]
58    /// Creates a client backed by an in-process embedded engine.
59    #[must_use]
60    pub fn embedded(engine: Arc<aion::Engine>) -> Self {
61        let config = ClientConfig {
62            endpoint: String::from("embedded://engine"),
63            stream_endpoint: None,
64            auth: None,
65            tls: None,
66            namespace: String::from("default"),
67            subject: None,
68            authorized_namespaces: Vec::new(),
69        };
70        Self::from_transport(
71            config,
72            Arc::new(crate::transport::EmbeddedWorkflowTransport::new(engine)),
73        )
74    }
75
76    pub(crate) fn namespace(&self) -> &str {
77        &self.config.namespace
78    }
79
80    /// Looks up a cached start for `fingerprint`'s idempotency key.
81    ///
82    /// Returns the whole cached entry — handle AND the display name the cached
83    /// start requested — because the caller has to compare that standing name
84    /// against the one this call asked for (#211): the name is deliberately
85    /// outside the fingerprint, so a replay can match while carrying a
86    /// different, unapplied name that must be reported rather than dropped.
87    pub(crate) async fn cached_start(
88        &self,
89        fingerprint: &StartFingerprint,
90    ) -> Result<Option<CachedStart>, ClientError> {
91        let cache = self.idempotent_starts.lock().await;
92        let Some(cached) = cache.get(fingerprint.key()) else {
93            return Ok(None);
94        };
95        if cached.fingerprint() == fingerprint {
96            Ok(Some(cached.clone()))
97        } else {
98            Err(idempotency_conflict())
99        }
100    }
101
102    pub(crate) async fn record_start(&self, cached: CachedStart) -> Result<(), ClientError> {
103        let mut cache = self.idempotent_starts.lock().await;
104        let key = cached.fingerprint().key().to_owned();
105        match cache.get(&key) {
106            Some(existing) if existing.fingerprint() == cached.fingerprint() => Ok(()),
107            Some(_) => Err(idempotency_conflict()),
108            None => {
109                cache.insert(key, cached);
110                Ok(())
111            }
112        }
113    }
114}
115
116/// The SDK-boundary idempotency conflict: the same key was reused with a
117/// different start request.
118fn idempotency_conflict() -> ClientError {
119    ClientError::already_exists(
120        "idempotency key was already used by a different start request \
121         (namespace, workflow type, or input differ)",
122    )
123}
124
125/// Builder for [`Client`] connection, authentication, and TLS options.
126#[derive(Clone, Debug)]
127pub struct ClientBuilder {
128    endpoint: String,
129    stream_endpoint: Option<String>,
130    auth: Option<ClientAuth>,
131    tls: Option<TlsOptions>,
132    namespace: String,
133    subject: Option<String>,
134    authorized_namespaces: Vec<String>,
135}
136
137impl ClientBuilder {
138    /// Creates a builder for the supplied server endpoint.
139    #[must_use]
140    pub fn new(endpoint: impl Into<String>) -> Self {
141        Self {
142            endpoint: endpoint.into(),
143            stream_endpoint: None,
144            auth: None,
145            tls: None,
146            namespace: String::from("default"),
147            subject: None,
148            authorized_namespaces: Vec::new(),
149        }
150    }
151
152    /// Configures the WebSocket event-stream endpoint used by subscribe
153    /// operations: the full URL of the server's `/events/stream` route, e.g.
154    /// `ws://127.0.0.1:8080/events/stream` (`http`/`https` URLs are accepted
155    /// and protocol-mapped to `ws`/`wss`).
156    ///
157    /// There is no default and nothing is derived: the gRPC endpoint and the
158    /// HTTP/WebSocket listener are separate addresses. Subscribing without
159    /// this option returns [`ClientError::InvalidArgument`] with a precise
160    /// message.
161    #[must_use]
162    pub fn with_stream_endpoint(mut self, stream_endpoint: impl Into<String>) -> Self {
163        self.stream_endpoint = Some(stream_endpoint.into());
164        self
165    }
166
167    /// Configures the credential attached to every request.
168    #[must_use]
169    pub fn with_auth(mut self, auth: ClientAuth) -> Self {
170        self.auth = Some(auth);
171        self
172    }
173
174    /// Configures TLS options for the tonic channel.
175    #[must_use]
176    pub fn with_tls(mut self, tls: TlsOptions) -> Self {
177        self.tls = Some(tls);
178        self
179    }
180
181    /// Configures the namespace used by operations unless an operation option overrides it.
182    #[must_use]
183    pub fn with_namespace(mut self, namespace: impl Into<String>) -> Self {
184        self.namespace = namespace.into();
185        self
186    }
187
188    /// Configures the caller subject metadata sent to the server.
189    #[must_use]
190    pub fn with_subject(mut self, subject: impl Into<String>) -> Self {
191        self.subject = Some(subject.into());
192        self
193    }
194
195    /// Configures the namespaces advertised in auth metadata.
196    #[must_use]
197    pub fn with_authorized_namespaces<I, S>(mut self, namespaces: I) -> Self
198    where
199        I: IntoIterator<Item = S>,
200        S: Into<String>,
201    {
202        self.authorized_namespaces = namespaces.into_iter().map(Into::into).collect();
203        self
204    }
205
206    /// Connects once and returns a cheaply cloneable [`Client`].
207    ///
208    /// # Errors
209    ///
210    /// Returns [`ClientError::Unavailable`] for malformed endpoints and failed
211    /// channel/TLS handshakes. Server-side credential rejection is surfaced as
212    /// [`ClientError::Unauthenticated`] when AW returns gRPC `Unauthenticated`.
213    pub async fn build(self) -> Result<Client, ClientError> {
214        let config = ClientConfig::from(self);
215        let transport = GrpcWorkflowTransport::connect(config.clone()).await?;
216        Ok(Client::from_transport(config, Arc::new(transport)))
217    }
218}
219
220/// Bearer authentication credential for server calls.
221#[derive(Clone, PartialEq, Eq)]
222pub struct ClientAuth {
223    token: String,
224}
225
226impl ClientAuth {
227    /// Creates a bearer-token credential.
228    #[must_use]
229    pub fn bearer(token: impl Into<String>) -> Self {
230        Self {
231            token: token.into(),
232        }
233    }
234
235    pub(crate) fn token(&self) -> &str {
236        &self.token
237    }
238}
239
240impl fmt::Debug for ClientAuth {
241    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
242        formatter
243            .debug_struct("ClientAuth")
244            .field("token", &"<redacted>")
245            .finish()
246    }
247}
248
249/// TLS options for connecting to an HTTPS/TLS endpoint.
250#[derive(Clone, Debug, Default, PartialEq, Eq)]
251pub struct TlsOptions {
252    pub(crate) domain_name: Option<String>,
253    pub(crate) ca_certificate_pem: Option<Vec<u8>>,
254}
255
256impl TlsOptions {
257    /// Creates empty TLS options using platform/webpki roots.
258    #[must_use]
259    pub fn new() -> Self {
260        Self::default()
261    }
262
263    /// Overrides the TLS domain name checked during the gRPC channel
264    /// handshake. The WebSocket event stream always verifies against its own
265    /// stream-endpoint host (`ClientBuilder::with_stream_endpoint`), so no
266    /// override is needed there: point the stream endpoint at the name the
267    /// server's certificate carries.
268    #[must_use]
269    pub fn with_domain_name(mut self, domain_name: impl Into<String>) -> Self {
270        self.domain_name = Some(domain_name.into());
271        self
272    }
273
274    /// Adds a PEM-encoded CA certificate trusted by BOTH transports: the
275    /// tonic gRPC channel and the `wss://` WebSocket event stream.
276    #[must_use]
277    pub fn with_ca_certificate_pem(mut self, ca_certificate_pem: impl Into<Vec<u8>>) -> Self {
278        self.ca_certificate_pem = Some(ca_certificate_pem.into());
279        self
280    }
281}
282
283/// Fully resolved client connection configuration.
284#[derive(Clone, Debug, PartialEq, Eq)]
285pub struct ClientConfig {
286    pub(crate) endpoint: String,
287    pub(crate) stream_endpoint: Option<String>,
288    pub(crate) auth: Option<ClientAuth>,
289    pub(crate) tls: Option<TlsOptions>,
290    pub(crate) namespace: String,
291    pub(crate) subject: Option<String>,
292    pub(crate) authorized_namespaces: Vec<String>,
293}
294
295impl From<ClientBuilder> for ClientConfig {
296    fn from(builder: ClientBuilder) -> Self {
297        Self {
298            endpoint: builder.endpoint,
299            stream_endpoint: builder.stream_endpoint,
300            auth: builder.auth,
301            tls: builder.tls,
302            namespace: builder.namespace,
303            subject: builder.subject,
304            authorized_namespaces: builder.authorized_namespaces,
305        }
306    }
307}
308
309#[cfg(test)]
310mod tests {
311    use super::{Client, ClientAuth, ClientBuilder, ClientConfig, TlsOptions};
312
313    fn assert_send_sync<T: Send + Sync>() {}
314
315    #[test]
316    fn client_is_clone_send_sync() {
317        assert_send_sync::<Client>();
318    }
319
320    #[test]
321    fn auth_debug_redacts_token() {
322        let auth = ClientAuth::bearer("secret-token");
323        assert_eq!(format!("{auth:?}"), "ClientAuth { token: \"<redacted>\" }");
324    }
325
326    #[test]
327    fn builder_captures_connection_options() {
328        let config = ClientConfig::from(
329            ClientBuilder::new("https://aion.example.com")
330                .with_stream_endpoint("wss://aion-http.example.com/events/stream")
331                .with_auth(ClientAuth::bearer("secret-token"))
332                .with_tls(TlsOptions::new().with_domain_name("aion.example.com"))
333                .with_namespace("tenant-a")
334                .with_subject("alice")
335                .with_authorized_namespaces(["tenant-a", "tenant-b"]),
336        );
337
338        assert_eq!(config.endpoint, "https://aion.example.com");
339        assert_eq!(
340            config.stream_endpoint,
341            Some(String::from("wss://aion-http.example.com/events/stream"))
342        );
343        assert!(config.auth.is_some());
344        assert!(config.tls.is_some());
345        assert_eq!(config.namespace, "tenant-a");
346        assert_eq!(config.subject, Some(String::from("alice")));
347        assert_eq!(
348            config.authorized_namespaces,
349            vec![String::from("tenant-a"), String::from("tenant-b")]
350        );
351    }
352
353    #[test]
354    fn stream_endpoint_has_no_default() {
355        let config = ClientConfig::from(ClientBuilder::new("https://aion.example.com"));
356        assert_eq!(config.stream_endpoint, None);
357    }
358}
aion_client/client.rs

aion_client/
client.rs
