1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
use arc_bytes::serde::Bytes;
use serde::{Deserialize, Serialize};

use crate::{
    connection::{AsyncLowLevelConnection, LowLevelConnection},
    document::{CollectionHeader, DocumentId, Header},
    schema::{CollectionName, SerializedCollection},
    Error,
};

/// A list of operations to execute as a single unit. If any operation fails,
/// all changes are aborted. Transactions are ACID-compliant. ACID stands for:
///
/// - Atomic: All transactions are atomically applied. Readers outside of the
///   active transaction will never be able to read partially written data. In
///   BonsaiDb, readers are not blocked while writes are happening -- reads will
///   continue to read the existing value until the transaction is fully
///   executed. Once the transaction is fully executed, all future queries will
///   reflect the updated state immediately.
///
/// - Consistent: All transactions will be applied only if the data model is
///   able to remain fully consistent. This means that all constraints, such as
///   unique view keys, are validated before a transaction is allowed to be
///   committed.
///
/// - Isolated: Each transaction is executed in an isolated environment.
///   Currently, BonsaiDb does not offer interactive transactions, so this is
///   easily guaranteed. When BonsaiDb eventually has interactive transactions,
///   the transaction will have a fully isolated state until it is committed. No
///   two transactions can be affected by each other's changes.
///
///   In the event of a transaction being aborted or a power outage occurs while
///   a transaction is being applied, this isolation ensures that once BonsaiDb
///   opens the database again, the database will reflect the most recently
///   committed.
///
/// - Durable: When the transaction apply function has finished exectuing,
///   BonsaiDb guarantees that all data has been confirmed by the operating
///   system as being fully written to disk. This ensures that in the event of a
///   power outage, no data that has been confirmed will be lost.
///
/// When using one of the high-level functions to push/insert/update/delete
/// documents, behind the scenes single-[`Operation`] `Transaction`s are
/// applied. To ensure multiple changes happen in the same database operation,
/// multiple operations can be added to a `Transaction`:
///
/// ```rust
/// # bonsaidb_core::__doctest_prelude!();
/// # use bonsaidb_core::connection::Connection;
/// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {
/// use bonsaidb_core::transaction::{Operation, Transaction};
/// let mut tx = Transaction::new();
/// tx.push(Operation::push_serialized::<MyCollection>(
///     &MyCollection::default(),
/// )?);
/// tx.push(Operation::push_serialized::<MyCollection>(
///     &MyCollection::default(),
/// )?);
/// let results = tx.apply(db)?;
/// assert_eq!(results.len(), 2);
/// println!("Two new documents inserted: {results:?}");
/// # Ok(())
/// # }
/// ```
#[derive(Clone, Serialize, Deserialize, Default, Debug)]
#[must_use]
pub struct Transaction {
    /// The operations in this transaction.
    pub operations: Vec<Operation>,
}

impl Transaction {
    /// Returns a new, empty transaction.
    pub fn new() -> Self {
        Self::default()
    }

    /// Adds an operation to the transaction.
    pub fn push(&mut self, operation: Operation) {
        self.operations.push(operation);
    }

    /// Appends an operation to the transaction and returns self.
    pub fn with(mut self, operation: Operation) -> Self {
        self.push(operation);
        self
    }

    /// Applies the transaction to the `database`, returning the results of the
    /// operations. All operations will succeed or none will be performed and an
    /// error will be returned.
    pub fn apply<Connection: LowLevelConnection>(
        self,
        db: &Connection,
    ) -> Result<Vec<OperationResult>, Error> {
        db.apply_transaction(self)
    }

    /// Applies the transaction to the `database`, returning the results of the
    /// operations. All operations will succeed or none will be performed and an
    /// error will be returned.
    pub async fn apply_async<Connection: AsyncLowLevelConnection>(
        self,
        db: &Connection,
    ) -> Result<Vec<OperationResult>, Error> {
        db.apply_transaction(self).await
    }
}

impl From<Operation> for Transaction {
    fn from(operation: Operation) -> Self {
        Self {
            operations: vec![operation],
        }
    }
}

impl Transaction {
    /// Inserts a new document with `contents` into `collection`.  If `id` is
    /// `None` a unique id will be generated. If an id is provided and a
    /// document already exists with that id, a conflict error will be returned.
    pub fn insert(
        collection: CollectionName,
        id: Option<DocumentId>,
        contents: impl Into<Bytes>,
    ) -> Self {
        Self::from(Operation::insert(collection, id, contents))
    }

    /// Updates a document in `collection`.
    pub fn update(collection: CollectionName, header: Header, contents: impl Into<Bytes>) -> Self {
        Self::from(Operation::update(collection, header, contents))
    }

    /// Overwrites a document in `collection`. If a document with `id` exists,
    /// it will be overwritten. If a document with `id` doesn't exist, it will
    /// be created.
    pub fn overwrite(
        collection: CollectionName,
        id: DocumentId,
        contents: impl Into<Bytes>,
    ) -> Self {
        Self::from(Operation::overwrite(collection, id, contents))
    }

    /// Deletes a document from a `collection`.
    pub fn delete(collection: CollectionName, header: Header) -> Self {
        Self::from(Operation::delete(collection, header))
    }
}

/// A single operation performed on a `Collection`.
#[derive(Clone, Serialize, Deserialize, Debug)]
#[must_use]
pub struct Operation {
    /// The id of the `Collection`.
    pub collection: CollectionName,

    /// The command being performed.
    pub command: Command,
}

impl Operation {
    /// Inserts a new document with `contents` into `collection`.  If `id` is
    /// `None` a unique id will be generated. If an id is provided and a
    /// document already exists with that id, a conflict error will be returned.
    pub fn insert(
        collection: CollectionName,
        id: Option<DocumentId>,
        contents: impl Into<Bytes>,
    ) -> Self {
        Self {
            collection,
            command: Command::Insert {
                id,
                contents: contents.into(),
            },
        }
    }

    /// Inserts a new document with the serialized representation of `contents`
    /// into `collection`. If `id` is `None` a unique id will be generated. If
    /// an id is provided and a document already exists with that id, a conflict
    /// error will be returned.
    pub fn insert_serialized<C: SerializedCollection>(
        id: Option<C::PrimaryKey>,
        contents: &C::Contents,
    ) -> Result<Self, Error> {
        let id = id.map(DocumentId::new).transpose()?;
        let contents = C::serialize(contents)?;
        Ok(Self::insert(C::collection_name(), id, contents))
    }

    /// Pushes a new document with the serialized representation of `contents`
    /// into `collection`.
    ///
    /// ## Automatic ID Assignment
    ///
    /// This function calls [`SerializedCollection::natural_id()`] to try to
    /// retrieve a primary key value from `contents`. If an id is returned, the
    /// item is inserted with that id. If an id is not returned, an id will be
    /// automatically assigned, if possible, by the storage backend, which uses
    /// the [`Key`](crate::key::Key) trait to assign ids.
    pub fn push_serialized<C: SerializedCollection>(contents: &C::Contents) -> Result<Self, Error> {
        let id = C::natural_id(contents);
        let id = id.map(DocumentId::new).transpose()?;
        let contents = C::serialize(contents)?;
        Ok(Self::insert(C::collection_name(), id, contents))
    }

    /// Updates a document in `collection`.
    pub fn update(collection: CollectionName, header: Header, contents: impl Into<Bytes>) -> Self {
        Self {
            collection,
            command: Command::Update {
                header,
                contents: contents.into(),
            },
        }
    }

    /// Updates a document with the serialized representation of `contents` in
    /// `collection`.
    pub fn update_serialized<C: SerializedCollection>(
        header: CollectionHeader<C::PrimaryKey>,
        contents: &C::Contents,
    ) -> Result<Self, Error> {
        let contents = C::serialize(contents)?;
        Ok(Self::update(
            C::collection_name(),
            Header::try_from(header)?,
            contents,
        ))
    }

    /// Overwrites a document in `collection`. If a document with `id` exists,
    /// it will be overwritten. If a document with `id` doesn't exist, it will
    /// be created.
    pub fn overwrite(
        collection: CollectionName,
        id: DocumentId,
        contents: impl Into<Bytes>,
    ) -> Self {
        Self {
            collection,
            command: Command::Overwrite {
                id,
                contents: contents.into(),
            },
        }
    }

    /// Overwrites a document with the serialized representation of `contents`
    /// in `collection`. If a document with `id` exists, it will be overwritten.
    /// If a document with `id` doesn't exist, it will be created.
    pub fn overwrite_serialized<C: SerializedCollection>(
        id: C::PrimaryKey,
        contents: &C::Contents,
    ) -> Result<Self, Error> {
        let contents = C::serialize(contents)?;
        Ok(Self::overwrite(
            C::collection_name(),
            DocumentId::new(id)?,
            contents,
        ))
    }

    /// Deletes a document from a `collection`.
    pub const fn delete(collection: CollectionName, header: Header) -> Self {
        Self {
            collection,
            command: Command::Delete { header },
        }
    }
}

/// A command to execute within a `Collection`.
#[derive(Clone, Serialize, Deserialize, Debug)]
pub enum Command {
    /// Inserts a new document containing `contents`.
    Insert {
        /// An optional id for the document. If this is `None`, a unique id will
        /// be generated. If this is `Some()` and a document already exists with
        /// that id, a conflict error will be returned.
        id: Option<DocumentId>,
        /// The initial contents of the document.
        contents: Bytes,
    },

    /// Update an existing `Document` identified by `header`. `header.revision` must match
    /// the currently stored revision on the `Document`. If it does not, the
    /// command fill fail with a `DocumentConflict` error.
    Update {
        /// The header of the `Document`. The revision must match the current
        /// document.
        header: Header,

        /// The new contents to store within the `Document`.
        contents: Bytes,
    },

    /// Overwrite an existing `Document` identified by `id`. The revision will
    /// not be checked before the document is updated. If the document does not
    /// exist, it will be created.
    Overwrite {
        /// The id of the document to overwrite.
        id: DocumentId,

        /// The new contents to store within the `Document`.
        contents: Bytes,
    },

    /// Delete an existing `Document` identified by `id`. `revision` must match
    /// the currently stored revision on the `Document`. If it does not, the
    /// command fill fail with a `DocumentConflict` error.
    Delete {
        /// The current header of the `Document`.
        header: Header,
    },
}

/// Information about the result of each `Operation` in a transaction.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum OperationResult {
    /// An operation succeeded but had no information to output.
    Success,

    /// A `Document` was updated.
    DocumentUpdated {
        /// The id of the `Collection` of the updated `Document`.
        collection: CollectionName,

        /// The header of the updated `Document`.
        header: Header,
    },

    /// A `Document` was deleted.
    DocumentDeleted {
        /// The id of the `Collection` of the deleted `Document`.
        collection: CollectionName,

        /// The id of the deleted `Document`.
        id: DocumentId,
    },
}

/// Details about an executed transaction.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Executed {
    /// The id of the transaction.
    pub id: u64,

    /// A list of containing ids of `Documents` changed.
    pub changes: Changes,
}

/// A list of changes.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum Changes {
    /// A list of changed documents.
    Documents(DocumentChanges),
    /// A list of changed keys.
    Keys(Vec<ChangedKey>),
}

impl Changes {
    /// Returns the list of documents changed in this transaction, or None if
    /// the transaction was not a document transaction.
    #[must_use]
    pub const fn documents(&self) -> Option<&DocumentChanges> {
        if let Self::Documents(changes) = self {
            Some(changes)
        } else {
            None
        }
    }

    /// Returns the list of keys changed in this transaction, or None if the
    /// transaction was not a `KeyValue` transaction.
    #[must_use]
    pub fn keys(&self) -> Option<&[ChangedKey]> {
        if let Self::Keys(keys) = self {
            Some(keys)
        } else {
            None
        }
    }
}

/// A list of changed documents.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct DocumentChanges {
    /// All of the collections changed.
    pub collections: Vec<CollectionName>,
    /// The individual document changes.
    pub documents: Vec<ChangedDocument>,
}

impl DocumentChanges {
    /// Returns the changed document and the name of the collection the change
    /// happened to.
    #[must_use]
    pub fn get(&self, index: usize) -> Option<(&CollectionName, &ChangedDocument)> {
        self.documents.get(index).and_then(|doc| {
            self.collections
                .get(usize::from(doc.collection))
                .map(|collection| (collection, doc))
        })
    }

    /// Returns the number of changes in this collection.
    #[must_use]
    pub fn len(&self) -> usize {
        self.documents.len()
    }

    /// Returns true if there are no changes.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.documents.is_empty()
    }

    /// Returns an interator over all of the changed documents.
    pub const fn iter(&self) -> DocumentChangesIter<'_> {
        DocumentChangesIter {
            changes: self,
            index: Some(0),
        }
    }
}

/// An iterator over [`DocumentChanges`].
#[must_use]
pub struct DocumentChangesIter<'a> {
    changes: &'a DocumentChanges,
    index: Option<usize>,
}

impl<'a> Iterator for DocumentChangesIter<'a> {
    type Item = (&'a CollectionName, &'a ChangedDocument);

    fn next(&mut self) -> Option<Self::Item> {
        self.index.and_then(|index| {
            let result = self.changes.get(index);
            if result.is_some() {
                self.index = index.checked_add(1);
            }
            result
        })
    }
}

/// A draining iterator over [`ChangedDocument`]s.
#[must_use]
pub struct DocumentChangesIntoIter {
    collections: Vec<CollectionName>,
    documents: std::vec::IntoIter<ChangedDocument>,
}

impl Iterator for DocumentChangesIntoIter {
    type Item = (CollectionName, ChangedDocument);

    fn next(&mut self) -> Option<Self::Item> {
        self.documents.next().and_then(|doc| {
            self.collections
                .get(usize::from(doc.collection))
                .map(|collection| (collection.clone(), doc))
        })
    }
}

impl IntoIterator for DocumentChanges {
    type Item = (CollectionName, ChangedDocument);

    type IntoIter = DocumentChangesIntoIter;

    fn into_iter(self) -> Self::IntoIter {
        DocumentChangesIntoIter {
            collections: self.collections,
            documents: self.documents.into_iter(),
        }
    }
}

#[test]
fn document_changes_iter() {
    use crate::schema::Qualified;

    let changes = DocumentChanges {
        collections: vec![CollectionName::private("a"), CollectionName::private("b")],
        documents: vec![
            ChangedDocument {
                collection: 0,
                id: DocumentId::from_u64(0),
                deleted: false,
            },
            ChangedDocument {
                collection: 0,
                id: DocumentId::from_u64(1),
                deleted: false,
            },
            ChangedDocument {
                collection: 1,
                id: DocumentId::from_u64(2),
                deleted: false,
            },
            ChangedDocument {
                collection: 2,
                id: DocumentId::from_u64(3),
                deleted: false,
            },
        ],
    };

    assert_eq!(changes.len(), 4);
    assert!(!changes.is_empty());

    let mut a_changes = 0;
    let mut b_changes = 0;
    let mut ids = Vec::new();
    for (collection, document) in changes.iter() {
        assert!(!ids.contains(&document.id));
        ids.push(document.id);
        match collection.name.as_ref() {
            "a" => a_changes += 1,
            "b" => b_changes += 1,
            _ => unreachable!("invalid collection name {collection}"),
        }
    }
    assert_eq!(a_changes, 2);
    assert_eq!(b_changes, 1);

    let mut a_changes = 0;
    let mut b_changes = 0;
    let mut ids = Vec::new();
    for (collection, document) in changes {
        assert!(!ids.contains(&document.id));
        ids.push(document.id);
        match collection.name.as_ref() {
            "a" => a_changes += 1,
            "b" => b_changes += 1,
            _ => unreachable!("invalid collection name {collection}"),
        }
    }
    assert_eq!(a_changes, 2);
    assert_eq!(b_changes, 1);
}

/// A record of a changed document.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChangedDocument {
    /// The index of the `CollectionName` within the `collections` field of [`Changes::Documents`].
    pub collection: u16,

    /// The id of the changed `Document`.
    pub id: DocumentId,

    /// If the `Document` has been deleted, this will be `true`.
    pub deleted: bool,
}

/// A record of a changed `KeyValue` entry.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ChangedKey {
    /// The namespace of the key.
    pub namespace: Option<String>,

    /// The key that was changed.
    pub key: String,

    /// True if the key was deleted.
    pub deleted: bool,
}