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

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 load_state<'life0, 'async_trait, S>( &'life0 self, correlation_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<Option<(S, i32)>>> + Send + 'async_trait>>

where S: for<'de> Deserialize<'de> + Send + 'async_trait, Self: 'async_trait, 'life0: 'async_trait,

Load state for workflow

fn save_state<'life0, 'life1, 'async_trait, S>( &'life0 self, correlation_id: Uuid, state: &'life1 S, expected_version: i32, ) -> Pin<Box<dyn Future<Output = Result<i32>> + Send + 'async_trait>>

where S: Serialize + Send + Sync + 'async_trait, Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Save state for workflow (optimistic locking)

fn insert_effect_intent<'life0, 'async_trait>( &'life0 self, event_id: Uuid, effect_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, ) -> 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<QueuedEffectExecution>>> + 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, effect_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, effect_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, effect_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, effect_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, effect_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_effect_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, ) -> 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_effect_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_effect_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 state after a handler error so retries can claim it again.