simploxide_client/
ws.rs

1//! WebSocket backend that connects to a `simplex-chat` WebSocket server.
2//!
3//! Use [`BotBuilder`] to launch or connect to `simplex-chat` and get a ready-to-use [`Bot`].
4//! For lower-level access, [`connect`] and [`retry_connect`] return a [`Client`] and an
5//! [`EventStream`](crate::EventStream) directly.
6
7use std::sync::Arc;
8
9pub use simploxide_ws_core::{
10    self as core, Error as CoreError, Event as CoreEvent, Result as CoreResult, SimplexVersion,
11    VersionError, tungstenite::Error as WsError,
12};
13
14#[cfg(feature = "cli")]
15pub use simploxide_ws_core::cli;
16
17use serde::Deserialize;
18use simploxide_api_types::{
19    Preferences, Profile,
20    client_api::{ExtractResponse, WebSocketResponseShape, WebSocketResponseShapeInner},
21    events::{Event, EventKind},
22};
23use simploxide_core::{MAX_SUPPORTED_VERSION, MIN_SUPPORTED_VERSION};
24use simploxide_ws_core::RawClient;
25
26use crate::{
27    BadResponseError, ClientApi, ClientApiError, EventParser,
28    bot::{BotProfileSettings, BotSettings},
29    id::UserId,
30    preview::ImagePreview,
31    util,
32};
33
34#[cfg(not(feature = "xftp"))]
35pub type Bot = crate::bot::Bot<Client>;
36
37#[cfg(feature = "xftp")]
38pub type Bot = crate::bot::Bot<crate::xftp::XftpClient<Client>>;
39
40#[cfg(feature = "farm")]
41pub type FarmBot = crate::bot::farm::FarmBot<Client>;
42
43pub type EventStream = crate::EventStream<CoreResult<CoreEvent>>;
44pub type ClientResult<T = ()> = ::std::result::Result<T, ClientError>;
45
46/// Connects to a `simplex-chat` WebSocket server, returning a [`Client`] and an [`EventStream`]
47/// that handle serialization/deserialization of commands and events.
48///
49/// ```ignore
50/// let (client, mut events) = simploxide_client::ws::connect("ws://127.0.0.1:5225").await?;
51///
52/// let current_user = client.api_show_active_user().await?;
53/// println!("{current_user:#?}");
54///
55/// while let Some(ev) = events.try_next().await? {
56///     // Process events...
57/// }
58/// ```
59pub async fn connect<S: AsRef<str>>(uri: S) -> Result<(Client, EventStream), ConnectError> {
60    let (raw_client, raw_event_queue) = simploxide_ws_core::connect(uri.as_ref()).await?;
61
62    let version = raw_client
63        .version()
64        .await
65        .map_err(ConnectError::VersionError)?;
66
67    if !version.is_supported() {
68        return Err(ConnectError::VersionMismatch(version));
69    }
70
71    Ok((
72        Client::from(raw_client),
73        EventStream::from(raw_event_queue.into_receiver()),
74    ))
75}
76
77/// Like [`connect`] but retries to connect `retries_count` times before returning an error. This
78/// method is needed when you run simplex-cli programmatically and don't know when WebSocket port
79/// becomes available.
80///
81/// ```ignore
82/// let port = 5225;
83/// let cli = SimplexCli::spawn(port);
84/// let uri = format!("ws://127.0.0.1:{port}");
85///
86/// let (client, mut events) = simploxide_client::retry_connect(&uri, Duration::from_secs(1), 10).await?;
87///
88/// //...
89///
90/// ```
91pub async fn retry_connect<S: AsRef<str>>(
92    uri: S,
93    retry_delay: std::time::Duration,
94    mut retries_count: usize,
95) -> Result<(Client, EventStream), ConnectError> {
96    loop {
97        match connect(uri.as_ref()).await {
98            Ok(connection) => break Ok(connection),
99            Err(e) if !e.is_server() || retries_count == 0 => break Err(e),
100            Err(_) => {
101                retries_count -= 1;
102                tokio::time::sleep(retry_delay).await
103            }
104        }
105    }
106}
107
108impl EventParser for CoreResult<String> {
109    type Error = ClientError;
110
111    fn parse_kind(&self) -> Result<EventKind, Self::Error> {
112        match parse_data::<util::TypeField<'_>>(self) {
113            Ok(f) => Ok(EventKind::from_type_str(f.typ)),
114            Err(ClientError::BadResponse(BadResponseError::Undocumented(_))) => {
115                Ok(EventKind::Undocumented)
116            }
117            Err(e) => Err(e),
118        }
119    }
120
121    fn parse_user_id(&self) -> Result<Option<UserId>, Self::Error> {
122        match parse_data::<util::UserField>(self) {
123            Ok(f) => Ok(UserId::try_from(f.user.user_id).ok()),
124            Err(ClientError::BadResponse(_)) => Ok(None),
125            Err(e) => Err(e),
126        }
127    }
128
129    fn parse_event(&self) -> Result<Event, Self::Error> {
130        match parse_data(self) {
131            Ok(ev) => Ok(ev),
132            Err(ClientError::BadResponse(BadResponseError::Undocumented(json))) => {
133                Ok(Event::Undocumented(json))
134            }
135            Err(e) => Err(e),
136        }
137    }
138}
139
140fn parse_data<'de, 'r: 'de, D: 'de + Deserialize<'de>>(
141    res: &'r CoreResult<String>,
142) -> ClientResult<D> {
143    res.as_ref()
144        .map_err(|e| ClientError::WebSocketFailure(e.clone()))
145        .and_then(|ev| {
146            serde_json::from_str::<EventShape<D>>(ev)
147                .map_err(BadResponseError::InvalidJson)
148                .and_then(|shape| shape.extract_response())
149                .map_err(ClientError::BadResponse)
150        })
151}
152
153#[derive(Deserialize)]
154#[serde(untagged)]
155pub enum EventShape<T> {
156    ResponseShape(WebSocketResponseShape<T>),
157    InlineShape(WebSocketResponseShapeInner<T>),
158}
159
160impl<'de, T: 'de + Deserialize<'de>> ExtractResponse<'de, T> for EventShape<T> {
161    fn extract_response(self) -> Result<T, BadResponseError> {
162        match self {
163            Self::ResponseShape(resp) => resp.extract_response(),
164            Self::InlineShape(inline) => inline.extract_response(),
165        }
166    }
167}
168
169/// A high level SimpleX-Chat client which provides typed API methods with automatic command
170/// serialization and response deserialization.
171#[derive(Clone)]
172pub struct Client {
173    inner: RawClient,
174}
175
176impl From<RawClient> for Client {
177    fn from(inner: RawClient) -> Self {
178        Self { inner }
179    }
180}
181
182impl Client {
183    pub fn version(&self) -> impl Future<Output = Result<SimplexVersion, VersionError>> {
184        self.inner.version()
185    }
186
187    /// Initiates a graceful shutdown for the underlying web socket connection. See
188    /// [`simploxide_ws_core::RawClient::disconnect`] for details.
189    pub fn disconnect(self) -> impl Future<Output = ()> {
190        self.inner.disconnect()
191    }
192}
193
194impl ClientApi for Client {
195    type ResponseShape<'de, T>
196        = WebSocketResponseShape<T>
197    where
198        T: 'de + Deserialize<'de>;
199
200    type Error = ClientError;
201
202    async fn send_raw(&self, command: String) -> Result<String, Self::Error> {
203        self.inner
204            .send(command)
205            .await
206            .map_err(ClientError::WebSocketFailure)
207    }
208}
209
210/// See [`crate::client_api::AllowUndocumentedResponses`] if you don't want to trigger an error when
211/// you receive undocumeted responses(you usually receive undocumented responses when your
212/// simplex-chat server version is not compatible with the simploxide-client version. Keep an eye
213/// on the
214/// [Version compatability table](https://github.com/a1akris/simploxide?tab=readme-ov-file#version-compatability-table)
215/// )
216#[derive(Debug)]
217pub enum ClientError {
218    /// Critical error signalling that the web socket connection is dropped for some reason. You
219    /// will have to reconnect to the SimpleX server to recover from this one.
220    WebSocketFailure(CoreError),
221    /// SimpleX command error or unexpected(undocumented) response.
222    BadResponse(BadResponseError),
223}
224
225impl std::error::Error for ClientError {
226    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
227        match self {
228            Self::WebSocketFailure(error) => Some(error),
229            Self::BadResponse(error) => Some(error),
230        }
231    }
232}
233
234impl std::fmt::Display for ClientError {
235    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
236        match self {
237            ClientError::WebSocketFailure(err) => writeln!(f, "Web socket failure: {err}"),
238            ClientError::BadResponse(err) => err.fmt(f),
239        }
240    }
241}
242
243impl From<BadResponseError> for ClientError {
244    fn from(err: BadResponseError) -> Self {
245        Self::BadResponse(err)
246    }
247}
248
249impl ClientApiError for ClientError {
250    fn bad_response(&self) -> Option<&BadResponseError> {
251        if let Self::BadResponse(resp) = self {
252            Some(resp)
253        } else {
254            None
255        }
256    }
257
258    fn bad_response_mut(&mut self) -> Option<&mut BadResponseError> {
259        if let Self::BadResponse(resp) = self {
260            Some(resp)
261        } else {
262            None
263        }
264    }
265}
266
267#[derive(Debug)]
268pub enum ConnectError {
269    /// Failure to establish the connection to the server
270    Server(CoreError),
271    /// Failure to get the server version
272    VersionError(VersionError),
273    /// Unsupported server version
274    VersionMismatch(SimplexVersion),
275}
276
277impl ConnectError {
278    pub fn is_server(&self) -> bool {
279        matches!(self, Self::Server(_))
280    }
281
282    pub fn is_version_mismatch(&self) -> bool {
283        matches!(self, Self::VersionMismatch(_))
284    }
285}
286
287impl From<WsError> for ConnectError {
288    fn from(value: WsError) -> Self {
289        Self::Server(Arc::new(value))
290    }
291}
292
293impl std::fmt::Display for ConnectError {
294    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
295        match self {
296            Self::Server(error) => write!(f, "Cannot connect to the server: {error}"),
297            Self::VersionError(error) => write!(f, "Cannot get the server version: {error}"),
298            Self::VersionMismatch(v) => write!(
299                f,
300                "Version {v} is unsupported by the current client. Supported versions are {MIN_SUPPORTED_VERSION}..{MAX_SUPPORTED_VERSION}"
301            ),
302        }
303    }
304}
305
306impl std::error::Error for ConnectError {
307    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
308        match self {
309            Self::Server(error) => Some(error),
310            Self::VersionError(error) => Some(error),
311            Self::VersionMismatch(_) => None,
312        }
313    }
314}
315
316#[derive(Clone)]
317pub struct BotBuilder {
318    name: String,
319    port: u16,
320    retry_delay: std::time::Duration,
321    retries: usize,
322    auto_accept: Option<String>,
323    profile: Option<Profile>,
324    preferences: Option<Preferences>,
325    avatar: Option<ImagePreview>,
326    #[cfg(feature = "cli")]
327    db_prefix: String,
328    #[cfg(feature = "cli")]
329    db_key: Option<String>,
330    #[cfg(feature = "cli")]
331    extra_args: Vec<std::ffi::OsString>,
332}
333
334impl BotBuilder {
335    pub fn new(name: impl Into<String>, port: u16) -> Self {
336        Self {
337            name: name.into(),
338            port,
339            retry_delay: std::time::Duration::from_secs(1),
340            retries: 5,
341            auto_accept: None,
342            profile: None,
343            preferences: None,
344            avatar: None,
345            #[cfg(feature = "cli")]
346            db_prefix: "bot".into(),
347            #[cfg(feature = "cli")]
348            db_key: None,
349            #[cfg(feature = "cli")]
350            extra_args: Vec::new(),
351        }
352    }
353
354    #[cfg(feature = "cli")]
355    /// Path prefix for the SimpleX database
356    ///
357    /// "{dir}/{prefix}" creates a {dir} with `{prefix}_agent.db` and `{prefix}_chat.db`;
358    /// "{prefix}" creates `{prefix}_agent.db` and `{prefix}_chat.db` at the current dir
359    pub fn db_prefix(mut self, prefix: impl Into<String>) -> Self {
360        self.db_prefix = prefix.into();
361        self
362    }
363
364    #[cfg(feature = "cli")]
365    /// Database encryption key.
366    pub fn db_key(mut self, key: impl Into<String>) -> Self {
367        self.db_key = Some(key.into());
368        self
369    }
370
371    /// Delay between connection retry attempt. Default: 1s
372    pub fn connect_retry_delay(mut self, delay: std::time::Duration) -> Self {
373        self.retry_delay = delay;
374        self
375    }
376
377    /// Number of connection retry attempts. Default: 5
378    pub fn retries(mut self, n: usize) -> Self {
379        self.retries = n;
380        self
381    }
382
383    /// Create public address and auto accept users
384    pub fn auto_accept(mut self) -> Self {
385        self.auto_accept = Some(String::default());
386        self
387    }
388
389    /// Set a welcome message. This automatically creates a public address with enabled auto_accept
390    pub fn auto_accept_with(mut self, welcome_message: impl Into<String>) -> Self {
391        self.auto_accept = Some(welcome_message.into());
392        self
393    }
394
395    /// Set the bot avatar during initialisation
396    pub fn with_avatar(mut self, avatar: ImagePreview) -> Self {
397        self.avatar = Some(avatar);
398        self
399    }
400
401    /// Update/create the whole bot profile on launch
402    pub fn with_profile(mut self, profile: Profile) -> Self {
403        self.profile = Some(profile);
404        self
405    }
406
407    /// Apply these preferences to the bot's profile during initialisation.
408    pub fn with_preferences(mut self, prefs: Preferences) -> Self {
409        self.preferences = Some(prefs);
410        self
411    }
412
413    /// Pass extra arguments to the `simplex-chat` process.
414    #[cfg(feature = "cli")]
415    pub fn cli_args<I, S>(mut self, args: I) -> Self
416    where
417        I: IntoIterator<Item = S>,
418        S: Into<std::ffi::OsString>,
419    {
420        self.extra_args.extend(args.into_iter().map(|s| s.into()));
421        self
422    }
423
424    /// Connect to an already-running `simplex-chat` instance.
425    pub async fn connect(self) -> Result<(Bot, EventStream), BotInitError> {
426        let url = format!("ws://127.0.0.1:{}", self.port);
427
428        let (client, events) = retry_connect(url, self.retry_delay, self.retries)
429            .await
430            .map_err(BotInitError::Connect)?;
431
432        #[cfg(feature = "xftp")]
433        let (client, events) = events.hook_xftp(client);
434
435        let settings = BotSettings {
436            display_name: self.name,
437            auto_accept: self.auto_accept,
438            profile_settings: match (self.profile, self.preferences) {
439                (Some(mut profile), Some(preferences)) => {
440                    profile.preferences = Some(preferences);
441                    Some(BotProfileSettings::FullProfile(profile))
442                }
443                (Some(profile), None) => Some(BotProfileSettings::FullProfile(profile)),
444                (None, Some(preferences)) => Some(BotProfileSettings::Preferences(preferences)),
445                (None, None) => None,
446            },
447            avatar: self.avatar,
448        };
449
450        let bot = Bot::init(client, settings).await?;
451
452        let mut events = events;
453        events.set_owner(bot.user_id());
454
455        Ok((bot, events))
456    }
457
458    /// Spawn `simplex-chat`, then connect and initialise.
459    ///
460    /// Returns `(bot, events, cli)`. The caller is responsible for calling
461    /// [`cli::SimplexCli::kill`] after the bot finishes.
462    #[cfg(feature = "cli")]
463    pub async fn launch(mut self) -> Result<(Bot, EventStream, cli::SimplexCli), BotInitError> {
464        let mut builder = cli::SimplexCli::builder(&self.name, self.port)
465            .db_prefix(std::mem::take(&mut self.db_prefix));
466
467        if let Some(ref mut key) = self.db_key {
468            builder = builder.db_key(std::mem::take(key));
469        }
470
471        let cli = builder
472            .args(std::mem::take(&mut self.extra_args))
473            .spawn()
474            .await
475            .map_err(BotInitError::CliSpawn)?;
476
477        let (bot, events) = self.connect().await?;
478        Ok((bot, events, cli))
479    }
480}
481
482#[cfg(feature = "farm")]
483#[derive(Clone)]
484pub struct BotFarmBuilder {
485    name: String,
486    port: u16,
487    retry_delay: std::time::Duration,
488    retries: usize,
489    #[cfg(feature = "cli")]
490    db_prefix: String,
491    #[cfg(feature = "cli")]
492    db_key: Option<String>,
493    #[cfg(feature = "cli")]
494    extra_args: Vec<std::ffi::OsString>,
495}
496
497#[cfg(feature = "farm")]
498impl BotFarmBuilder {
499    pub fn new(name: impl Into<String>, port: u16) -> Self {
500        Self {
501            name: name.into(),
502            port,
503            retry_delay: std::time::Duration::from_secs(1),
504            retries: 5,
505            #[cfg(feature = "cli")]
506            db_prefix: "bot".into(),
507            #[cfg(feature = "cli")]
508            db_key: None,
509            #[cfg(feature = "cli")]
510            extra_args: Vec::new(),
511        }
512    }
513
514    #[cfg(feature = "cli")]
515    /// Path prefix for the SimpleX database
516    ///
517    /// "{dir}/{prefix}" creates a {dir} with `{prefix}_agent.db` and `{prefix}_chat.db`;
518    /// "{prefix}" creates `{prefix}_agent.db` and `{prefix}_chat.db` at the current dir
519    pub fn db_prefix(mut self, prefix: impl Into<String>) -> Self {
520        self.db_prefix = prefix.into();
521        self
522    }
523
524    #[cfg(feature = "cli")]
525    /// Database encryption key.
526    pub fn db_key(mut self, key: impl Into<String>) -> Self {
527        self.db_key = Some(key.into());
528        self
529    }
530
531    /// Delay between connection retry attempt. Default: 1s
532    pub fn connect_retry_delay(mut self, delay: std::time::Duration) -> Self {
533        self.retry_delay = delay;
534        self
535    }
536
537    /// Number of connection retry attempts. Default: 5
538    pub fn retries(mut self, n: usize) -> Self {
539        self.retries = n;
540        self
541    }
542
543    #[cfg(feature = "cli")]
544    /// Pass extra arguments to the `simplex-chat` process.
545    pub fn cli_args<I, S>(mut self, args: I) -> Self
546    where
547        I: IntoIterator<Item = S>,
548        S: Into<std::ffi::OsString>,
549    {
550        self.extra_args.extend(args.into_iter().map(|s| s.into()));
551        self
552    }
553
554    /// Connect to an already-running `simplex-chat` instance.
555    pub async fn connect(
556        self,
557    ) -> Result<
558        crate::bot::BotFarm<crate::bot::farm::Init<Client, CoreResult<CoreEvent>>>,
559        BotInitError,
560    > {
561        let url = format!("ws://127.0.0.1:{}", self.port);
562
563        let (client, events) = retry_connect(url, self.retry_delay, self.retries)
564            .await
565            .map_err(BotInitError::Connect)?;
566
567        let farm = crate::bot::BotFarm::init(self.name, client, events).await?;
568        Ok(farm)
569    }
570
571    #[cfg(feature = "cli")]
572    /// Spawn `simplex-chat`, then connect and initialise.
573    ///
574    /// Returns `(farm, cli)`. The caller is responsible for calling
575    /// [`cli::SimplexCli::kill`] after the farm finishes.
576    pub async fn launch(
577        mut self,
578    ) -> Result<
579        (
580            crate::bot::BotFarm<crate::bot::farm::Init<Client, CoreResult<CoreEvent>>>,
581            cli::SimplexCli,
582        ),
583        BotInitError,
584    > {
585        let mut builder = cli::SimplexCli::builder(&self.name, self.port)
586            .db_prefix(std::mem::take(&mut self.db_prefix));
587
588        if let Some(ref mut key) = self.db_key {
589            builder = builder.db_key(std::mem::take(key));
590        }
591
592        let cli = builder
593            .args(std::mem::take(&mut self.extra_args))
594            .spawn()
595            .await
596            .map_err(BotInitError::CliSpawn)?;
597
598        let farm = self.connect().await?;
599        Ok((farm, cli))
600    }
601}
602
603/// Error returned by [`BotBuilder::connect`] and [`BotBuilder::launch`].
604#[derive(Debug)]
605pub enum BotInitError {
606    Connect(ConnectError),
607    Api(ClientError),
608    #[cfg(feature = "cli")]
609    CliSpawn(std::io::Error),
610}
611
612impl std::fmt::Display for BotInitError {
613    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
614        match self {
615            #[cfg(feature = "cli")]
616            Self::CliSpawn(e) => write!(f, "failed to spawn simplex-chat: {e}"),
617            Self::Connect(e) => write!(f, "websocket connection failed: {e}"),
618            Self::Api(e) => write!(f, "SimpleX API error during init: {e}"),
619        }
620    }
621}
622
623impl std::error::Error for BotInitError {
624    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
625        match self {
626            #[cfg(feature = "cli")]
627            Self::CliSpawn(e) => Some(e),
628            Self::Connect(e) => Some(e),
629            Self::Api(e) => Some(e),
630        }
631    }
632}
633
634impl From<ClientError> for BotInitError {
635    fn from(e: ClientError) -> Self {
636        Self::Api(e)
637    }
638}
simploxide_client/ws.rs

simploxide_client/
ws.rs
