Struct PostgresEventStore

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

PostgreSQL event store implementation

TODO: The serialization_format configuration is currently stored but not used. The PostgreSQL adapter still uses hardcoded JSON serialization. Implementing pluggable serialization formats requires changes to the database schema and query logic to handle binary data instead of JSON columns.

Implementations

impl<E> PostgresEventStore<E>

where E: Serialize + for<'de> Deserialize<'de> + Send + Sync + Debug + Clone + PartialEq + Eq + 'static,

pub async fn read_streams_paginated_impl( &self, stream_ids: &[StreamId], options: &ReadOptions, continuation_token: Option<EventId>, ) -> Result<(Vec<StoredEvent<E>>, Option<EventId>), EventStoreError>

Read streams with pagination support for large result sets Returns events and a continuation token for the next page

impl<E> PostgresEventStore<E>

where E: Serialize + for<'de> Deserialize<'de> + Send + Sync + Debug + Clone + PartialEq + Eq + 'static,

pub async fn new(config: PostgresConfig) -> Result<Self, PostgresError>

Create a new PostgreSQL event store with the given configuration

pub async fn new_with_retry_strategy( config: PostgresConfig, retry_strategy: RetryStrategy, ) -> Result<Self, PostgresError>

Create a new PostgreSQL event store with custom retry strategy

pub fn pool(&self) -> &PgPool

Get a reference to the connection pool

pub const fn config(&self) -> &PostgresConfig

Get the configuration

pub async fn migrate(&self) -> Result<(), PostgresError>

Run database migrations

pub async fn initialize(&self) -> Result<(), PostgresError>

Initialize the database schema

This method creates the necessary tables and indexes for the event store. It is idempotent and can be called multiple times safely. Uses PostgreSQL advisory locks to prevent concurrent initialization conflicts.

pub async fn health_check(&self) -> Result<HealthStatus, PostgresError>

Check database connectivity with comprehensive health checks

pub async fn recover_connection(&self) -> Result<(), PostgresError>

Attempt to recover from connection issues

pub fn get_pool_metrics(&self) -> PoolMetrics

Get current pool metrics

pub fn monitor(&self) -> Arc<PoolMonitor>

Get pool monitor for advanced monitoring setup

pub fn start_pool_monitoring(&self) -> (JoinHandle<()>, Sender<bool>)

Start background pool monitoring task

pub async fn read_paginated( &self, stream_ids: &[StreamId], options: &ReadOptions, continuation_token: Option<EventId>, ) -> Result<(Vec<StoredEvent<E>>, Option<EventId>), EventStoreError>

where E: Serialize + for<'de> Deserialize<'de> + Send + Sync + Clone + Debug + PartialEq + Eq + 'static,

Read events from multiple streams with pagination support.

This method is designed for efficiently processing large result sets without loading all events into memory at once.

Arguments

• stream_ids - The streams to read from
• options - Read options (version filters, max events)
• continuation_token - Optional continuation token from previous page

Returns

A tuple containing:

• Vector of events for this page
• Optional continuation token for the next page (None if no more results)

Example

let stream_ids = vec![StreamId::try_new("stream-1")?, StreamId::try_new("stream-2")?];
let options = ReadOptions::default();

let mut continuation = None;
loop {
    let (events, next_token) = store.read_paginated(&stream_ids, &options, continuation).await?;
     
    // Process events for this page
    for event in events {
        println!("Processing event: {:?}", event.event_id);
    }
     
    // Check if there are more pages
    match next_token {
        Some(token) => continuation = Some(token),
        None => break, // No more pages
    }
}

Trait Implementations

impl<E> Clone for PostgresEventStore<E>

where E: Serialize + for<'de> Deserialize<'de> + Send + Sync + Debug + Clone + PartialEq + Eq + 'static,

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<E> EventStore for PostgresEventStore<E>

where E: Serialize + for<'de> Deserialize<'de> + Send + Sync + Debug + Clone + PartialEq + Eq + 'static,

type Event = E

The event type this store handles.

fn read_streams<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, stream_ids: &'life1 [StreamId], options: &'life2 ReadOptions, ) -> Pin<Box<dyn Future<Output = Result<StreamData<Self::Event>, EventStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Reads events from multiple streams.

fn write_events_multi<'life0, 'async_trait>( &'life0 self, stream_events: Vec<StreamEvents<Self::Event>>, ) -> Pin<Box<dyn Future<Output = Result<HashMap<StreamId, EventVersion>, EventStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Writes events to multiple streams atomically.

fn stream_exists<'life0, 'life1, 'async_trait>( &'life0 self, stream_id: &'life1 StreamId, ) -> Pin<Box<dyn Future<Output = Result<bool, EventStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Checks if a stream exists.

fn get_stream_version<'life0, 'life1, 'async_trait>( &'life0 self, stream_id: &'life1 StreamId, ) -> Pin<Box<dyn Future<Output = Result<Option<EventVersion>, EventStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Gets the current version of a stream.

fn subscribe<'life0, 'async_trait>( &'life0 self, options: SubscriptionOptions, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn Subscription<Event = Self::Event>>, EventStoreError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Creates a subscription to events.