1
use std::{ops::Deref, sync::Arc};
2

            
3
use async_trait::async_trait;
4
use bonsaidb_core::{
5
    arc_bytes::serde::Bytes,
6
    connection::{
7
        AccessPolicy, AsyncConnection, AsyncLowLevelConnection, HasSession, QueryKey, Range,
8
        Session, Sort,
9
    },
10
    document::{DocumentId, Header, OwnedDocument},
11
    networking::{
12
        ApplyTransaction, Compact, CompactCollection, CompactKeyValueStore, Count, DeleteDocs, Get,
13
        GetMultiple, LastTransactionId, List, ListExecutedTransactions, ListHeaders, Query,
14
        QueryWithDocs, Reduce, ReduceGrouped,
15
    },
16
    schema::{self, view::map::MappedSerializedValue, CollectionName, Schematic, ViewName},
17
    transaction::{Executed, OperationResult, Transaction},
18
};
19

            
20
use crate::Client;
21

            
22
mod pubsub;
23
pub use pubsub::*;
24

            
25
mod keyvalue;
26

            
27
/// A database on a remote server.
28
392
#[derive(Debug, Clone)]
29
pub struct RemoteDatabase {
30
    pub(crate) client: Client,
31
    pub(crate) name: Arc<String>,
32
    pub(crate) schema: Arc<Schematic>,
33
}
34
impl RemoteDatabase {
35
    /// Returns the name of the database.
36
    #[must_use]
37
    pub fn name(&self) -> &str {
38
        self.name.as_ref()
39
    }
40
}
41

            
42
impl Deref for RemoteDatabase {
43
    type Target = Client;
44

            
45
171
    fn deref(&self) -> &Self::Target {
46
171
        &self.client
47
171
    }
48
}
49

            
50
impl RemoteDatabase {
51
23769
    pub(crate) fn new(client: Client, name: String, schema: Arc<Schematic>) -> Self {
52
23769
        Self {
53
23769
            client,
54
23769
            name: Arc::new(name),
55
23769
            schema,
56
23769
        }
57
23769
    }
58
}
59

            
60
impl HasSession for RemoteDatabase {
61
    fn session(&self) -> Option<&Session> {
62
        Some(&self.session)
63
    }
64
}
65

            
66
#[async_trait]
67
impl AsyncConnection for RemoteDatabase {
68
    type Storage = Client;
69

            
70
    fn storage(&self) -> Self::Storage {
71
        self.client.clone()
72
    }
73

            
74
19684
    async fn list_executed_transactions(
75
19684
        &self,
76
19684
        starting_id: Option<u64>,
77
19684
        result_limit: Option<u32>,
78
19684
    ) -> Result<Vec<Executed>, bonsaidb_core::Error> {
79
19684
        Ok(self
80
19684
            .client
81
19684
            .send_api_request_async(&ListExecutedTransactions {
82
19684
                database: self.name.to_string(),
83
19684
                starting_id,
84
19684
                result_limit,
85
59584
            })
86
59584
            .await?)
87
39368
    }
88

            
89
38
    async fn last_transaction_id(&self) -> Result<Option<u64>, bonsaidb_core::Error> {
90
38
        Ok(self
91
38
            .client
92
38
            .send_api_request_async(&LastTransactionId {
93
38
                database: self.name.to_string(),
94
38
            })
95
38
            .await?)
96
76
    }
97

            
98
38
    async fn compact(&self) -> Result<(), bonsaidb_core::Error> {
99
38
        self.send_api_request_async(&Compact {
100
38
            database: self.name.to_string(),
101
38
        })
102
38
        .await?;
103
38
        Ok(())
104
76
    }
105

            
106
38
    async fn compact_key_value_store(&self) -> Result<(), bonsaidb_core::Error> {
107
38
        self.send_api_request_async(&CompactKeyValueStore {
108
38
            database: self.name.to_string(),
109
38
        })
110
38
        .await?;
111
38
        Ok(())
112
76
    }
113
}
114

            
115
#[async_trait]
116
impl AsyncLowLevelConnection for RemoteDatabase {
117
251579
    fn schematic(&self) -> &Schematic {
118
251579
        &self.schema
119
251579
    }
120

            
121
199253
    async fn apply_transaction(
122
199253
        &self,
123
199253
        transaction: Transaction,
124
199253
    ) -> Result<Vec<OperationResult>, bonsaidb_core::Error> {
125
199253
        Ok(self
126
199253
            .client
127
199253
            .send_api_request_async(&ApplyTransaction {
128
199253
                database: self.name.to_string(),
129
199253
                transaction,
130
407512
            })
131
407512
            .await?)
132
398506
    }
133

            
134
165414
    async fn get_from_collection(
135
165414
        &self,
136
165414
        id: DocumentId,
137
165414
        collection: &CollectionName,
138
165414
    ) -> Result<Option<OwnedDocument>, bonsaidb_core::Error> {
139
165414
        Ok(self
140
165414
            .client
141
165414
            .send_api_request_async(&Get {
142
165414
                database: self.name.to_string(),
143
165414
                collection: collection.clone(),
144
165414
                id,
145
241452
            })
146
241452
            .await?)
147
330828
    }
148

            
149
83486
    async fn get_multiple_from_collection(
150
83486
        &self,
151
83486
        ids: &[DocumentId],
152
83486
        collection: &CollectionName,
153
83486
    ) -> Result<Vec<OwnedDocument>, bonsaidb_core::Error> {
154
83486
        Ok(self
155
83486
            .client
156
83486
            .send_api_request_async(&GetMultiple {
157
83486
                database: self.name.to_string(),
158
83486
                collection: collection.clone(),
159
83486
                ids: ids.to_vec(),
160
83486
            })
161
83486
            .await?)
162
166972
    }
163

            
164
152
    async fn list_from_collection(
165
152
        &self,
166
152
        ids: Range<DocumentId>,
167
152
        order: Sort,
168
152
        limit: Option<u32>,
169
152
        collection: &CollectionName,
170
152
    ) -> Result<Vec<OwnedDocument>, bonsaidb_core::Error> {
171
152
        Ok(self
172
152
            .client
173
152
            .send_api_request_async(&List {
174
152
                database: self.name.to_string(),
175
152
                collection: collection.clone(),
176
152
                ids,
177
152
                order,
178
152
                limit,
179
152
            })
180
152
            .await?)
181
304
    }
182

            
183
38
    async fn list_headers_from_collection(
184
38
        &self,
185
38
        ids: Range<DocumentId>,
186
38
        order: Sort,
187
38
        limit: Option<u32>,
188
38
        collection: &CollectionName,
189
38
    ) -> Result<Vec<Header>, bonsaidb_core::Error> {
190
38
        Ok(self
191
38
            .client
192
38
            .send_api_request_async(&ListHeaders(List {
193
38
                database: self.name.to_string(),
194
38
                collection: collection.clone(),
195
38
                ids,
196
38
                order,
197
38
                limit,
198
38
            }))
199
38
            .await?)
200
76
    }
201

            
202
76
    async fn count_from_collection(
203
76
        &self,
204
76
        ids: Range<DocumentId>,
205
76
        collection: &CollectionName,
206
76
    ) -> Result<u64, bonsaidb_core::Error> {
207
76
        Ok(self
208
76
            .client
209
76
            .send_api_request_async(&Count {
210
76
                database: self.name.to_string(),
211
76
                collection: collection.clone(),
212
76
                ids,
213
76
            })
214
76
            .await?)
215
152
    }
216

            
217
38
    async fn compact_collection_by_name(
218
38
        &self,
219
38
        collection: CollectionName,
220
38
    ) -> Result<(), bonsaidb_core::Error> {
221
38
        self.send_api_request_async(&CompactCollection {
222
38
            database: self.name.to_string(),
223
38
            name: collection,
224
38
        })
225
38
        .await?;
226
38
        Ok(())
227
76
    }
228

            
229
84398
    async fn query_by_name(
230
84398
        &self,
231
84398
        view: &ViewName,
232
84398
        key: Option<QueryKey<Bytes>>,
233
84398
        order: Sort,
234
84398
        limit: Option<u32>,
235
84398
        access_policy: AccessPolicy,
236
84398
    ) -> Result<Vec<schema::view::map::Serialized>, bonsaidb_core::Error> {
237
84398
        Ok(self
238
84398
            .client
239
84398
            .send_api_request_async(&Query {
240
84398
                database: self.name.to_string(),
241
84398
                view: view.clone(),
242
84398
                key,
243
84398
                order,
244
84398
                limit,
245
84398
                access_policy,
246
84398
            })
247
84398
            .await?)
248
168796
    }
249

            
250
    async fn query_by_name_with_docs(
251
        &self,
252
        view: &ViewName,
253
        key: Option<QueryKey<Bytes>>,
254
        order: Sort,
255
        limit: Option<u32>,
256
        access_policy: AccessPolicy,
257
    ) -> Result<schema::view::map::MappedSerializedDocuments, bonsaidb_core::Error> {
258
        Ok(self
259
            .client
260
            .send_api_request_async(&QueryWithDocs(Query {
261
                database: self.name.to_string(),
262
                view: view.clone(),
263
                key,
264
                order,
265
                limit,
266
                access_policy,
267
            }))
268
            .await?)
269
    }
270

            
271
166991
    async fn reduce_by_name(
272
166991
        &self,
273
166991
        view: &ViewName,
274
166991
        key: Option<QueryKey<Bytes>>,
275
166991
        access_policy: AccessPolicy,
276
166991
    ) -> Result<Vec<u8>, bonsaidb_core::Error> {
277
166991
        Ok(self
278
166991
            .client
279
166991
            .send_api_request_async(&Reduce {
280
166991
                database: self.name.to_string(),
281
166991
                view: view.clone(),
282
166991
                key,
283
166991
                access_policy,
284
166991
            })
285
166991
            .await?
286
166953
            .into_vec())
287
333982
    }
288

            
289
114
    async fn reduce_grouped_by_name(
290
114
        &self,
291
114
        view: &ViewName,
292
114
        key: Option<QueryKey<Bytes>>,
293
114
        access_policy: AccessPolicy,
294
114
    ) -> Result<Vec<MappedSerializedValue>, bonsaidb_core::Error> {
295
114
        Ok(self
296
114
            .client
297
114
            .send_api_request_async(&ReduceGrouped(Reduce {
298
114
                database: self.name.to_string(),
299
114
                view: view.clone(),
300
114
                key,
301
114
                access_policy,
302
133
            }))
303
133
            .await?)
304
228
    }
305

            
306
76
    async fn delete_docs_by_name(
307
76
        &self,
308
76
        view: &ViewName,
309
76
        key: Option<QueryKey<Bytes>>,
310
76
        access_policy: AccessPolicy,
311
76
    ) -> Result<u64, bonsaidb_core::Error> {
312
76
        Ok(self
313
76
            .client
314
76
            .send_api_request_async(&DeleteDocs {
315
76
                database: self.name.to_string(),
316
76
                view: view.clone(),
317
76
                key,
318
76
                access_policy,
319
76
            })
320
76
            .await?)
321
152
    }
322
}