Struct PgEventStore

pub struct PgEventStore<E, S>
where
    S: Serde<E> + Send + Sync,
{ /* private fields */ }

PostgreSQL event store implementation.

Implementations

impl<E, S> PgEventStore<E, S>

where S: Serde<E> + Send + Sync, E: Event,

pub async fn new(pool: PgPool, serde: S) -> Result<Self, Error>

Initializes the PostgreSQL DB and returns a new instance of PgEventStore.

Arguments

• pool - The PostgreSQL connection pool.
• serde - The serialization implementation for the event payload.

pub fn new_uninitialized(pool: PgPool, serde: S) -> Self

Creates a new instance of PgEventStore.

This constructor does not initialize the database or add the domain_identifier columns necessary for disintegrate to function properly. If you need to initialize the database, use PgEventStore::new instead.

If you plan to use this constructor, ensure that the disintegrate is properly initialized. Refer to the SQL files in the “event_store/sql” directory to recreate the default structure. Additionally, all domain_identifier columns and their corresponding indexes must be created manually.

Arguments

• pool - The PostgreSQL connection pool.
• serde - The serialization implementation for the event payload.

pub fn with_max_appends_connections_percent(self, percentage: f64) -> Self

Limits the maximum number of concurrent appends based on the PostgreSQL connection pool.

By default, PgEventStore allows up to 50% of the available database connections to be used for concurrent appends. This method allows adjusting that limit.

The number of concurrent appends is determined by multiplying the total maximum connections of the PgPool by the specified percentage.

Arguments

• percentage - A floating-point number (0.0 to 1.0) representing the proportion of the total database connections allocated for concurrent appends.

Returns

Returns a modified PgEventStore instance with the updated append concurrency limit.

Trait Implementations

impl<E: Clone, S> Clone for PgEventStore<E, S>

where S: Serde<E> + Send + Sync + Clone,

fn clone(&self) -> PgEventStore<E, S>

Returns a duplicate of the value.

const fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<E, S> EventStore<i64, E> for PgEventStore<E, S>

where E: Event + Send + Sync, S: Serde<E> + Send + Sync,

Implementation of the event store using PostgreSQL.

This module provides the implementation of the EventStore trait for PgEventStore, allowing interaction with a PostgreSQL event store. It enables streaming events based on a query and appending new events to the event store.

fn stream<'a, QE>( &'a self, query: &'a StreamQuery<PgEventId, QE>, ) -> BoxStream<'a, Result<PersistedEvent<PgEventId, QE>, Self::Error>>

where QE: TryFrom<E> + Event + 'static + Clone + Send + Sync, <QE as TryFrom<E>>::Error: StdError + 'static + Send + Sync,

Streams events based on the provided query.

This function fetches events from the PostgreSQL event store based on the provided query. It constructs a SQL query using the SqlEventsCriteriaBuilder and executes the query using the sqlx crate. The fetched events are then converted into PersistedEvent instances and streamed as a boxed stream.

Arguments

• query - The stream query specifying the criteria for filtering events.

Returns

A Result containing a boxed stream of PersistedEvent that matches the query criteria, or an error of type Self::Error.

fn append<'life0, 'async_trait, QE>( &'life0 self, events: Vec<E>, query: StreamQuery<PgEventId, QE>, version: PgEventId, ) -> Pin<Box<dyn Future<Output = Result<Vec<PersistedEvent<PgEventId, E>>, Self::Error>> + Send + 'async_trait>>

where E: Clone + 'async_trait, QE: Event + Clone + Send + Sync + 'async_trait, Self: 'async_trait, 'life0: 'async_trait,

Appends new events to the event store.

This function inserts the provided events into the PostgreSQL event store by performing two separate inserts. First, it inserts the events into the event_sequence table to reclaim a set of IDs for the events. Then, it inserts the events into the event table along with their IDs, event types, domain identifiers, and payloads. Finally, it marks the event IDs as consumed in the event sequence table. If marking the event IDs as consumed fails (e.g., another process has already consumed the IDs), a conflict error is raised. This conflict indicates that the data retrieved by the query is stale, meaning that the events generated are no longer valid due to being generated from an old version of the event store.

Arguments

• events - A vector of events to be appended.
• query - The stream query specifying the criteria for filtering events.
• version - The ID of the last consumed event.

Returns

A Result containing a vector of PersistedEvent representing the appended events, or an error of type Self::Error.

fn append_without_validation<'life0, 'async_trait>( &'life0 self, events: Vec<E>, ) -> Pin<Box<dyn Future<Output = Result<Vec<PersistedEvent<PgEventId, E>>, Self::Error>> + Send + 'async_trait>>

where E: Clone + 'async_trait, Self: 'async_trait, 'life0: 'async_trait,

Appends a batch of events to the PostgreSQL-backed event store without verifying
whether new events have been added since the last read.

Arguments

• events - A vector of events to be appended.

Returns

A Result containing a vector of PersistedEvent representing the appended events, or an error of type Self::Error.

type Error = Error