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

simploxide_client/
ws.rs
