eventlogs 0.2.0

High-Performance, Batteries-Included, Event Sourcing for Rust
Documentation 
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

use crate::ids::LogId;
use crate::{AppendOptions, EventRecord};
use futures_util::Stream;
use thiserror::Error;

/// Fake implementations for testing purposes.
pub mod fake;

/// A [EventStore] backed by a Postgres database.
#[cfg(feature = "postgres-store")]
pub mod postgres;

/// The error type returned from [EventStore] methods.
#[derive(Debug, Error)]
pub enum EventStoreError {
    /// An event with the specified index already exists in the specified log.
    #[error("log_id={log_id} already has an event_index={event_index}")]
    EventIndexAlreadyExists { log_id: LogId, event_index: u32 },
    /// An event with the specified idempotency key already exists.
    #[error("an event with the provided idempotency key already exists")]
    IdempotentReplay {
        idempotency_key: String,
        log_id: LogId,
        event_index: u32,
    },
    /// An unexpected database or connectivity error occurred.
    #[error("unexpected database error: {error}")]
    DatabaseError {
        error: Box<dyn std::error::Error + Sync + Send>,
    },
}

impl PartialEq for EventStoreError {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (
                Self::EventIndexAlreadyExists {
                    log_id: l_log_id,
                    event_index: l_event_index,
                },
                Self::EventIndexAlreadyExists {
                    log_id: r_log_id,
                    event_index: r_event_index,
                },
            ) => l_log_id == r_log_id && l_event_index == r_event_index,
            (
                Self::IdempotentReplay {
                    idempotency_key: l_idempotency_key,
                    log_id: l_log_id,
                    event_index: l_event_index,
                },
                Self::IdempotentReplay {
                    idempotency_key: r_idempotency_key,
                    log_id: r_log_id,
                    event_index: r_event_index,
                },
            ) => {
                l_idempotency_key == r_idempotency_key
                    && l_log_id == r_log_id
                    && l_event_index == r_event_index
            }
            (Self::DatabaseError { error: l_error }, Self::DatabaseError { error: r_error }) => {
                l_error.to_string() == r_error.to_string()
            }
            _ => false,
        }
    }
}

/// The trait implemented by all event stores.
#[trait_variant::make(Send)]
pub trait EventStore<E>
where
    E: Send,
{
    /// Appends an event to a log at the specified index.
    ///
    /// If an event already exists with that same index, this returns an
    /// [EventStoreError::EventIndexAlreadyExists] error.
    ///
    /// If an idempotency key is provided in the options, and an event already
    /// exists with that same key, this will return an
    /// [EventStoreError::IdempotentReplay] error.
    async fn append(
        &self,
        log_id: &LogId,
        event: &E,
        event_index: u32,
        append_options: &AppendOptions,
    ) -> Result<(), EventStoreError>;

    /// Returns an asynchronous stream of [EventRecord]s from the specified
    /// log ID, starting at the specified event index. Events must be ordered
    /// by index in ascending order.
    async fn load(
        &self,
        log_id: &LogId,
        starting_index: u32,
        max_events: u32,
    ) -> Result<impl Stream<Item = Result<impl EventRecord<E>, EventStoreError>>, EventStoreError>;
}