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
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: String,
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    #[cfg(feature = "cli")]
332    db_prefix: String,
333    #[cfg(feature = "cli")]
334    db_key: Option<String>,
335    #[cfg(feature = "cli")]
336    extra_args: Vec<std::ffi::OsString>,
337}
338
339impl BotBuilder {
340    pub fn new(name: impl Into<String>, port: u16) -> Self {
341        Self {
342            name: name.into(),
343            port,
344            retry_delay: std::time::Duration::from_secs(1),
345            retries: 5,
346            auto_accept: None,
347            profile: None,
348            preferences: None,
349            avatar: None,
350            #[cfg(feature = "cli")]
351            db_prefix: "bot".into(),
352            #[cfg(feature = "cli")]
353            db_key: None,
354            #[cfg(feature = "cli")]
355            extra_args: Vec::new(),
356        }
357    }
358
359    #[cfg(feature = "cli")]
360    /// Path prefix for the SimpleX database
361    ///
362    /// "{dir}/{prefix}" creates a {dir} with `{prefix}_agent.db` and `{prefix}_chat.db`;
363    /// "{prefix}" creates `{prefix}_agent.db` and `{prefix}_chat.db` at the current dir
364    pub fn db_prefix(mut self, prefix: impl Into<String>) -> Self {
365        self.db_prefix = prefix.into();
366        self
367    }
368
369    #[cfg(feature = "cli")]
370    /// Database encryption key.
371    pub fn db_key(mut self, key: impl Into<String>) -> Self {
372        self.db_key = Some(key.into());
373        self
374    }
375
376    /// Delay between connection retry attempt. Default: 1s
377    pub fn connect_retry_delay(mut self, delay: std::time::Duration) -> Self {
378        self.retry_delay = delay;
379        self
380    }
381
382    /// Number of connection retry attempts. Default: 5
383    pub fn retries(mut self, n: usize) -> Self {
384        self.retries = n;
385        self
386    }
387
388    /// Create public address and auto accept users
389    pub fn auto_accept(mut self) -> Self {
390        self.auto_accept = Some(String::default());
391        self
392    }
393
394    /// Set a welcome message. This automatically creates a public address with enabled auto_accept
395    pub fn auto_accept_with(mut self, welcome_message: impl Into<String>) -> Self {
396        self.auto_accept = Some(welcome_message.into());
397        self
398    }
399
400    /// Set the bot avatar during initialisation
401    pub fn with_avatar(mut self, avatar: ImagePreview) -> Self {
402        self.avatar = Some(avatar);
403        self
404    }
405
406    /// Update/create the whole bot profile on launch
407    pub fn with_profile(mut self, profile: Profile) -> Self {
408        self.profile = Some(profile);
409        self
410    }
411
412    /// Apply these preferences to the bot's profile during initialisation.
413    pub fn with_preferences(mut self, prefs: Preferences) -> Self {
414        self.preferences = Some(prefs);
415        self
416    }
417
418    /// Pass extra arguments to the `simplex-chat` process.
419    #[cfg(feature = "cli")]
420    pub fn cli_args<I, S>(mut self, args: I) -> Self
421    where
422        I: IntoIterator<Item = S>,
423        S: Into<std::ffi::OsString>,
424    {
425        self.extra_args.extend(args.into_iter().map(|s| s.into()));
426        self
427    }
428
429    /// Connect to an already-running `simplex-chat` instance.
430    pub async fn connect(self) -> Result<(Bot, EventStream), BotInitError> {
431        let url = format!("ws://127.0.0.1:{}", self.port);
432
433        let (client, events) = retry_connect(url, self.retry_delay, self.retries)
434            .await
435            .map_err(BotInitError::Connect)?;
436
437        #[cfg(feature = "xftp")]
438        let (client, events) = events.hook_xftp(client);
439
440        let settings = BotSettings {
441            display_name: self.name,
442            auto_accept: self.auto_accept,
443            profile_settings: match (self.profile, self.preferences) {
444                (Some(mut profile), Some(preferences)) => {
445                    profile.preferences = Some(preferences);
446                    Some(BotProfileSettings::FullProfile(profile))
447                }
448                (Some(profile), None) => Some(BotProfileSettings::FullProfile(profile)),
449                (None, Some(preferences)) => Some(BotProfileSettings::Preferences(preferences)),
450                (None, None) => None,
451            },
452            avatar: self.avatar,
453        };
454
455        let bot = Bot::init(client, settings).await?;
456
457        let mut events = events;
458        events.set_owner(bot.user_id());
459
460        Ok((bot, events))
461    }
462
463    /// Spawn `simplex-chat`, then connect and initialise.
464    ///
465    /// Returns `(bot, events, cli)`. The caller is responsible for calling
466    /// [`cli::SimplexCli::kill`] after the bot finishes.
467    #[cfg(feature = "cli")]
468    pub async fn launch(mut self) -> Result<(Bot, EventStream, cli::SimplexCli), BotInitError> {
469        let mut builder = cli::SimplexCli::builder(&self.name, self.port)
470            .db_prefix(std::mem::take(&mut self.db_prefix));
471
472        if let Some(ref mut key) = self.db_key {
473            builder = builder.db_key(std::mem::take(key));
474        }
475
476        let cli = builder
477            .args(std::mem::take(&mut self.extra_args))
478            .spawn()
479            .await
480            .map_err(BotInitError::CliSpawn)?;
481
482        let (bot, events) = self.connect().await?;
483        Ok((bot, events, cli))
484    }
485}
486
487#[cfg(feature = "farm")]
488#[derive(Clone)]
489pub struct BotFarmBuilder {
490    name: String,
491    port: u16,
492    retry_delay: std::time::Duration,
493    retries: usize,
494    #[cfg(feature = "cli")]
495    db_prefix: String,
496    #[cfg(feature = "cli")]
497    db_key: Option<String>,
498    #[cfg(feature = "cli")]
499    extra_args: Vec<std::ffi::OsString>,
500}
501
502#[cfg(feature = "farm")]
503impl BotFarmBuilder {
504    pub fn new(name: impl Into<String>, port: u16) -> Self {
505        Self {
506            name: name.into(),
507            port,
508            retry_delay: std::time::Duration::from_secs(1),
509            retries: 5,
510            #[cfg(feature = "cli")]
511            db_prefix: "bot".into(),
512            #[cfg(feature = "cli")]
513            db_key: None,
514            #[cfg(feature = "cli")]
515            extra_args: Vec::new(),
516        }
517    }
518
519    #[cfg(feature = "cli")]
520    /// Path prefix for the SimpleX database
521    ///
522    /// "{dir}/{prefix}" creates a {dir} with `{prefix}_agent.db` and `{prefix}_chat.db`;
523    /// "{prefix}" creates `{prefix}_agent.db` and `{prefix}_chat.db` at the current dir
524    pub fn db_prefix(mut self, prefix: impl Into<String>) -> Self {
525        self.db_prefix = prefix.into();
526        self
527    }
528
529    #[cfg(feature = "cli")]
530    /// Database encryption key.
531    pub fn db_key(mut self, key: impl Into<String>) -> Self {
532        self.db_key = Some(key.into());
533        self
534    }
535
536    /// Delay between connection retry attempt. Default: 1s
537    pub fn connect_retry_delay(mut self, delay: std::time::Duration) -> Self {
538        self.retry_delay = delay;
539        self
540    }
541
542    /// Number of connection retry attempts. Default: 5
543    pub fn retries(mut self, n: usize) -> Self {
544        self.retries = n;
545        self
546    }
547
548    #[cfg(feature = "cli")]
549    /// Pass extra arguments to the `simplex-chat` process.
550    pub fn cli_args<I, S>(mut self, args: I) -> Self
551    where
552        I: IntoIterator<Item = S>,
553        S: Into<std::ffi::OsString>,
554    {
555        self.extra_args.extend(args.into_iter().map(|s| s.into()));
556        self
557    }
558
559    /// Connect to an already-running `simplex-chat` instance.
560    pub async fn connect(self) -> Result<InitFarm, BotInitError> {
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(mut self) -> Result<(InitFarm, cli::SimplexCli), BotInitError> {
577        let mut builder = cli::SimplexCli::builder(&self.name, self.port)
578            .db_prefix(std::mem::take(&mut self.db_prefix));
579
580        if let Some(ref mut key) = self.db_key {
581            builder = builder.db_key(std::mem::take(key));
582        }
583
584        let cli = builder
585            .args(std::mem::take(&mut self.extra_args))
586            .spawn()
587            .await
588            .map_err(BotInitError::CliSpawn)?;
589
590        let farm = self.connect().await?;
591        Ok((farm, cli))
592    }
593}
594
595/// Error returned by [`BotBuilder::connect`] and [`BotBuilder::launch`].
596#[derive(Debug)]
597pub enum BotInitError {
598    Connect(ConnectError),
599    Api(ClientError),
600    #[cfg(feature = "cli")]
601    CliSpawn(std::io::Error),
602}
603
604impl std::fmt::Display for BotInitError {
605    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
606        match self {
607            #[cfg(feature = "cli")]
608            Self::CliSpawn(e) => write!(f, "failed to spawn simplex-chat: {e}"),
609            Self::Connect(e) => write!(f, "websocket connection failed: {e}"),
610            Self::Api(e) => write!(f, "SimpleX API error during init: {e}"),
611        }
612    }
613}
614
615impl std::error::Error for BotInitError {
616    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
617        match self {
618            #[cfg(feature = "cli")]
619            Self::CliSpawn(e) => Some(e),
620            Self::Connect(e) => Some(e),
621            Self::Api(e) => Some(e),
622        }
623    }
624}
625
626impl From<ClientError> for BotInitError {
627    fn from(e: ClientError) -> Self {
628        Self::Api(e)
629    }
630}
simploxide_client/ws.rs

simploxide_client/
ws.rs
