//! Storage abstraction separating the pure decision logic from IO. //! //! The [`Store`] trait composes focused sub-traits, each a dumb instruction //! follower: writes apply one update and report an affected-row count (or an I/O //! error). The saga, not the store, interprets counts and owns idempotency and //! compensation. //! - [`AccountStore`] — account CRUD and versioning //! - [`PostingStore`] — posting reads and lifecycle transitions //! - [`TransferStore`] — transfer persistence and queries //! - [`SagaStore`] — saga state for crash recovery //! - [`EventStore`] — the ledger event log //! - [`BookStore`] — book persistence use async_trait::async_trait; use kuatia_types::{ Account, AccountId, AssetId, Book, BookId, Envelope, EnvelopeId, Posting, PostingId, PostingStatus, Receipt, ReservationId, }; use crate::error::StoreError; use crate::events::EventStore; /// Pairs a committed transfer with its receipt. #[derive(Debug, Clone)] pub struct EnvelopeRecord { /// The envelope that was committed. pub envelope: Envelope, /// The receipt proving commitment. pub receipt: Receipt, /// Unix milliseconds when this record was created. pub created_at: i64, } /// Pagination and filtering parameters for posting queries. #[derive(Debug, Clone)] pub struct PostingQuery { /// Filter to postings owned by this account. pub account: AccountId, /// Filter by asset. pub asset: Option, /// Filter by posting status. pub status: Option, /// Max results to return. pub limit: Option, /// Number of results to skip. pub offset: Option, } /// Pagination and filtering parameters for transfer queries. #[derive(Debug, Clone, Default)] pub struct TransferQuery { /// Filter to transfers involving this account. pub account: Option, /// Inclusive lower bound (unix millis). pub from_ts: Option, /// Exclusive upper bound (unix millis). pub to_ts: Option, /// Filter by book. pub book: Option, /// Max results to return. pub limit: Option, /// Number of results to skip. pub offset: Option, } /// A page of results with total count for pagination. #[derive(Debug, Clone)] pub struct Page { /// The items in this page. pub items: Vec, /// Total number of matching items (before pagination). pub total: u64, } // --------------------------------------------------------------------------- // Sub-traits // --------------------------------------------------------------------------- /// Account persistence: create, version, query. #[async_trait] pub trait AccountStore: Send + Sync { /// Fetch a single account by id. async fn get_account(&self, id: &AccountId) -> Result; /// Fetch multiple accounts by id. async fn get_accounts(&self, ids: &[AccountId]) -> Result, StoreError>; /// Persist a new account (version 1). async fn create_account(&self, account: Account) -> Result<(), StoreError>; /// Append a new version to an existing account. async fn append_account_version(&self, account: Account) -> Result<(), StoreError>; /// Return the full version history for an account. async fn get_account_history(&self, id: &AccountId) -> Result, StoreError>; /// List all accounts (latest version of each). async fn list_accounts(&self) -> Result, StoreError>; } /// Posting persistence: reads and lifecycle transitions. #[async_trait] pub trait PostingStore: Send + Sync { /// Fetch postings by their ids. async fn get_postings(&self, ids: &[PostingId]) -> Result, StoreError>; /// Return postings owned by an account, optionally filtered by asset and/or status. async fn get_postings_by_account( &self, account: &AccountId, asset: Option<&AssetId>, status: Option, ) -> Result, StoreError>; /// Reserve postings: `Active → PendingInactive`, stamping `reservation` as /// the owner token. A dumb instruction — each id flips only if still `Active`; /// returns the **number of rows reserved** (0 ≤ n ≤ ids.len()). It does not /// error on a short count; the caller (saga) interprets it. async fn reserve_postings( &self, ids: &[PostingId], reservation: ReservationId, ) -> Result; /// Release postings: `PendingInactive` owned by `reservation` → `Active`, /// clearing the owner. A dumb instruction — only postings reserved by this /// `reservation` flip; returns the **number of rows released**. Releasing an /// `Active` (already released) or differently-owned posting simply does not /// count. The caller interprets the result. async fn release_postings( &self, ids: &[PostingId], reservation: ReservationId, ) -> Result; /// Deactivate postings: flip to `Inactive`. A dumb instruction — it applies /// the conditional update and returns the **number of rows changed**; it does /// not decide whether that count is correct. The caller (saga) interprets it. /// - `reservation == None` (raw): only postings still `Active` flip. /// - `reservation == Some(rid)`: only postings `PendingInactive` owned by /// `rid` flip. /// Returns the count of postings actually transitioned (0 ≤ n ≤ ids.len()). async fn deactivate_postings( &self, ids: &[PostingId], reservation: Option, ) -> Result; /// Insert postings if absent (idempotent). A dumb instruction — inserts each /// posting unless one with the same id already exists, and returns the /// **number of rows inserted** (already-present postings contribute 0). The /// caller decides what a short count means. async fn insert_postings(&self, postings: &[Posting]) -> Result; /// Query postings with filtering and pagination. async fn query_postings(&self, query: &PostingQuery) -> Result, StoreError> { let all = self .get_postings_by_account(&query.account, query.asset.as_ref(), query.status) .await?; let total = all.len() as u64; let offset = query.offset.unwrap_or(0) as usize; let limit = query.limit.unwrap_or(u32::MAX) as usize; let items = all.into_iter().skip(offset).take(limit).collect(); Ok(Page { items, total }) } } /// Transfer persistence: store and query committed transfers. #[async_trait] pub trait TransferStore: Send + Sync { /// Fetch a transfer record by its content-addressed id. async fn get_transfer(&self, id: &EnvelopeId) -> Result, StoreError>; /// Persist a transfer record if absent (idempotent) and index it under every /// account in `involved` (both created and consumed owners — the caller /// supplies the set so storage computes nothing). A dumb instruction: /// returns **1** if the transfer row was newly inserted, **0** if it already /// existed. The caller decides what `0` means. async fn store_transfer( &self, record: EnvelopeRecord, involved: &[AccountId], ) -> Result; /// Return all transfers involving the given account. async fn get_transfers_for_account( &self, account: &AccountId, ) -> Result, StoreError>; /// Query transfers with filtering and pagination. async fn query_transfers( &self, query: &TransferQuery, ) -> Result, StoreError> { // Default in-memory implementation let all = if let Some(ref account) = query.account { self.get_transfers_for_account(account).await? } else { return Err(StoreError::Internal( "query_transfers requires account filter in default implementation".into(), )); }; let filtered: Vec = all .into_iter() .filter(|r| { if let Some(from) = query.from_ts && r.created_at < from { return false; } if let Some(to) = query.to_ts && r.created_at >= to { return false; } if let Some(book) = query.book && r.envelope.book() != book { return false; } true }) .collect(); let total = filtered.len() as u64; let offset = query.offset.unwrap_or(0) as usize; let limit = query.limit.unwrap_or(u32::MAX) as usize; let items = filtered.into_iter().skip(offset).take(limit).collect(); Ok(Page { items, total }) } } /// Saga state persistence for crash recovery. #[async_trait] pub trait SagaStore: Send + Sync { /// Persist a saga execution state. async fn save_saga(&self, id: &i64, data: Vec) -> Result<(), StoreError>; /// Load all pending (incomplete) saga states. async fn list_pending_sagas(&self) -> Result)>, StoreError>; /// Delete a completed saga state. async fn delete_saga(&self, id: &i64) -> Result<(), StoreError>; } /// Book persistence. #[async_trait] pub trait BookStore: Send + Sync { /// Create a new book. async fn create_book(&self, book: Book) -> Result<(), StoreError>; /// Fetch a book by id. async fn get_book(&self, id: &BookId) -> Result; /// List all books. async fn list_books(&self) -> Result, StoreError>; } // --------------------------------------------------------------------------- // Composite trait // --------------------------------------------------------------------------- /// Async storage abstraction composing all sub-traits. pub trait Store: AccountStore + PostingStore + TransferStore + SagaStore + EventStore + BookStore { } impl Store for T { }