1
#[cfg(feature = "test-util")]
2
use std::sync::atomic::AtomicBool;
3
use std::{
4
    any::TypeId,
5
    collections::HashMap,
6
    fmt::Debug,
7
    ops::Deref,
8
    sync::{
9
        atomic::{AtomicU32, Ordering},
10
        Arc,
11
    },
12
};
13

            
14
use async_trait::async_trait;
15
#[cfg(feature = "password-hashing")]
16
use bonsaidb_core::connection::Authentication;
17
use bonsaidb_core::{
18
    admin::{Admin, ADMIN_DATABASE_NAME},
19
    api::{self, Api},
20
    arc_bytes::{serde::Bytes, OwnedBytes},
21
    connection::{AsyncStorageConnection, Database, HasSession, IdentityReference, Session},
22
    networking::{
23
        AlterUserPermissionGroupMembership, AlterUserRoleMembership, AssumeIdentity,
24
        CreateDatabase, CreateUser, DeleteDatabase, DeleteUser, ListAvailableSchemas,
25
        ListDatabases, LogOutSession, MessageReceived, Payload, UnregisterSubscriber,
26
        CURRENT_PROTOCOL_VERSION,
27
    },
28
    permissions::Permissions,
29
    schema::{ApiName, Nameable, Schema, SchemaName, Schematic},
30
};
31
use bonsaidb_utils::fast_async_lock;
32
use flume::Sender;
33
use futures::{future::BoxFuture, Future, FutureExt};
34
use parking_lot::Mutex;
35
#[cfg(not(target_arch = "wasm32"))]
36
use tokio::{runtime::Handle, task::JoinHandle};
37
use url::Url;
38

            
39
pub use self::remote_database::{RemoteDatabase, RemoteSubscriber};
40
use crate::{error::Error, ApiError, Builder};
41

            
42
#[cfg(not(target_arch = "wasm32"))]
43
mod quic_worker;
44
mod remote_database;
45
#[cfg(not(target_arch = "wasm32"))]
46
mod sync;
47
#[cfg(all(feature = "websockets", not(target_arch = "wasm32")))]
48
mod tungstenite_worker;
49
#[cfg(all(feature = "websockets", target_arch = "wasm32"))]
50
mod wasm_websocket_worker;
51

            
52
6517
#[derive(Debug, Clone, Default)]
53
pub struct SubscriberMap(Arc<Mutex<HashMap<u64, flume::Sender<Message>>>>);
54

            
55
impl SubscriberMap {
56
2964
    pub fn clear(&self) {
57
2964
        let mut data = self.lock();
58
2964
        data.clear();
59
2964
    }
60
}
61

            
62
impl Deref for SubscriberMap {
63
    type Target = Mutex<HashMap<u64, flume::Sender<Message>>>;
64

            
65
4465
    fn deref(&self) -> &Self::Target {
66
4465
        &self.0
67
4465
    }
68
}
69

            
70
use bonsaidb_core::circulate::Message;
71

            
72
#[cfg(all(feature = "websockets", not(target_arch = "wasm32")))]
73
pub type WebSocketError = tokio_tungstenite::tungstenite::Error;
74

            
75
#[cfg(all(feature = "websockets", target_arch = "wasm32"))]
76
pub type WebSocketError = wasm_websocket_worker::WebSocketError;
77

            
78
/// Client for connecting to a BonsaiDb server.
79
///
80
///
81
///
82
/// ## Connecting via QUIC
83
///
84
/// The URL scheme to connect via QUIC is `bonsaidb`. If no port is specified,
85
/// port 5645 is assumed.
86
///
87
/// ### With a valid TLS certificate
88
///
89
/// ```rust
90
/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};
91
/// # async fn test_fn() -> anyhow::Result<()> {
92
/// let client = Client::build(Url::parse("bonsaidb://my-server.com")?).finish()?;
93
/// # Ok(())
94
/// # }
95
/// ```
96
///
97
/// ### With a Self-Signed Pinned Certificate
98
///
99
/// When using `install_self_signed_certificate()`, clients will need the
100
/// contents of the `pinned-certificate.der` file within the database. It can be
101
/// specified when building the client:
102
///
103
/// ```rust
104
/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};
105
/// # async fn test_fn() -> anyhow::Result<()> {
106
/// let certificate =
107
///     Certificate::from_der(std::fs::read("mydb.bonsaidb/pinned-certificate.der")?)?;
108
/// let client = Client::build(Url::parse("bonsaidb://localhost")?)
109
///     .with_certificate(certificate)
110
///     .finish()?;
111
/// # Ok(())
112
/// # }
113
/// ```
114
///
115
/// ## Connecting via WebSockets
116
///
117
/// WebSockets are built atop the HTTP protocol. There are two URL schemes for
118
/// WebSockets:
119
///
120
/// - `ws`: Insecure WebSockets. Port 80 is assumed if no port is specified.
121
/// - `wss`: Secure WebSockets. Port 443 is assumed if no port is specified.
122
///
123
/// ### Without TLS
124
///
125
/// ```rust
126
/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};
127
/// # async fn test_fn() -> anyhow::Result<()> {
128
/// let client = Client::build(Url::parse("ws://localhost")?).finish()?;
129
/// # Ok(())
130
/// # }
131
/// ```
132
///
133
/// ### With TLS
134
///
135
/// ```rust
136
/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};
137
/// # async fn test_fn() -> anyhow::Result<()> {
138
/// let client = Client::build(Url::parse("wss://my-server.com")?).finish()?;
139
/// # Ok(())
140
/// # }
141
/// ```
142
///
143
/// ## Using a `Api`
144
///
145
/// Our user guide has a [section on creating and
146
/// using](https://dev.bonsaidb.io/release/guide/about/access-models/custom-api-server.html)
147
/// an [`Api`](api::Api).
148
///
149
/// ```rust
150
/// # use bonsaidb_client::{Client, fabruic::Certificate, url::Url};
151
/// // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_client::core`.
152
/// use bonsaidb_core::{
153
///     api::{Api, Infallible},
154
///     schema::{ApiName, Qualified},
155
/// };
156
/// use serde::{Deserialize, Serialize};
157
///
158
/// #[derive(Serialize, Deserialize, Debug)]
159
/// pub struct Ping;
160
///
161
/// #[derive(Serialize, Deserialize, Clone, Debug)]
162
/// pub struct Pong;
163
///
164
/// impl Api for Ping {
165
///     type Response = Pong;
166
///     type Error = Infallible;
167
///
168
///     fn name() -> ApiName {
169
///         ApiName::private("ping")
170
///     }
171
/// }
172
///
173
/// # async fn test_fn() -> anyhow::Result<()> {
174
/// let client = Client::build(Url::parse("bonsaidb://localhost")?).finish()?;
175
/// let Pong = client.send_api_request_async(&Ping).await?;
176
/// # Ok(())
177
/// # }
178
/// ```
179
///
180
/// ### Receiving out-of-band messages from the server
181
///
182
/// If the server sends a message that isn't in response to a request, the
183
/// client will invoke it's [api callback](Builder::with_api_callback):
184
///
185
/// ```rust
186
/// # use bonsaidb_client::{Client, ApiCallback, fabruic::Certificate, url::Url};
187
/// # // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_client::core`.
188
/// # use bonsaidb_core::{api::{Api, Infallible}, schema::{ApiName, Qualified}};
189
/// # use serde::{Serialize, Deserialize};
190
/// # #[derive(Serialize, Deserialize, Debug)]
191
/// # pub struct Ping;
192
/// # #[derive(Serialize, Deserialize, Clone, Debug)]
193
/// # pub struct Pong;
194
/// # impl Api for Ping {
195
/// #     type Response = Pong;
196
/// #     type Error = Infallible;
197
/// #
198
/// #     fn name() -> ApiName {
199
/// #         ApiName::private("ping")
200
/// #     }
201
/// # }
202
/// # async fn test_fn() -> anyhow::Result<()> {
203
/// let client = Client::build(Url::parse("bonsaidb://localhost")?)
204
///     .with_api_callback(ApiCallback::<Ping>::new(|result: Pong| async move {
205
///         println!("Received out-of-band Pong");
206
///     }))
207
///     .finish()?;
208
/// # Ok(())
209
/// # }
210
/// ```
211
2471
#[derive(Debug, Clone)]
212
pub struct Client {
213
    pub(crate) data: Arc<Data>,
214
    session: Arc<Session>,
215
}
216

            
217
impl Drop for Client {
218
35188
    fn drop(&mut self) {
219
35188
        if Arc::strong_count(&self.session) == 1 {
220
2945
            if let Some(session_id) = self.session.id {
221
133
                // Final reference to an authenticated session
222
133
                drop(self.invoke_api_request(&LogOutSession(session_id)));
223
2812
            }
224
32243
        }
225
35188
    }
226
}
227

            
228
impl PartialEq for Client {
229
    fn eq(&self, other: &Self) -> bool {
230
        Arc::ptr_eq(&self.data, &other.data)
231
    }
232
}
233

            
234
#[derive(Debug)]
235
pub struct Data {
236
    request_sender: Sender<PendingRequest>,
237
    #[cfg(not(target_arch = "wasm32"))]
238
    _worker: CancellableHandle<Result<(), Error>>,
239
    effective_permissions: Mutex<Option<Permissions>>,
240
    schemas: Mutex<HashMap<TypeId, Arc<Schematic>>>,
241
    request_id: AtomicU32,
242
    subscribers: SubscriberMap,
243
    #[cfg(feature = "test-util")]
244
    background_task_running: Arc<AtomicBool>,
245
}
246

            
247
impl Client {
248
    /// Returns a builder for a new client connecting to `url`.
249
1634
    pub fn build(url: Url) -> Builder {
250
1634
        Builder::new(url)
251
1634
    }
252
}
253

            
254
impl Client {
255
    /// Initialize a client connecting to `url`. This client can be shared by
256
    /// cloning it. All requests are done asynchronously over the same
257
    /// connection.
258
    ///
259
    /// If the client has an error connecting, the first request made will
260
    /// present that error. If the client disconnects while processing requests,
261
    /// all requests being processed will exit and return
262
    /// [`Error::Disconnected`]. The client will automatically try reconnecting.
263
    ///
264
    /// The goal of this design of this reconnection strategy is to make it
265
    /// easier to build resilliant apps. By allowing existing Client instances
266
    /// to recover and reconnect, each component of the apps built can adopt a
267
    /// "retry-to-recover" design, or "abort-and-fail" depending on how critical
268
    /// the database is to operation.
269
1197
    pub fn new(url: Url) -> Result<Self, Error> {
270
1197
        Self::new_from_parts(
271
1197
            url,
272
1197
            CURRENT_PROTOCOL_VERSION,
273
1197
            HashMap::default(),
274
1197
            #[cfg(not(target_arch = "wasm32"))]
275
1197
            None,
276
1197
            #[cfg(not(target_arch = "wasm32"))]
277
1197
            Handle::try_current().ok(),
278
1197
        )
279
1197
    }
280

            
281
    /// Initialize a client connecting to `url` with `certificate` being used to
282
    /// validate and encrypt the connection. This client can be shared by
283
    /// cloning it. All requests are done asynchronously over the same
284
    /// connection.
285
    ///
286
    /// If the client has an error connecting, the first request made will
287
    /// present that error. If the client disconnects while processing requests,
288
    /// all requests being processed will exit and return
289
    /// [`Error::Disconnected`]. The client will automatically try reconnecting.
290
    ///
291
    /// The goal of this design of this reconnection strategy is to make it
292
    /// easier to build resilliant apps. By allowing existing Client instances
293
    /// to recover and reconnect, each component of the apps built can adopt a
294
    /// "retry-to-recover" design, or "abort-and-fail" depending on how critical
295
    /// the database is to operation.
296
2831
    pub(crate) fn new_from_parts(
297
2831
        url: Url,
298
2831
        protocol_version: &'static str,
299
2831
        mut custom_apis: HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,
300
2831
        #[cfg(not(target_arch = "wasm32"))] certificate: Option<fabruic::Certificate>,
301
2831
        #[cfg(not(target_arch = "wasm32"))] tokio: Option<Handle>,
302
2831
    ) -> Result<Self, Error> {
303
2831
        let subscribers = SubscriberMap::default();
304
2831
        let callback_subscribers = subscribers.clone();
305
2831
        custom_apis.insert(
306
2831
            MessageReceived::name(),
307
2831
            Some(Arc::new(ApiCallback::<MessageReceived>::new(
308
2831
                move |message: MessageReceived| {
309
855
                    let callback_subscribers = callback_subscribers.clone();
310
855
                    async move {
311
855
                        let mut subscribers = callback_subscribers.lock();
312
855
                        if let Some(sender) = subscribers.get(&message.subscriber_id) {
313
855
                            if sender
314
855
                                .send(bonsaidb_core::circulate::Message {
315
855
                                    topic: OwnedBytes::from(message.topic.into_vec()),
316
855
                                    payload: OwnedBytes::from(message.payload.into_vec()),
317
855
                                })
318
855
                                .is_err()
319
                            {
320
                                subscribers.remove(&message.subscriber_id);
321
855
                            }
322
                        }
323
855
                    }
324
2831
                },
325
2831
            ))),
326
2831
        );
327
2831
        match url.scheme() {
328
2831
            #[cfg(not(target_arch = "wasm32"))]
329
2831
            "bonsaidb" => Ok(Self::new_bonsai_client(
330
1178
                url,
331
1178
                protocol_version,
332
1178
                certificate,
333
1178
                custom_apis,
334
1178
                tokio,
335
1178
                subscribers,
336
1178
            )),
337
            #[cfg(feature = "websockets")]
338
1653
            "wss" | "ws" => Ok(Self::new_websocket_client(
339
1653
                url,
340
1653
                protocol_version,
341
1653
                custom_apis,
342
1653
                #[cfg(not(target_arch = "wasm32"))]
343
1653
                tokio,
344
1653
                subscribers,
345
1653
            )),
346
            other => {
347
                return Err(Error::InvalidUrl(format!("unsupported scheme {}", other)));
348
            }
349
        }
350
2831
    }
351

            
352
    #[cfg(not(target_arch = "wasm32"))]
353
1178
    fn new_bonsai_client(
354
1178
        url: Url,
355
1178
        protocol_version: &'static str,
356
1178
        certificate: Option<fabruic::Certificate>,
357
1178
        custom_apis: HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,
358
1178
        tokio: Option<Handle>,
359
1178
        subscribers: SubscriberMap,
360
1178
    ) -> Self {
361
1178
        let (request_sender, request_receiver) = flume::unbounded();
362
1178

            
363
1178
        let worker = sync::spawn_client(
364
1178
            quic_worker::reconnecting_client_loop(
365
1178
                url,
366
1178
                protocol_version,
367
1178
                certificate,
368
1178
                request_receiver,
369
1178
                Arc::new(custom_apis),
370
1178
                subscribers.clone(),
371
1178
            ),
372
1178
            tokio,
373
1178
        );
374
1178

            
375
1178
        #[cfg(feature = "test-util")]
376
1178
        let background_task_running = Arc::new(AtomicBool::new(true));
377
1178

            
378
1178
        Self {
379
1178
            data: Arc::new(Data {
380
1178
                request_sender,
381
1178
                _worker: CancellableHandle {
382
1178
                    worker,
383
1178
                    #[cfg(feature = "test-util")]
384
1178
                    background_task_running: background_task_running.clone(),
385
1178
                },
386
1178
                schemas: Mutex::default(),
387
1178
                request_id: AtomicU32::default(),
388
1178
                effective_permissions: Mutex::default(),
389
1178
                subscribers,
390
1178
                #[cfg(feature = "test-util")]
391
1178
                background_task_running,
392
1178
            }),
393
1178
            session: Arc::new(Session::default()),
394
1178
        }
395
1178
    }
396

            
397
    #[cfg(all(feature = "websockets", not(target_arch = "wasm32")))]
398
1653
    fn new_websocket_client(
399
1653
        url: Url,
400
1653
        protocol_version: &'static str,
401
1653
        custom_apis: HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,
402
1653
        tokio: Option<Handle>,
403
1653
        subscribers: SubscriberMap,
404
1653
    ) -> Self {
405
1653
        let (request_sender, request_receiver) = flume::unbounded();
406
1653

            
407
1653
        let worker = sync::spawn_client(
408
1653
            tungstenite_worker::reconnecting_client_loop(
409
1653
                url,
410
1653
                protocol_version,
411
1653
                request_receiver,
412
1653
                Arc::new(custom_apis),
413
1653
                subscribers.clone(),
414
1653
            ),
415
1653
            tokio,
416
1653
        );
417
1653

            
418
1653
        #[cfg(feature = "test-util")]
419
1653
        let background_task_running = Arc::new(AtomicBool::new(true));
420
1653

            
421
1653
        Self {
422
1653
            data: Arc::new(Data {
423
1653
                request_sender,
424
1653
                #[cfg(not(target_arch = "wasm32"))]
425
1653
                _worker: CancellableHandle {
426
1653
                    worker,
427
1653
                    #[cfg(feature = "test-util")]
428
1653
                    background_task_running: background_task_running.clone(),
429
1653
                },
430
1653
                schemas: Mutex::default(),
431
1653
                request_id: AtomicU32::default(),
432
1653
                effective_permissions: Mutex::default(),
433
1653
                subscribers,
434
1653
                #[cfg(feature = "test-util")]
435
1653
                background_task_running,
436
1653
            }),
437
1653
            session: Arc::new(Session::default()),
438
1653
        }
439
1653
    }
440

            
441
    #[cfg(all(feature = "websockets", target_arch = "wasm32"))]
442
    fn new_websocket_client(
443
        url: Url,
444
        protocol_version: &'static str,
445
        custom_apis: HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,
446
        subscribers: SubscriberMap,
447
    ) -> Self {
448
        let (request_sender, request_receiver) = flume::unbounded();
449

            
450
        wasm_websocket_worker::spawn_client(
451
            Arc::new(url),
452
            protocol_version,
453
            request_receiver,
454
            Arc::new(custom_apis),
455
            subscribers.clone(),
456
        );
457

            
458
        #[cfg(feature = "test-util")]
459
        let background_task_running = Arc::new(AtomicBool::new(true));
460

            
461
        Self {
462
            data: Arc::new(Data {
463
                request_sender,
464
                #[cfg(not(target_arch = "wasm32"))]
465
                worker: CancellableHandle {
466
                    worker,
467
                    #[cfg(feature = "test-util")]
468
                    background_task_running: background_task_running.clone(),
469
                },
470
                schemas: Mutex::default(),
471
                request_id: AtomicU32::default(),
472
                effective_permissions: Mutex::default(),
473
                subscribers,
474
                #[cfg(feature = "test-util")]
475
                background_task_running,
476
            }),
477
            session: Arc::new(Session::default()),
478
        }
479
    }
480

            
481
1342122
    fn send_request_without_confirmation(
482
1342122
        &self,
483
1342122
        name: ApiName,
484
1342122
        bytes: Bytes,
485
1342122
    ) -> Result<flume::Receiver<Result<Bytes, Error>>, Error> {
486
1342122
        let (result_sender, result_receiver) = flume::bounded(1);
487
1342122
        let id = self.data.request_id.fetch_add(1, Ordering::SeqCst);
488
1342122
        self.data.request_sender.send(PendingRequest {
489
1342122
            request: Payload {
490
1342122
                session_id: self.session.id,
491
1342122
                id: Some(id),
492
1342122
                name,
493
1342122
                value: Ok(bytes),
494
1342122
            },
495
1342122
            responder: result_sender,
496
1342122
        })?;
497

            
498
1342122
        Ok(result_receiver)
499
1342122
    }
500

            
501
1126681
    async fn send_request_async(&self, name: ApiName, bytes: Bytes) -> Result<Bytes, Error> {
502
1126681
        let result_receiver = self.send_request_without_confirmation(name, bytes)?;
503

            
504
1506662
        result_receiver.recv_async().await?
505
1126681
    }
506

            
507
215308
    fn send_request(&self, name: ApiName, bytes: Bytes) -> Result<Bytes, Error> {
508
215308
        let result_receiver = self.send_request_without_confirmation(name, bytes)?;
509

            
510
215308
        result_receiver.recv()?
511
215308
    }
512

            
513
    /// Sends an api `request`.
514
1126015
    pub async fn send_api_request_async<Api: api::Api>(
515
1126015
        &self,
516
1126015
        request: &Api,
517
1126015
    ) -> Result<Api::Response, ApiError<Api::Error>> {
518
1126015
        let request = Bytes::from(pot::to_vec(request).map_err(Error::from)?);
519
1505834
        let response = self.send_request_async(Api::name(), request).await?;
520
1115770
        let response =
521
1115751
            pot::from_slice::<Result<Api::Response, Api::Error>>(&response).map_err(Error::from)?;
522
1115770
        response.map_err(ApiError::Api)
523
1126015
    }
524

            
525
    /// Sends an api `request` without waiting for a result. The response from
526
    /// the server will be ignored.
527
133
    pub fn invoke_api_request<Api: api::Api>(&self, request: &Api) -> Result<(), Error> {
528
133
        let request = Bytes::from(pot::to_vec(request).map_err(Error::from)?);
529
133
        self.send_request_without_confirmation(Api::name(), request)
530
133
            .map(|_| ())
531
133
    }
532

            
533
    /// Sends an api `request`.
534
215128
    pub fn send_api_request<Api: api::Api>(
535
215128
        &self,
536
215128
        request: &Api,
537
215128
    ) -> Result<Api::Response, ApiError<Api::Error>> {
538
215128
        let request = Bytes::from(pot::to_vec(request).map_err(Error::from)?);
539
215128
        let response = self.send_request(Api::name(), request)?;
540

            
541
214824
        let response =
542
214824
            pot::from_slice::<Result<Api::Response, Api::Error>>(&response).map_err(Error::from)?;
543
214824
        response.map_err(ApiError::Api)
544
215128
    }
545

            
546
    /// Returns the current effective permissions for the client. Returns None
547
    /// if unauthenticated.
548
    pub async fn effective_permissions(&self) -> Option<Permissions> {
549
        let effective_permissions = self.data.effective_permissions.lock();
550
        effective_permissions.clone()
551
    }
552

            
553
    #[cfg(feature = "test-util")]
554
    #[doc(hidden)]
555
    #[must_use]
556
    pub fn background_task_running(&self) -> Arc<AtomicBool> {
557
        self.data.background_task_running.clone()
558
    }
559

            
560
456
    pub(crate) fn register_subscriber(&self, id: u64, sender: flume::Sender<Message>) {
561
456
        let mut subscribers = self.data.subscribers.lock();
562
456
        subscribers.insert(id, sender);
563
456
    }
564

            
565
38
    pub(crate) async fn unregister_subscriber_async(&self, database: String, id: u64) {
566
38
        drop(
567
38
            self.send_api_request_async(&UnregisterSubscriber {
568
38
                database,
569
38
                subscriber_id: id,
570
38
            })
571
38
            .await,
572
        );
573
38
        let mut subscribers = self.data.subscribers.lock();
574
38
        subscribers.remove(&id);
575
38
    }
576

            
577
    #[cfg(not(target_arch = "wasm32"))]
578
152
    pub(crate) fn unregister_subscriber(&self, database: String, id: u64) {
579
152
        drop(self.send_api_request(&UnregisterSubscriber {
580
152
            database,
581
152
            subscriber_id: id,
582
152
        }));
583
152
        let mut subscribers = self.data.subscribers.lock();
584
152
        subscribers.remove(&id);
585
152
    }
586

            
587
1305
    fn database<DB: bonsaidb_core::schema::Schema>(
588
1305
        &self,
589
1305
        name: &str,
590
1305
    ) -> Result<RemoteDatabase, bonsaidb_core::Error> {
591
1305
        let mut schemas = self.data.schemas.lock();
592
1305
        let type_id = TypeId::of::<DB>();
593
1305
        let schematic = if let Some(schematic) = schemas.get(&type_id) {
594
1103
            schematic.clone()
595
        } else {
596
202
            let schematic = Arc::new(DB::schematic()?);
597
202
            schemas.insert(type_id, schematic.clone());
598
202
            schematic
599
        };
600
1305
        Ok(RemoteDatabase::new(
601
1305
            self.clone(),
602
1305
            name.to_string(),
603
1305
            schematic,
604
1305
        ))
605
1305
    }
606
}
607

            
608
impl HasSession for Client {
609
    fn session(&self) -> Option<&Session> {
610
        Some(&self.session)
611
    }
612
}
613

            
614
#[async_trait]
615
impl AsyncStorageConnection for Client {
616
    type Database = RemoteDatabase;
617
    type Authenticated = Self;
618

            
619
    async fn admin(&self) -> Self::Database {
620
        self.database::<Admin>(ADMIN_DATABASE_NAME).unwrap()
621
    }
622

            
623
10868
    async fn create_database_with_schema(
624
10868
        &self,
625
10868
        name: &str,
626
10868
        schema: SchemaName,
627
10868
        only_if_needed: bool,
628
10868
    ) -> Result<(), bonsaidb_core::Error> {
629
10868
        self.send_api_request_async(&CreateDatabase {
630
10868
            database: Database {
631
10868
                name: name.to_string(),
632
10868
                schema,
633
10868
            },
634
10868
            only_if_needed,
635
40850
        })
636
40850
        .await?;
637
10754
        Ok(())
638
21736
    }
639

            
640
1186
    async fn database<DB: Schema>(
641
1186
        &self,
642
1186
        name: &str,
643
1186
    ) -> Result<Self::Database, bonsaidb_core::Error> {
644
1186
        self.database::<DB>(name)
645
1186
    }
646

            
647
9576
    async fn delete_database(&self, name: &str) -> Result<(), bonsaidb_core::Error> {
648
9576
        self.send_api_request_async(&DeleteDatabase {
649
9576
            name: name.to_string(),
650
35131
        })
651
35131
        .await?;
652
9538
        Ok(())
653
19152
    }
654

            
655
38
    async fn list_databases(&self) -> Result<Vec<Database>, bonsaidb_core::Error> {
656
38
        Ok(self.send_api_request_async(&ListDatabases).await?)
657
76
    }
658

            
659
38
    async fn list_available_schemas(&self) -> Result<Vec<SchemaName>, bonsaidb_core::Error> {
660
38
        Ok(self.send_api_request_async(&ListAvailableSchemas).await?)
661
76
    }
662

            
663
76
    async fn create_user(&self, username: &str) -> Result<u64, bonsaidb_core::Error> {
664
76
        Ok(self
665
76
            .send_api_request_async(&CreateUser {
666
76
                username: username.to_string(),
667
76
            })
668
76
            .await?)
669
152
    }
670

            
671
2
    async fn delete_user<'user, U: Nameable<'user, u64> + Send + Sync>(
672
2
        &self,
673
2
        user: U,
674
2
    ) -> Result<(), bonsaidb_core::Error> {
675
        Ok(self
676
            .send_api_request_async(&DeleteUser {
677
2
                user: user.name()?.into_owned(),
678
2
            })
679
2
            .await?)
680
4
    }
681

            
682
    #[cfg(feature = "password-hashing")]
683
    async fn set_user_password<'user, U: Nameable<'user, u64> + Send + Sync>(
684
        &self,
685
        user: U,
686
        password: bonsaidb_core::connection::SensitiveString,
687
    ) -> Result<(), bonsaidb_core::Error> {
688
        Ok(self
689
            .send_api_request_async(&bonsaidb_core::networking::SetUserPassword {
690
                user: user.name()?.into_owned(),
691
                password,
692
            })
693
            .await?)
694
    }
695

            
696
    #[cfg(feature = "password-hashing")]
697
5
    async fn authenticate<'user, U: Nameable<'user, u64> + Send + Sync>(
698
5
        &self,
699
5
        user: U,
700
5
        authentication: Authentication,
701
5
    ) -> Result<Self::Authenticated, bonsaidb_core::Error> {
702
5
        let session = self
703
            .send_api_request_async(&bonsaidb_core::networking::Authenticate {
704
5
                user: user.name()?.into_owned(),
705
5
                authentication,
706
11
            })
707
11
            .await?;
708
5
        Ok(Self {
709
5
            data: self.data.clone(),
710
5
            session: Arc::new(session),
711
5
        })
712
10
    }
713

            
714
38
    async fn assume_identity(
715
38
        &self,
716
38
        identity: IdentityReference<'_>,
717
38
    ) -> Result<Self::Authenticated, bonsaidb_core::Error> {
718
38
        let session = self
719
38
            .send_api_request_async(&AssumeIdentity(identity.into_owned()))
720
38
            .await?;
721
38
        Ok(Self {
722
38
            data: self.data.clone(),
723
38
            session: Arc::new(session),
724
38
        })
725
76
    }
726

            
727
4
    async fn add_permission_group_to_user<
728
4
        'user,
729
4
        'group,
730
4
        U: Nameable<'user, u64> + Send + Sync,
731
4
        G: Nameable<'group, u64> + Send + Sync,
732
4
    >(
733
4
        &self,
734
4
        user: U,
735
4
        permission_group: G,
736
4
    ) -> Result<(), bonsaidb_core::Error> {
737
        self.send_api_request_async(&AlterUserPermissionGroupMembership {
738
4
            user: user.name()?.into_owned(),
739
4
            group: permission_group.name()?.into_owned(),
740
            should_be_member: true,
741
4
        })
742
4
        .await?;
743
4
        Ok(())
744
8
    }
745

            
746
4
    async fn remove_permission_group_from_user<
747
4
        'user,
748
4
        'group,
749
4
        U: Nameable<'user, u64> + Send + Sync,
750
4
        G: Nameable<'group, u64> + Send + Sync,
751
4
    >(
752
4
        &self,
753
4
        user: U,
754
4
        permission_group: G,
755
4
    ) -> Result<(), bonsaidb_core::Error> {
756
        self.send_api_request_async(&AlterUserPermissionGroupMembership {
757
4
            user: user.name()?.into_owned(),
758
4
            group: permission_group.name()?.into_owned(),
759
            should_be_member: false,
760
4
        })
761
4
        .await?;
762
4
        Ok(())
763
8
    }
764

            
765
4
    async fn add_role_to_user<
766
4
        'user,
767
4
        'group,
768
4
        U: Nameable<'user, u64> + Send + Sync,
769
4
        G: Nameable<'group, u64> + Send + Sync,
770
4
    >(
771
4
        &self,
772
4
        user: U,
773
4
        role: G,
774
4
    ) -> Result<(), bonsaidb_core::Error> {
775
        self.send_api_request_async(&AlterUserRoleMembership {
776
4
            user: user.name()?.into_owned(),
777
4
            role: role.name()?.into_owned(),
778
            should_be_member: true,
779
4
        })
780
4
        .await?;
781
4
        Ok(())
782
8
    }
783

            
784
4
    async fn remove_role_from_user<
785
4
        'user,
786
4
        'group,
787
4
        U: Nameable<'user, u64> + Send + Sync,
788
4
        G: Nameable<'group, u64> + Send + Sync,
789
4
    >(
790
4
        &self,
791
4
        user: U,
792
4
        role: G,
793
4
    ) -> Result<(), bonsaidb_core::Error> {
794
        self.send_api_request_async(&AlterUserRoleMembership {
795
4
            user: user.name()?.into_owned(),
796
4
            role: role.name()?.into_owned(),
797
            should_be_member: false,
798
4
        })
799
4
        .await?;
800
4
        Ok(())
801
8
    }
802
}
803

            
804
type OutstandingRequestMap = HashMap<u32, PendingRequest>;
805
type OutstandingRequestMapHandle = Arc<async_lock::Mutex<OutstandingRequestMap>>;
806
type PendingRequestResponder = Sender<Result<Bytes, Error>>;
807

            
808
#[derive(Debug)]
809
pub struct PendingRequest {
810
    request: Payload,
811
    responder: PendingRequestResponder,
812
}
813

            
814
#[cfg(not(target_arch = "wasm32"))]
815
#[derive(Debug)]
816
struct CancellableHandle<T> {
817
    worker: JoinHandle<T>,
818
    #[cfg(feature = "test-util")]
819
    background_task_running: Arc<AtomicBool>,
820
}
821

            
822
#[cfg(not(target_arch = "wasm32"))]
823
impl<T> Drop for CancellableHandle<T> {
824
2812
    fn drop(&mut self) {
825
2812
        self.worker.abort();
826
2812
        #[cfg(feature = "test-util")]
827
2812
        self.background_task_running.store(false, Ordering::Release);
828
2812
    }
829
}
830

            
831
1342806
async fn process_response_payload(
832
1342806
    payload: Payload,
833
1342806
    outstanding_requests: &OutstandingRequestMapHandle,
834
1342806
    custom_apis: &HashMap<ApiName, Option<Arc<dyn AnyApiCallback>>>,
835
1342806
) {
836
1342806
    if let Some(payload_id) = payload.id {
837
1341951
        if let Some(outstanding_request) = {
838
1341951
            let mut outstanding_requests = fast_async_lock!(outstanding_requests);
839
1341951
            outstanding_requests.remove(&payload_id)
840
1341951
        } {
841
1341951
            drop(
842
1341951
                outstanding_request
843
1341951
                    .responder
844
1341951
                    .send(payload.value.map_err(Error::from)),
845
1341951
            );
846
1341951
        }
847
855
    } else if let (Some(custom_api_callback), Ok(value)) = (
848
855
        custom_apis.get(&payload.name).and_then(Option::as_ref),
849
855
        payload.value,
850
855
    ) {
851
855
        custom_api_callback.response_received(value).await;
852
    } else {
853
        log::warn!("unexpected api response received ({})", payload.name);
854
    }
855
1342806
}
856

            
857
trait ApiWrapper<Response>: Send + Sync {
858
    fn invoke(&self, response: Response) -> BoxFuture<'static, ()>;
859
}
860

            
861
/// A callback that is invoked when an [`Api::Response`](Api::Response)
862
/// value is received out-of-band (not in reply to a request).
863
pub struct ApiCallback<Api: api::Api> {
864
    generator: Box<dyn ApiWrapper<Api::Response>>,
865
}
866

            
867
/// The trait bounds required for the function wrapped in a [`ApiCallback`].
868
pub trait ApiCallbackFn<Request, F>: Fn(Request) -> F + Send + Sync + 'static {}
869

            
870
impl<T, Request, F> ApiCallbackFn<Request, F> for T where T: Fn(Request) -> F + Send + Sync + 'static
871
{}
872

            
873
struct ApiFutureBoxer<Response: Send + Sync, F: Future<Output = ()> + Send + Sync>(
874
    Box<dyn ApiCallbackFn<Response, F>>,
875
);
876

            
877
impl<Response: Send + Sync, F: Future<Output = ()> + Send + Sync + 'static> ApiWrapper<Response>
878
    for ApiFutureBoxer<Response, F>
879
{
880
855
    fn invoke(&self, response: Response) -> BoxFuture<'static, ()> {
881
855
        (&self.0)(response).boxed()
882
855
    }
883
}
884

            
885
impl<Api: api::Api> ApiCallback<Api> {
886
    /// Returns a new instance wrapping the provided function.
887
2831
    pub fn new<
888
2831
        F: ApiCallbackFn<Api::Response, Fut>,
889
2831
        Fut: Future<Output = ()> + Send + Sync + 'static,
890
2831
    >(
891
2831
        callback: F,
892
2831
    ) -> Self {
893
2831
        Self {
894
2831
            generator: Box::new(ApiFutureBoxer::<Api::Response, Fut>(Box::new(callback))),
895
2831
        }
896
2831
    }
897
    /// Returns a new instance wrapping the provided function, passing a clone
898
    /// of `context` as the second parameter. This is just a convenience wrapper
899
    /// around `new()` that produces more readable code when needing to access
900
    /// external information inside of the callback.
901
    pub fn new_with_context<
902
        Context: Send + Sync + Clone + 'static,
903
        F: Fn(Api::Response, Context) -> Fut + Send + Sync + 'static,
904
        Fut: Future<Output = ()> + Send + Sync + 'static,
905
    >(
906
        context: Context,
907
        callback: F,
908
    ) -> Self {
909
        Self {
910
            generator: Box::new(ApiFutureBoxer::<Api::Response, Fut>(Box::new(
911
                move |request| {
912
                    let context = context.clone();
913
                    callback(request, context)
914
                },
915
            ))),
916
        }
917
    }
918
}
919

            
920
#[async_trait]
921
pub trait AnyApiCallback: Send + Sync + 'static {
922
    /// An out-of-band `response` was received. This happens when the server
923
    /// sends a response that isn't in response to a request.
924
    async fn response_received(&self, response: Bytes);
925
}
926

            
927
#[async_trait]
928
impl<Api: api::Api> AnyApiCallback for ApiCallback<Api> {
929
855
    async fn response_received(&self, response: Bytes) {
930
855
        match pot::from_slice::<Result<Api::Response, Api::Error>>(&response) {
931
855
            Ok(response) => self.generator.invoke(response.unwrap()).await,
932
            Err(err) => {
933
                log::error!("error deserializing api: {err}");
934
            }
935
        }
936
855
    }
937
}