Struct PgStores 

pub struct PgStores { /* private fields */ }

Every Turnframe store over one PostgreSQL pool.

Implements ConversationStore, InteractionStore, CommandJournal, EventJournal, OutboxStore, ReplayStore and CommitStore, so a card written through the interaction trait is the card a commit bundle settles.

Cloning shares the pool; it never opens new connections.

Implementations

impl PgStores

pub async fn commit_in( &self, conn: &mut PgConnection, account: &AccountId, bundle: CommitBundle, ) -> Result<CommitReceipt, StoreError>

Applies a bundle inside a transaction the caller owns.

This is the seam an adopter needs when the domain tables live in the same database as the stores: begin one transaction, run the workflow executor’s own writes on it, hand it to this method, and commit once. The journal admission still happens before execution, so recovery works the same way if the process dies — the transaction simply removes the window in which the domain state and the bookkeeping could disagree.

Nothing is committed here. Commit the transaction to make the bundle visible; drop it to discard the bundle along with your own writes.

use turnframe_core::ids::AccountId;
use turnframe_store::commit::CommitBundle;
use turnframe_store_postgres::PgStores;

let mut transaction = store.pool().begin().await?;
// ... the executor's own writes go here, on the same transaction ...
let receipt = store
    .commit_in(&mut transaction, &AccountId::from("aurora"), bundle)
    .await?;
transaction.commit().await?;

Errors

Any error an item would raise through its own store trait, plus the refusals of CommitBundle::validate.

impl PgStores

pub async fn connect(url: &str) -> Result<Self, StoreError>

Opens a pool on url with the default PgStoreConfig.

The URL is a standard PostgreSQL connection string (postgres://user:password@host:port/database). It is never stored in a field this type can print.

Errors

• Other when the URL cannot be parsed, Unavailable when the first connection cannot be opened.

pub async fn connect_with( url: &str, config: &PgStoreConfig, ) -> Result<Self, StoreError>

Opens a pool on url with config.

Errors

• Other when the URL cannot be parsed, Unavailable when the first connection cannot be opened.

pub fn from_pool(pool: PgPool) -> Self

Uses a pool the caller already has.

Nothing is configured on it: the pool’s own settings, its search_path and its statement timeout are whatever the caller set. Use this to share one pool with the rest of an application, so the store and the domain tables are reached through the same connections and can be enlisted in the same transaction (see Self::commit_in).

pub fn with_schema(self, schema: impl Into<String>) -> Result<Self, ConfigError>

Puts the tables of a caller-supplied pool in schema.

Only tells Self::migrate which schema to create; the pool must already resolve to it, normally through a search_path in its connect options.

Errors

• ConfigError::InvalidSchemaName when the name is not a plain unquoted identifier.

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

Applies every migration this crate carries, creating the configured schema first when there is one.

It is safe to call on every start-up: sqlx records applied versions and skips them, and the statements are written to be applied twice anyway. Two processes racing it is safe too — the migrator takes a database-wide advisory lock — though a deployment normally runs it once, before the instances that will use the schema come up.

Errors

• MigrateError when the schema cannot be created, a migration fails, or an already-applied migration no longer matches its recorded checksum.

pub fn pool(&self) -> &PgPool

The pool underneath, for health checks, metrics, or statements this adapter does not make.

pub fn schema(&self) -> Option<&str>

The schema the tables live in, when one was configured.

pub async fn close(&self)

Closes the pool and waits for its connections to be released.

pub fn stores(&self) -> Result<Stores, StoresBuilderError>

Bundles this store behind all seven traits.

Errors

• StoresBuilderError — never in practice. Every role is supplied here, and the Result exists only because turnframe-store offers no infallible “one backend for every role” constructor for a backend other than its own in-memory one.

Trait Implementations

impl Clone for PgStores

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl CommandJournalReader for PgStores

fn get<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, command_id: &'life2 CommandId, ) -> Pin<Box<dyn Future<Output = Result<CommandJournalEntry, StoreError>> + Send + 'async_trait>>

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

Loads one entry.

fn for_turn<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<Vec<CommandJournalEntry>, StoreError>> + Send + 'async_trait>>

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

Every entry of a turn, ordered by created_at then command_id.

fn pending_for_turn<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<Vec<CommandJournalEntry>, StoreError>> + Send + 'async_trait>>

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

Entries of a turn whose status is Pending or Executing, ordered by created_at then command_id (spec §23.1: resume by idempotency key).

impl CommandJournalWriter for PgStores

fn begin<'life0, 'async_trait>( &'life0 self, entry: CommandJournalEntry, ) -> Pin<Box<dyn Future<Output = Result<JournalAdmission, StoreError>> + Send + 'async_trait>>

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

Admits a command under UNIQUE (account_id, idempotency_key).

fn mark_executing<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, command_id: &'life2 CommandId, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Moves Pending → Executing. Calling it on an Executing entry is accepted without change.

fn complete<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, command_id: &'life2 CommandId, outcome: JournalOutcome, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Records the outcome and moves the entry to JournalOutcome::status. Recording the same outcome again on an entry already in that status is accepted without change.

fn fail<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, account: &'life1 AccountId, command_id: &'life2 CommandId, error: &'life3 ExecutionError, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Records an execution error as the outcome, through JournalOutcome::from_execution_error. Same rules as Self::complete.

impl CommitStore for PgStores

fn commit<'life0, 'life1, 'async_trait>( &'life0 self, account: &'life1 AccountId, bundle: CommitBundle, ) -> Pin<Box<dyn Future<Output = Result<CommitReceipt, StoreError>> + Send + 'async_trait>>

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

Applies the bundle for account atomically.

impl ConversationReader for PgStores

fn load_conversation<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, id: &'life2 ConversationId, ) -> Pin<Box<dyn Future<Output = Result<ConversationRecord, StoreError>> + Send + 'async_trait>>

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

Loads a conversation of account.

fn load_recent_turns<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, conversation: &'life2 ConversationId, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<StoredTurn>, StoreError>> + Send + 'async_trait>>

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

Loads the most recent limit turns of a conversation in chronological order (oldest of the selected first). Ordering is by received_at, then turn_id.

fn load_turn<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<StoredTurn, StoreError>> + Send + 'async_trait>>

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

Loads one turn.

fn turn_phase<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<TurnPhaseMarker, StoreError>> + Send + 'async_trait>>

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

Reads the phase marker of a turn.

fn list_unfinished_turns<'life0, 'async_trait>( &'life0 self, scope: RecoveryScope, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<TurnPhaseMarker>, StoreError>> + Send + 'async_trait>>

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

Lists up to limit markers of turns whose phase is not terminal, oldest first (by received_at, then turn_id).

impl ConversationWriter for PgStores

fn create_conversation<'life0, 'async_trait>( &'life0 self, record: ConversationRecord, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Creates a conversation.

fn append_user_turn<'life0, 'async_trait>( &'life0 self, turn: StoredUserTurn, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Appends a user turn to its conversation and creates its phase marker at TurnPhase::Received.

fn append_assistant_turn<'life0, 'life1, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn: AssistantTurn, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Persists the assistant turn that answers a user turn, exactly as returned to the client.

fn set_turn_phase<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, phase: TurnPhase, ) -> Pin<Box<dyn Future<Output = Result<TurnPhaseMarker, StoreError>> + Send + 'async_trait>>

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

Writes the phase marker of a turn and returns the updated marker.

impl Debug for PgStores

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Names the schema and the pool’s size, never the connection string: this output is safe to log.

impl EventJournalReader for PgStores

fn list_since<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, since: CaseRevision, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<StoredEvent>, StoreError>> + Send + 'async_trait>>

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

Events of one case with case_revision > since, in append order, at most limit. since = CaseRevision::ZERO lists the whole history.

fn read_from<'life0, 'life1, 'async_trait>( &'life0 self, account: &'life1 AccountId, after: EventCursor, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<EventPage, StoreError>> + Send + 'async_trait>>

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

The account’s next events after after in the journal’s total order, across every case, at most limit, as an EventPage.

fn get_by_ids<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, ids: &'life2 [EventId], ) -> Pin<Box<dyn Future<Output = Result<Vec<StoredEvent>, StoreError>> + Send + 'async_trait>>

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

The events with the given identifiers that exist for account, in the order requested; unknown or foreign identifiers are omitted. Receipt verification reads events back through this method (ADR-012 point 6).

fn count<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, ) -> Pin<Box<dyn Future<Output = Result<u64, StoreError>> + Send + 'async_trait>>

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

Number of events of a case.

impl EventJournalWriter for PgStores

fn append<'life0, 'async_trait>( &'life0 self, batch: EventBatch, ) -> Pin<Box<dyn Future<Output = Result<Vec<EventId>, StoreError>> + Send + 'async_trait>>

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

Appends a batch atomically and returns the event identifiers in order.

fn redact_payload<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, account: &'life1 AccountId, event_id: &'life2 EventId, authority: &'life3 RedactionAuthority, ) -> Pin<Box<dyn Future<Output = Result<EventRedaction, StoreError>> + Send + 'async_trait>>

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

Erases one event’s payload in place, and records that it happened.

impl InteractionReader for PgStores

fn get<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, id: &'life2 InteractionId, ) -> Pin<Box<dyn Future<Output = Result<InteractionRecord, StoreError>> + Send + 'async_trait>>

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

Loads an interaction of account.

fn list_open_for_conversation<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, conversation: &'life2 ConversationId, ) -> Pin<Box<dyn Future<Output = Result<Vec<Interaction>, StoreError>> + Send + 'async_trait>>

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

Lists the open (Active or Resolving) interactions of a conversation, ordered by created_at then id.

fn list_open_for_case<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, ) -> Pin<Box<dyn Future<Output = Result<Vec<Interaction>, StoreError>> + Send + 'async_trait>>

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

Lists the open (Active or Resolving) interactions of a case, ordered by created_at then id.

fn blocking_answered_at<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, revision: CaseRevision, ) -> Pin<Box<dyn Future<Output = Result<bool, StoreError>> + Send + 'async_trait>>

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

Whether a blocking card of case_key bound to revision has already been answered by the user.

impl InteractionWriter for PgStores

fn insert<'life0, 'async_trait>( &'life0 self, interaction: Interaction, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Inserts a new Active interaction.

fn insert_replacing_blocking<'life0, 'async_trait>( &'life0 self, interaction: Interaction, ) -> Pin<Box<dyn Future<Output = Result<Vec<InteractionId>, StoreError>> + Send + 'async_trait>>

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

Inserts a new Active interaction, invalidating the Active blocking occupant of the same case when there is one. Returns the identifiers it invalidated (empty when the slot was free or the new card is not blocking).

fn begin_resolution<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, id: &'life2 InteractionId, expected_status: InteractionStatus, option_id: OptionId, resolved_by: TurnId, ) -> Pin<Box<dyn Future<Output = Result<InteractionRecord, StoreError>> + Send + 'async_trait>>

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

Compare-and-swap expected_status → Resolving, recording the chosen option and the resolving turn. The only legal source status is Active; the explicit parameter makes a lost race visible to the caller instead of hiding it behind a re-read.

fn finish_resolution<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, id: &'life2 InteractionId, outcome: ResolutionOutcome, ) -> Pin<Box<dyn Future<Output = Result<InteractionRecord, StoreError>> + Send + 'async_trait>>

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

Settles a Resolving interaction.

fn invalidate_case_cards<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, reason: InvalidationReason, ) -> Pin<Box<dyn Future<Output = Result<Vec<InteractionId>, StoreError>> + Send + 'async_trait>>

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

Invalidates every Active interaction of the case, whatever revision it is bound to, and whether or not it declares revision independence. Returns the invalidated identifiers in list order.

fn invalidate_for_case<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, case_key: &'life2 CaseKey, new_revision: CaseRevision, reason: InvalidationReason, ) -> Pin<Box<dyn Future<Output = Result<Vec<InteractionId>, StoreError>> + Send + 'async_trait>>

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

Invalidates every Active interaction of the case that is bound to a revision other than new_revision and does not declare revision independence. Returns the invalidated identifiers in list order.

fn expire_due<'life0, 'async_trait>( &'life0 self, at: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<Vec<InteractionId>, StoreError>> + Send + 'async_trait>>

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

Moves every Active interaction whose expires_at <= now to Expired, across all accounts. Returns the expired identifiers in list order.

impl OutboxReader for PgStores

fn get<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, ) -> Pin<Box<dyn Future<Output = Result<OutboxRecord, StoreError>> + Send + 'async_trait>>

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

Loads one record.

fn list_for_command<'life0, 'life1, 'async_trait>( &'life0 self, command_id: &'life1 CommandId, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxRecord>, StoreError>> + Send + 'async_trait>>

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

Every record produced by a command, ordered by created_at then id.

impl OutboxWriter for PgStores

fn enqueue<'life0, 'async_trait>( &'life0 self, entry: OutboxEntry, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Enqueues a Pending entry.

fn claim_due<'life0, 'life1, 'async_trait>( &'life0 self, at: DateTime<Utc>, limit: usize, worker_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxEntry>, StoreError>> + Send + 'async_trait>>

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

Claims up to limit due Pending entries for worker_id (see the module documentation), oldest first by created_at then id, and returns them already in Dispatching with attempt_count incremented.

fn mark_completed<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Dispatching | OutcomeUnknown → Completed; clears the claim and stamps completed_at. Accepted without change when already Completed.

fn mark_failed<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, reason: String, retry_at: Option<DateTime<Utc>>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Records a failure. reason is a stable code, never free text, and is stored on the record either way.

fn mark_outcome_unknown<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, remote_ref: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Dispatching → OutcomeUnknown, releasing the claim and recording the remote reference when the remote returned one (I15).

fn reschedule<'life0, 'life1, 'async_trait>( &'life0 self, outbox_id: &'life1 OutboxId, next_attempt_at: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Returns the entry to Pending with next_attempt_at, releasing any claim. Legal from Dispatching, OutcomeUnknown and Pending (which only moves the time).

fn release_expired_claims<'life0, 'async_trait>( &'life0 self, claimed_before: DateTime<Utc>, ) -> Pin<Box<dyn Future<Output = Result<Vec<OutboxId>, StoreError>> + Send + 'async_trait>>

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

Releases every Dispatching entry whose claim was taken strictly before claimed_before back to Pending, due immediately. Returns the released identifiers. Meant for a reaper that recovers entries a crashed dispatcher left behind.

impl ReplayReader for PgStores

fn get<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, turn_id: &'life2 TurnId, ) -> Pin<Box<dyn Future<Output = Result<ReplayRecord, StoreError>> + Send + 'async_trait>>

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

Loads the record of a turn.

fn list_for_conversation<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, account: &'life1 AccountId, conversation: &'life2 ConversationId, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ReplayRecord>, StoreError>> + Send + 'async_trait>>

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

The most recent limit records of a conversation in chronological order (by recorded_at, then turn_id).

impl ReplayWriter for PgStores

fn put<'life0, 'async_trait>( &'life0 self, record: ReplayRecord, ) -> Pin<Box<dyn Future<Output = Result<(), StoreError>> + Send + 'async_trait>>

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

Inserts or replaces the record of (record.account_id, record.turn_id).