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

use std::collections::HashMap;
use std::marker::PhantomData;

use async_trait::async_trait;
use cqrs_es::{Aggregate, AggregateContext, AggregateError, EventEnvelope, EventStore};
use sqlx::{Pool, Postgres};

use crate::event_repository::EventRepository;
use crate::snapshot_repository::SnapshotRepository;

/// Storage engine using a Postgres database backing.
/// This is an snapshot-sourced `EventStore`, meaning it uses the serialized aggregate as the
/// primary source of truth for the state of the aggregate.
///
/// The individual events are also persisted but are used only for updating queries.
///
/// For a event-sourced `EventStore` see [`PostgresStore`](struct.PostgresStore.html).
///
pub struct PostgresSnapshotStore<A: Aggregate> {
    repo: SnapshotRepository<A>,
    // TODO: combine event retrieval and remove EventRepository here
    event_repo: EventRepository<A>,
    _phantom: PhantomData<A>,
}

impl<A: Aggregate> PostgresSnapshotStore<A> {
    /// Creates a new `PostgresSnapshotStore` from the provided database connection pool,
    /// an `EventStore` used for configuring a new cqrs framework.
    ///
    /// ```ignore
    /// # use postgres_es::PostgresSnapshotStore;
    /// # use cqrs_es::CqrsFramework;
    /// let store = PostgresSnapshotStore::<MyAggregate>::new(pool);
    /// let cqrs = CqrsFramework::new(store, vec![]);
    /// ```
    pub fn new(pool: Pool<Postgres>) -> Self {
        let repo = SnapshotRepository::new(pool.clone());
        let event_repo = EventRepository::new(pool);
        PostgresSnapshotStore {
            repo,
            event_repo,
            _phantom: PhantomData,
        }
    }
}

#[async_trait]
impl<A: Aggregate> EventStore<A> for PostgresSnapshotStore<A> {
    type AC = PostgresSnapshotStoreAggregateContext<A>;

    async fn load(&self, aggregate_id: &str) -> Vec<EventEnvelope<A>> {
        // TODO: combine with event store
        match self.event_repo.get_events(aggregate_id).await {
            Ok(val) => val,
            Err(_err) => {
                // TODO: improved error handling
                Default::default()
            }
        }
    }
    async fn load_aggregate(&self, aggregate_id: &str) -> PostgresSnapshotStoreAggregateContext<A> {
        match self.repo.get_snapshot(aggregate_id).await {
            Ok(snapshot) => match snapshot {
                Some(snapshot) => {
                    let _tmp = serde_json::to_string(&snapshot.aggregate).unwrap();
                    snapshot
                }
                None => PostgresSnapshotStoreAggregateContext {
                    aggregate_id: aggregate_id.to_string(),
                    aggregate: Default::default(),
                    current_sequence: 0,
                    current_snapshot: 0,
                },
            },
            Err(e) => {
                panic!("{}", e);
            }
        }
    }

    async fn commit(
        &self,
        events: Vec<A::Event>,
        mut context: PostgresSnapshotStoreAggregateContext<A>,
        metadata: HashMap<String, String>,
    ) -> Result<Vec<EventEnvelope<A>>, AggregateError> {
        for event in events.clone() {
            context.aggregate.apply(event);
        }
        let aggregate_id = context.aggregate_id.clone();
        let wrapped_events =
            self.wrap_events(&aggregate_id, context.current_sequence, events, metadata);

        if context.current_sequence == 0 {
            self.repo
                .insert(context.aggregate, aggregate_id, 1, &wrapped_events)
                .await?;
        } else {
            self.repo
                .update(
                    context.aggregate,
                    aggregate_id,
                    context.current_snapshot + 1,
                    &wrapped_events,
                )
                .await?;
        }

        Ok(wrapped_events)
    }
}

/// Holds context for the snapshot-sourced implementation PostgresSnapshotStore.
/// This is only used internally within the `EventStore`.
#[derive(Debug, PartialEq)]
pub struct PostgresSnapshotStoreAggregateContext<A>
where
    A: Aggregate,
{
    pub(crate) aggregate_id: String,
    pub(crate) aggregate: A,
    pub(crate) current_sequence: usize,
    pub(crate) current_snapshot: usize,
}

impl<A> AggregateContext<A> for PostgresSnapshotStoreAggregateContext<A>
where
    A: Aggregate,
{
    fn aggregate(&self) -> &A {
        &self.aggregate
    }
}