Struct PostgresStore 

pub struct PostgresStore { /* private fields */ }

PostgreSQL implementation of Store trait

Implementations

impl PostgresStore

pub fn new(pool: PgPool) -> Self

pub fn pool(&self) -> &PgPool

pub async fn is_processed(&self, event_id: Uuid) -> Result<bool>

Check if an event has already been processed (for idempotency).

This is used by the Kafka backend to prevent duplicate processing when Kafka redelivers messages.

Trait Implementations

impl Clone for PostgresStore

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl InsightStore for PostgresStore

fn subscribe_events<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn Stream<Item = InsightEvent> + Send + Unpin>>> + Send + 'async_trait>>

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

Subscribe to live events for all workflows

fn get_workflow_tree<'life0, 'async_trait>( &'life0 self, correlation_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<WorkflowTree>> + Send + 'async_trait>>

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

Get workflow tree for visualization

fn get_stats<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<InsightStats>> + Send + 'async_trait>>

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

Get aggregate statistics

fn get_recent_events<'life0, 'async_trait>( &'life0 self, cursor: Option<i64>, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<InsightEvent>>> + Send + 'async_trait>>

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

Get recent events with cursor-based pagination

fn get_effect_logs<'life0, 'async_trait>( &'life0 self, correlation_id: Option<Uuid>, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<EffectExecutionLog>>> + Send + 'async_trait>>

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

Get effect execution logs for operational debugging.

fn get_dead_letters<'life0, 'async_trait>( &'life0 self, unresolved_only: bool, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<DeadLetterEntry>>> + Send + 'async_trait>>

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

Get dead letter queue entries.

fn get_failed_workflows<'life0, 'async_trait>( &'life0 self, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<FailedWorkflow>>> + Send + 'async_trait>>

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

Get workflows that currently have failures or dead letters.

impl Store for PostgresStore

fn publish<'life0, 'async_trait>( &'life0 self, event: QueuedEvent, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Publish event to queue

fn poll_next<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<QueuedEvent>>> + Send + 'async_trait>>

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

Poll next event (per-workflow FIFO with advisory locks)

fn ack<'life0, 'async_trait>( &'life0 self, id: i64, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Acknowledge event processing completion

fn nack<'life0, 'async_trait>( &'life0 self, id: i64, retry_after_secs: u64, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Nack event (failed, retry later)

fn commit_event_processing<'life0, 'async_trait>( &'life0 self, commit: EventProcessingCommit, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Atomically commit event processing side effects.

fn insert_effect_intent<'life0, 'async_trait>( &'life0 self, event_id: Uuid, handler_id: String, correlation_id: Uuid, event_type: String, event_payload: Value, parent_event_id: Option<Uuid>, batch_id: Option<Uuid>, batch_index: Option<i32>, batch_size: Option<i32>, execute_at: DateTime<Utc>, timeout_seconds: i32, max_attempts: i32, priority: i32, join_window_timeout_seconds: Option<i32>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Insert effect execution intent

fn poll_next_effect<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<QueuedHandlerExecution>>> + Send + 'async_trait>>

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

Poll next ready effect (priority-based)

fn complete_effect<'life0, 'async_trait>( &'life0 self, event_id: Uuid, handler_id: String, result: Value, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Mark effect execution as completed

fn complete_effect_with_events<'life0, 'async_trait>( &'life0 self, event_id: Uuid, handler_id: String, result: Value, emitted_events: Vec<EmittedEvent>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Complete effect and atomically publish emitted events (crash-safe)

fn fail_effect<'life0, 'async_trait>( &'life0 self, event_id: Uuid, handler_id: String, error: String, attempts: i32, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Mark effect execution as failed and schedule retry (store-defined backoff)

fn dlq_effect<'life0, 'async_trait>( &'life0 self, event_id: Uuid, handler_id: String, error: String, reason: String, attempts: i32, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Move effect to DLQ (permanently failed)

fn dlq_effect_with_events<'life0, 'async_trait>( &'life0 self, event_id: Uuid, handler_id: String, error: String, reason: String, attempts: i32, emitted_events: Vec<EmittedEvent>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Move effect to DLQ and atomically publish mapped terminal events.

fn subscribe_workflow_events<'life0, 'async_trait>( &'life0 self, correlation_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<Box<dyn Stream<Item = WorkflowEvent> + Send + Unpin>>> + Send + 'async_trait>>

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

Subscribe to events for a specific workflow via LISTEN/NOTIFY.

fn get_workflow_status<'life0, 'async_trait>( &'life0 self, correlation_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<WorkflowStatus>> + Send + 'async_trait>>

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

Get workflow status for a correlation ID

fn join_same_batch_append_and_maybe_claim<'life0, 'async_trait>( &'life0 self, join_handler_id: String, correlation_id: Uuid, source_event_id: Uuid, source_event_type: String, source_payload: Value, source_created_at: DateTime<Utc>, batch_id: Uuid, batch_index: i32, batch_size: i32, join_window_timeout_seconds: Option<i32>, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<JoinEntry>>>> + Send + 'async_trait>>

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

Append a terminal item into a same-batch join window and attempt to claim the window for processing when complete.

fn join_same_batch_complete<'life0, 'async_trait>( &'life0 self, join_handler_id: String, correlation_id: Uuid, batch_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Mark a claimed same-batch window as completed and clear durable join rows.

fn join_same_batch_release<'life0, 'async_trait>( &'life0 self, join_handler_id: String, correlation_id: Uuid, batch_id: Uuid, error: String, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Release a claimed same-batch window back to open status after a handler error so retries can claim it again.

fn expire_same_batch_windows<'life0, 'async_trait>( &'life0 self, now: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<Vec<ExpiredJoinWindow>>> + Send + 'async_trait>>

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

Expire same-batch windows that have exceeded their configured timeout.