|
|
@@ -9,1481 +9,48 @@
|
|
|
//! let store = SqlStore::new(pool);
|
|
|
//! store.migrate().await?;
|
|
|
//! ```
|
|
|
+//!
|
|
|
+//! The [`Store`](kuatia_storage::store::Store) sub-traits are each implemented in
|
|
|
+//! their own module (`account`, `posting`, `transfer`, `saga`, `event`, `book`,
|
|
|
+//! `projection`); shared row mappers and codecs live in `row`, the schema
|
|
|
+//! migrations in `migrate`, and the one SQLite/PostgreSQL divergence behind the
|
|
|
+//! `Dialect` seam in `dialect`.
|
|
|
|
|
|
-use std::collections::{HashMap, HashSet};
|
|
|
-use std::str::FromStr;
|
|
|
-use std::sync::atomic::{AtomicU8, Ordering};
|
|
|
-
|
|
|
-use async_trait::async_trait;
|
|
|
-use sqlx::any::AnyRow;
|
|
|
-use sqlx::{Any, Pool, Row};
|
|
|
+use sqlx::{Any, Pool};
|
|
|
|
|
|
-use kuatia_storage::error::StoreError;
|
|
|
-use kuatia_storage::events::{EventStore, LedgerEvent, event_dedup_key};
|
|
|
-use kuatia_storage::query::{filter_transfers, paginate};
|
|
|
-use kuatia_storage::store::*;
|
|
|
use kuatia_types::autoid::AutoId;
|
|
|
-use kuatia_types::*;
|
|
|
|
|
|
-// Cached backend kind for `SqlStore::backend`.
|
|
|
-const BACKEND_UNKNOWN: u8 = 0;
|
|
|
-const BACKEND_POSTGRES: u8 = 1;
|
|
|
-const BACKEND_SQLITE: u8 = 2;
|
|
|
+use crate::dialect::Dialect;
|
|
|
|
|
|
-/// Row-locking clause appended to a `SELECT` on backends that support it
|
|
|
-/// (PostgreSQL). SQLite has no `FOR UPDATE` and serializes writers itself, so it
|
|
|
-/// gets an empty clause.
|
|
|
-const FOR_UPDATE: &str = " FOR UPDATE";
|
|
|
+mod account;
|
|
|
+mod book;
|
|
|
+mod dialect;
|
|
|
+mod event;
|
|
|
+mod migrate;
|
|
|
+mod posting;
|
|
|
+mod projection;
|
|
|
+mod row;
|
|
|
+mod saga;
|
|
|
+mod transfer;
|
|
|
|
|
|
-/// SQL-backed [`Store`] implementation.
|
|
|
+/// SQL-backed [`Store`](kuatia_storage::store::Store) implementation.
|
|
|
pub struct SqlStore {
|
|
|
pool: Pool<Any>,
|
|
|
autoid: AutoId,
|
|
|
- /// Detected backend kind (lazily probed): one of `BACKEND_*`.
|
|
|
- backend: AtomicU8,
|
|
|
+ /// Which backend this store talks to; resolved once at construction.
|
|
|
+ dialect: Dialect,
|
|
|
}
|
|
|
|
|
|
impl SqlStore {
|
|
|
- /// Create a new SQL store wrapping an existing connection pool.
|
|
|
+ /// Create a new SQL store wrapping an existing connection pool. The backend
|
|
|
+ /// dialect is resolved from the pool's connection URL, so no query is issued
|
|
|
+ /// here; call [`migrate`](Self::migrate) next to apply the schema.
|
|
|
pub fn new(pool: Pool<Any>) -> Self {
|
|
|
+ let dialect = Dialect::from_pool(&pool);
|
|
|
Self {
|
|
|
pool,
|
|
|
autoid: AutoId::new(),
|
|
|
- backend: AtomicU8::new(BACKEND_UNKNOWN),
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- /// Whether the backend is PostgreSQL. Probed once and cached: `SELECT
|
|
|
- /// sqlite_version()` succeeds only on SQLite, so a failure means Postgres.
|
|
|
- async fn is_postgres(&self) -> Result<bool, StoreError> {
|
|
|
- match self.backend.load(Ordering::Relaxed) {
|
|
|
- BACKEND_POSTGRES => return Ok(true),
|
|
|
- BACKEND_SQLITE => return Ok(false),
|
|
|
- _ => {}
|
|
|
- }
|
|
|
- let is_sqlite = sqlx::query("SELECT sqlite_version()")
|
|
|
- .fetch_optional(&self.pool)
|
|
|
- .await
|
|
|
- .is_ok();
|
|
|
- self.backend.store(
|
|
|
- if is_sqlite {
|
|
|
- BACKEND_SQLITE
|
|
|
- } else {
|
|
|
- BACKEND_POSTGRES
|
|
|
- },
|
|
|
- Ordering::Relaxed,
|
|
|
- );
|
|
|
- Ok(!is_sqlite)
|
|
|
- }
|
|
|
-
|
|
|
- /// The row-locking clause for the current backend: [`FOR_UPDATE`] on
|
|
|
- /// Postgres, empty on SQLite.
|
|
|
- async fn lock_clause(&self) -> Result<&'static str, StoreError> {
|
|
|
- Ok(if self.is_postgres().await? {
|
|
|
- FOR_UPDATE
|
|
|
- } else {
|
|
|
- ""
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- /// Run database migrations. Idempotent: a `_migrations` ledger records what
|
|
|
- /// has been applied, so re-running is a no-op. Every column is a text type,
|
|
|
- /// so the store holds no opaque binary and the DDL is identical for both
|
|
|
- /// backends. Content-addressed ids and opaque saga bytes are stored as hex
|
|
|
- /// `TEXT`, and JSON payloads as their `TEXT` serialization, keeping every
|
|
|
- /// row legible for auditing.
|
|
|
- pub async fn migrate(&self) -> Result<(), StoreError> {
|
|
|
- sqlx::query("CREATE TABLE IF NOT EXISTS _migrations (name TEXT PRIMARY KEY)")
|
|
|
- .execute(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let migrations: &[(&str, &str)] = &[
|
|
|
- ("001_init", include_str!("migrations/001_init.sql")),
|
|
|
- (
|
|
|
- "002_subaccounts",
|
|
|
- include_str!("migrations/002_subaccounts.sql"),
|
|
|
- ),
|
|
|
- (
|
|
|
- "003_drop_user_data",
|
|
|
- include_str!("migrations/003_drop_user_data.sql"),
|
|
|
- ),
|
|
|
- (
|
|
|
- "004_index_tables",
|
|
|
- include_str!("migrations/004_index_tables.sql"),
|
|
|
- ),
|
|
|
- (
|
|
|
- "005_account_head",
|
|
|
- include_str!("migrations/005_account_head.sql"),
|
|
|
- ),
|
|
|
- (
|
|
|
- "006_drop_policy",
|
|
|
- include_str!("migrations/006_drop_policy.sql"),
|
|
|
- ),
|
|
|
- (
|
|
|
- "007_balance_projection",
|
|
|
- include_str!("migrations/007_balance_projection.sql"),
|
|
|
- ),
|
|
|
- ];
|
|
|
-
|
|
|
- for (name, sql) in migrations {
|
|
|
- let applied = sqlx::query("SELECT 1 FROM _migrations WHERE name = $1")
|
|
|
- .bind(*name)
|
|
|
- .fetch_optional(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- if applied.is_some() {
|
|
|
- continue;
|
|
|
- }
|
|
|
-
|
|
|
- // Apply every statement and record the migration in one transaction,
|
|
|
- // so a crash mid-migration rolls back cleanly and the migration is
|
|
|
- // retried as a whole. Migration 004 drops and rebuilds `postings`;
|
|
|
- // without the transaction a partial apply would leave the schema in a
|
|
|
- // state the migration cannot be re-run against. Both SQLite and
|
|
|
- // PostgreSQL support transactional DDL.
|
|
|
- let mut tx = self
|
|
|
- .pool
|
|
|
- .begin()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- for statement in sql.split(';') {
|
|
|
- let trimmed = statement.trim();
|
|
|
- if !trimmed.is_empty() {
|
|
|
- sqlx::query(trimmed)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- sqlx::query("INSERT INTO _migrations (name) VALUES ($1)")
|
|
|
- .bind(*name)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- tx.commit()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- }
|
|
|
- Ok(())
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-// Serialization helpers
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-
|
|
|
-/// Serialize a value to a JSON string. Payload columns store JSON as `TEXT` so
|
|
|
-/// the database is directly readable for auditing; the ledger never queries
|
|
|
-/// into the JSON, so no binary or indexed representation is needed.
|
|
|
-fn serialize_json<T: serde::Serialize>(val: &T) -> Result<String, StoreError> {
|
|
|
- serde_json::to_string(val).map_err(|e| StoreError::Internal(format!("json serialization: {e}")))
|
|
|
-}
|
|
|
-
|
|
|
-fn deserialize_json<T: serde::de::DeserializeOwned>(s: &str) -> Result<T, StoreError> {
|
|
|
- serde_json::from_str(s).map_err(|e| StoreError::Internal(format!("bad json: {e}")))
|
|
|
-}
|
|
|
-
|
|
|
-/// Lower-case hex encoding. Binary identifiers (content-addressed hashes) and
|
|
|
-/// opaque saga bytes are stored as hex `TEXT` so a row is legible in any SQL
|
|
|
-/// client and matches the hex form used in logs and `Debug` output.
|
|
|
-fn to_hex(bytes: &[u8]) -> String {
|
|
|
- const HEX: &[u8; 16] = b"0123456789abcdef";
|
|
|
- let mut s = String::with_capacity(bytes.len() * 2);
|
|
|
- for &b in bytes {
|
|
|
- s.push(HEX[(b >> 4) as usize] as char);
|
|
|
- s.push(HEX[(b & 0x0f) as usize] as char);
|
|
|
- }
|
|
|
- s
|
|
|
-}
|
|
|
-
|
|
|
-fn from_hex(s: &str) -> Result<Vec<u8>, StoreError> {
|
|
|
- if s.len() % 2 != 0 {
|
|
|
- return Err(StoreError::Internal(format!("odd-length hex: {s:?}")));
|
|
|
- }
|
|
|
- (0..s.len())
|
|
|
- .step_by(2)
|
|
|
- .map(|i| {
|
|
|
- u8::from_str_radix(&s[i..i + 2], 16)
|
|
|
- .map_err(|e| StoreError::Internal(format!("bad hex: {e}")))
|
|
|
- })
|
|
|
- .collect()
|
|
|
-}
|
|
|
-
|
|
|
-fn envelope_id_to_hex(id: &EnvelopeId) -> String {
|
|
|
- to_hex(&id.0)
|
|
|
-}
|
|
|
-
|
|
|
-fn envelope_id_from_hex(s: &str) -> Result<EnvelopeId, StoreError> {
|
|
|
- let bytes = from_hex(s)?;
|
|
|
- let arr: [u8; 32] = bytes.as_slice().try_into().map_err(|_| {
|
|
|
- StoreError::Internal(format!("expected 32-byte id, got {} bytes", bytes.len()))
|
|
|
- })?;
|
|
|
- Ok(EnvelopeId(arr))
|
|
|
-}
|
|
|
-
|
|
|
-fn row_to_account(row: &AnyRow) -> Result<Account, StoreError> {
|
|
|
- let id: i64 = row
|
|
|
- .try_get("id")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let subaccount: i64 = row
|
|
|
- .try_get("subaccount")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let version: i64 = row
|
|
|
- .try_get("version")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let flags_bits: i32 = row
|
|
|
- .try_get("flags")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let book: i64 = row
|
|
|
- .try_get("book")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let metadata_json: String = row
|
|
|
- .try_get("metadata")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- Ok(Account {
|
|
|
- id: AccountId::with_sub(id, subaccount),
|
|
|
- version: version as u64,
|
|
|
- flags: AccountFlags::from_bits_truncate(flags_bits as u32),
|
|
|
- book: BookId::new(book),
|
|
|
- metadata: deserialize_json(&metadata_json)?,
|
|
|
- })
|
|
|
-}
|
|
|
-
|
|
|
-fn row_to_posting(row: &AnyRow) -> Result<Posting, StoreError> {
|
|
|
- let transfer_id: String = row
|
|
|
- .try_get("transfer_id")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let idx: i16 = row
|
|
|
- .try_get("idx")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let owner: i64 = row
|
|
|
- .try_get("owner")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let subaccount: i64 = row
|
|
|
- .try_get("subaccount")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let asset: i32 = row
|
|
|
- .try_get("asset")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let value: String = row
|
|
|
- .try_get("value")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let value = Cent::from_str(&value).map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- Ok(Posting {
|
|
|
- id: PostingId {
|
|
|
- transfer: envelope_id_from_hex(&transfer_id)?,
|
|
|
- index: idx as u16,
|
|
|
- },
|
|
|
- owner: AccountId::with_sub(owner, subaccount),
|
|
|
- asset: AssetId::new(asset as u32),
|
|
|
- value,
|
|
|
- })
|
|
|
-}
|
|
|
-
|
|
|
-/// The FROM source for a posting read of the given derived state. Each index
|
|
|
-/// table carries a full row copy, so the live-set reads target the index table
|
|
|
-/// directly with no merge back to the immutable `postings` record. `Live` is a
|
|
|
-/// `UNION ALL` of the two disjoint live sets (the shared 6 data columns), still
|
|
|
-/// with no join to history. Portable across SQLite and PostgreSQL.
|
|
|
-fn filter_source(filter: PostingFilter) -> &'static str {
|
|
|
- match filter {
|
|
|
- PostingFilter::Active => "active_postings",
|
|
|
- PostingFilter::Reserved => "reserved_postings",
|
|
|
- PostingFilter::All => "postings",
|
|
|
- PostingFilter::Live => {
|
|
|
- "(SELECT transfer_id, idx, owner, subaccount, asset, value FROM active_postings \
|
|
|
- UNION ALL \
|
|
|
- SELECT transfer_id, idx, owner, subaccount, asset, value FROM reserved_postings) AS live"
|
|
|
- }
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-/// Maximum posting ids matched by a single statement. `id_predicate` expands to
|
|
|
-/// an `OR` of `n` equality pairs, so the binding constraint is SQLite's
|
|
|
-/// expression-tree depth limit (`SQLITE_MAX_EXPR_DEPTH`, default 1000), which a
|
|
|
-/// chain of `n` `OR`s reaches at roughly `n` deep. It caps well before the
|
|
|
-/// bind-parameter limits (SQLite 32766, PostgreSQL 65535) that `2 * n (+1)`
|
|
|
-/// parameters would hit. `500` stays comfortably under the expression-depth
|
|
|
-/// limit; callers that pass more ids are chunked, so the id-batch primitives
|
|
|
-/// have no practical ceiling on batch size.
|
|
|
-const MAX_IDS_PER_QUERY: usize = 500;
|
|
|
-
|
|
|
-/// Build a portable predicate matching a set of posting ids:
|
|
|
-/// `(transfer_id = $s AND idx = $s+1) OR (transfer_id = $s+2 AND idx = $s+3) ...`
|
|
|
-/// starting at placeholder `$start`. Row-value `IN ((a, b), ...)` is not
|
|
|
-/// portable across SQLite and PostgreSQL; an `OR` of equality pairs is. The
|
|
|
-/// caller binds each id as `(hex(transfer), idx as i16)` in order, matching the
|
|
|
-/// placeholder sequence. `ids` must be non-empty and no longer than
|
|
|
-/// [`MAX_IDS_PER_QUERY`]; larger sets are split into chunks by the caller.
|
|
|
-fn id_predicate(count: usize, start: u32) -> String {
|
|
|
- (0..count)
|
|
|
- .map(|i| {
|
|
|
- let p = start + (i as u32) * 2;
|
|
|
- format!("(transfer_id = ${} AND idx = ${})", p, p + 1)
|
|
|
- })
|
|
|
- .collect::<Vec<_>>()
|
|
|
- .join(" OR ")
|
|
|
-}
|
|
|
-
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-// AccountStore
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-
|
|
|
-#[async_trait]
|
|
|
-impl AccountStore for SqlStore {
|
|
|
- async fn get_account(&self, id: &AccountId) -> Result<Account, StoreError> {
|
|
|
- // The head points at the current version, so this is a single indexed
|
|
|
- // lookup into the immutable history — no scan of the version chain.
|
|
|
- let row = sqlx::query(
|
|
|
- "SELECT a.* FROM accounts a \
|
|
|
- JOIN account_head h \
|
|
|
- ON h.id = a.id AND h.subaccount = a.subaccount AND h.version = a.version \
|
|
|
- WHERE h.id = $1 AND h.subaccount = $2",
|
|
|
- )
|
|
|
- .bind(id.id)
|
|
|
- .bind(id.sub)
|
|
|
- .fetch_optional(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?
|
|
|
- .ok_or_else(|| StoreError::NotFound(format!("account {id:?}")))?;
|
|
|
- row_to_account(&row)
|
|
|
- }
|
|
|
-
|
|
|
- async fn get_accounts(&self, ids: &[AccountId]) -> Result<Vec<Account>, StoreError> {
|
|
|
- let mut result = Vec::with_capacity(ids.len());
|
|
|
- for id in ids {
|
|
|
- result.push(self.get_account(id).await?);
|
|
|
- }
|
|
|
- Ok(result)
|
|
|
- }
|
|
|
-
|
|
|
- async fn create_account(&self, account: Account) -> Result<u64, StoreError> {
|
|
|
- // Pessimistic locking: inside one transaction, lock the account's head
|
|
|
- // row with `SELECT ... FOR UPDATE` so a concurrent creator waits. The
|
|
|
- // head is the single row per account; its `ON CONFLICT (id, subaccount)
|
|
|
- // DO NOTHING` insert is the portable backstop that decides the winner
|
|
|
- // (SQLite has no `FOR UPDATE`, and it turns a concurrent double-create
|
|
|
- // into a clean affected-row count instead of a unique violation).
|
|
|
- let lock = self.lock_clause().await?;
|
|
|
- let mut tx = self
|
|
|
- .pool
|
|
|
- .begin()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let existing = sqlx::query(&format!(
|
|
|
- "SELECT 1 FROM account_head WHERE id = $1 AND subaccount = $2 LIMIT 1{lock}"
|
|
|
- ))
|
|
|
- .bind(account.id.id)
|
|
|
- .bind(account.id.sub)
|
|
|
- .fetch_optional(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- if existing.is_some() {
|
|
|
- return Ok(0);
|
|
|
- }
|
|
|
-
|
|
|
- // Append the immutable first version, then point the head at it.
|
|
|
- sqlx::query(
|
|
|
- "INSERT INTO accounts (id, subaccount, version, flags, book, metadata) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (id, subaccount, version) DO NOTHING"
|
|
|
- )
|
|
|
- .bind(account.id.id)
|
|
|
- .bind(account.id.sub)
|
|
|
- .bind(account.version as i64)
|
|
|
- .bind(account.flags.bits() as i32)
|
|
|
- .bind(account.book.0)
|
|
|
- .bind(serialize_json(&account.metadata)?)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let res = sqlx::query(
|
|
|
- "INSERT INTO account_head (id, subaccount, version) VALUES ($1, $2, $3) ON CONFLICT (id, subaccount) DO NOTHING",
|
|
|
- )
|
|
|
- .bind(account.id.id)
|
|
|
- .bind(account.id.sub)
|
|
|
- .bind(account.version as i64)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- if res.rows_affected() == 0 {
|
|
|
- return Ok(0);
|
|
|
- }
|
|
|
-
|
|
|
- tx.commit()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(1)
|
|
|
- }
|
|
|
-
|
|
|
- async fn append_account_version(&self, account: Account) -> Result<u64, StoreError> {
|
|
|
- // Pessimistic locking: inside one transaction, lock the account's head
|
|
|
- // row with `SELECT ... FOR UPDATE` so a concurrent appender waits here
|
|
|
- // until we commit, then check the version, append the new immutable row,
|
|
|
- // and move the head. `ON CONFLICT` is the portable backstop (SQLite has
|
|
|
- // no `FOR UPDATE`, and it covers the append phantom-insert a row lock
|
|
|
- // does not). The head is maintained by delete + insert, never `UPDATE`,
|
|
|
- // so the write path issues only inserts and deletes.
|
|
|
- let lock = self.lock_clause().await?;
|
|
|
- let mut tx = self
|
|
|
- .pool
|
|
|
- .begin()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- // A guarded write: no such account, or a version that is not exactly one
|
|
|
- // past the head, matches nothing and reports 0. This is what keeps the
|
|
|
- // chain gap-free (a stale or skipped version never lands) and makes a
|
|
|
- // replay of an already-applied version a no-op.
|
|
|
- let current = sqlx::query(&format!(
|
|
|
- "SELECT version FROM account_head WHERE id = $1 AND subaccount = $2{lock}"
|
|
|
- ))
|
|
|
- .bind(account.id.id)
|
|
|
- .bind(account.id.sub)
|
|
|
- .fetch_optional(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let Some(current) = current else {
|
|
|
- return Ok(0);
|
|
|
- };
|
|
|
-
|
|
|
- let current_version: i64 = current
|
|
|
- .try_get("version")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let expected = current_version
|
|
|
- .checked_add(1)
|
|
|
- .ok_or_else(|| StoreError::Internal("account version overflow".to_string()))?;
|
|
|
-
|
|
|
- if account.version as i64 != expected {
|
|
|
- return Ok(0);
|
|
|
- }
|
|
|
-
|
|
|
- let res = sqlx::query(
|
|
|
- "INSERT INTO accounts (id, subaccount, version, flags, book, metadata) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (id, subaccount, version) DO NOTHING"
|
|
|
- )
|
|
|
- .bind(account.id.id)
|
|
|
- .bind(account.id.sub)
|
|
|
- .bind(account.version as i64)
|
|
|
- .bind(account.flags.bits() as i32)
|
|
|
- .bind(account.book.0)
|
|
|
- .bind(serialize_json(&account.metadata)?)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- if res.rows_affected() == 0 {
|
|
|
- return Ok(0);
|
|
|
- }
|
|
|
-
|
|
|
- // Move the head to the new version (delete + insert, never update).
|
|
|
- sqlx::query("DELETE FROM account_head WHERE id = $1 AND subaccount = $2")
|
|
|
- .bind(account.id.id)
|
|
|
- .bind(account.id.sub)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- sqlx::query("INSERT INTO account_head (id, subaccount, version) VALUES ($1, $2, $3)")
|
|
|
- .bind(account.id.id)
|
|
|
- .bind(account.id.sub)
|
|
|
- .bind(account.version as i64)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- tx.commit()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(1)
|
|
|
- }
|
|
|
-
|
|
|
- async fn get_account_history(&self, id: &AccountId) -> Result<Vec<Account>, StoreError> {
|
|
|
- let rows = sqlx::query(
|
|
|
- "SELECT * FROM accounts WHERE id = $1 AND subaccount = $2 ORDER BY version ASC",
|
|
|
- )
|
|
|
- .bind(id.id)
|
|
|
- .bind(id.sub)
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- if rows.is_empty() {
|
|
|
- return Err(StoreError::NotFound(format!("account {id:?}")));
|
|
|
- }
|
|
|
- rows.iter().map(row_to_account).collect()
|
|
|
- }
|
|
|
-
|
|
|
- async fn list_accounts(&self) -> Result<Vec<Account>, StoreError> {
|
|
|
- // One row per account via the head; no read-all-versions + dedup.
|
|
|
- let rows = sqlx::query(
|
|
|
- "SELECT a.* FROM accounts a \
|
|
|
- JOIN account_head h \
|
|
|
- ON h.id = a.id AND h.subaccount = a.subaccount AND h.version = a.version",
|
|
|
- )
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- rows.iter().map(row_to_account).collect()
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-// PostingStore
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-
|
|
|
-#[async_trait]
|
|
|
-impl PostingStore for SqlStore {
|
|
|
- async fn get_postings(&self, ids: &[PostingId]) -> Result<Vec<Posting>, StoreError> {
|
|
|
- if ids.is_empty() {
|
|
|
- return Ok(Vec::new());
|
|
|
- }
|
|
|
-
|
|
|
- // Set-based query per chunk instead of one probe per id, reusing the
|
|
|
- // portable `id_predicate` and binding each id in order as
|
|
|
- // `(hex(transfer), idx as i16)`. Chunked so a large batch never exceeds
|
|
|
- // the backend's bind-parameter limit (see `MAX_IDS_PER_QUERY`).
|
|
|
- let mut found: HashMap<(String, i16), Posting> = HashMap::with_capacity(ids.len());
|
|
|
- for chunk in ids.chunks(MAX_IDS_PER_QUERY) {
|
|
|
- let sql = format!(
|
|
|
- "SELECT * FROM postings WHERE {}",
|
|
|
- id_predicate(chunk.len(), 1)
|
|
|
- );
|
|
|
- let mut q = sqlx::query(&sql);
|
|
|
- for id in chunk {
|
|
|
- q = q
|
|
|
- .bind(envelope_id_to_hex(&id.transfer))
|
|
|
- .bind(id.index as i16);
|
|
|
- }
|
|
|
- let rows = q
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- // Index the fetched postings by the same `(hex, idx)` key that was bound.
|
|
|
- for row in &rows {
|
|
|
- let posting = row_to_posting(row)?;
|
|
|
- let key = (
|
|
|
- envelope_id_to_hex(&posting.id.transfer),
|
|
|
- posting.id.index as i16,
|
|
|
- );
|
|
|
- found.insert(key, posting);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // Return in input order, erroring on the first id absent from the batch
|
|
|
- // (matching the per-id lookup's `NotFound` semantics).
|
|
|
- let mut result = Vec::with_capacity(ids.len());
|
|
|
- for id in ids {
|
|
|
- let key = (envelope_id_to_hex(&id.transfer), id.index as i16);
|
|
|
- let posting = found
|
|
|
- .get(&key)
|
|
|
- .ok_or_else(|| StoreError::NotFound(format!("posting {id:?}")))?;
|
|
|
- result.push(posting.clone());
|
|
|
- }
|
|
|
- Ok(result)
|
|
|
- }
|
|
|
-
|
|
|
- async fn get_postings_by_account(
|
|
|
- &self,
|
|
|
- id: i64,
|
|
|
- sub: Option<i64>,
|
|
|
- asset: Option<&AssetId>,
|
|
|
- filter: PostingFilter,
|
|
|
- ) -> Result<Vec<Posting>, StoreError> {
|
|
|
- // Build the predicate dynamically: `sub == None` spans every subaccount
|
|
|
- // of `id`, `Some(s)` restricts to one. The subaccount is compared only
|
|
|
- // for equality, never as a magnitude. The derived-state filter selects
|
|
|
- // which table (index copy or immutable record) to read from directly.
|
|
|
- let mut sql = format!("SELECT * FROM {} WHERE owner = $1", filter_source(filter));
|
|
|
- let mut placeholder = 2u32;
|
|
|
- if sub.is_some() {
|
|
|
- sql.push_str(&format!(" AND subaccount = ${placeholder}"));
|
|
|
- placeholder += 1;
|
|
|
- }
|
|
|
- if asset.is_some() {
|
|
|
- sql.push_str(&format!(" AND asset = ${placeholder}"));
|
|
|
- }
|
|
|
- // Deterministic order by the posting primary key, matching
|
|
|
- // `query_postings`, so callers (and pagination built on top) see a
|
|
|
- // stable sequence.
|
|
|
- sql.push_str(" ORDER BY transfer_id, idx");
|
|
|
-
|
|
|
- let mut q = sqlx::query(&sql).bind(id);
|
|
|
- if let Some(s) = sub {
|
|
|
- q = q.bind(s);
|
|
|
- }
|
|
|
- if let Some(a) = asset {
|
|
|
- q = q.bind(a.0 as i32);
|
|
|
- }
|
|
|
-
|
|
|
- let rows = q
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- rows.iter().map(row_to_posting).collect()
|
|
|
- }
|
|
|
-
|
|
|
- async fn get_posting_states(&self, ids: &[PostingId]) -> Result<Vec<PostingState>, StoreError> {
|
|
|
- if ids.is_empty() {
|
|
|
- return Ok(Vec::new());
|
|
|
- }
|
|
|
-
|
|
|
- // One set-based query per state table instead of up to three probes per
|
|
|
- // id, reusing the portable `id_predicate` (an OR of equality pairs;
|
|
|
- // row-value `IN` is not portable across SQLite and PostgreSQL) and
|
|
|
- // binding every id in order as `(hex(transfer), idx as i16)`. Chunked so
|
|
|
- // a large batch never exceeds the bind-parameter limit.
|
|
|
-
|
|
|
- // Key membership by the same `(hex, idx)` values that were bound, so the
|
|
|
- // per-id lookup below matches without decoding transfer ids back.
|
|
|
- let row_key = |row: &AnyRow| -> Result<(String, i16), StoreError> {
|
|
|
- let transfer_id: String = row
|
|
|
- .try_get("transfer_id")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let idx: i16 = row
|
|
|
- .try_get("idx")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok((transfer_id, idx))
|
|
|
- };
|
|
|
-
|
|
|
- let mut active: HashSet<(String, i16)> = HashSet::new();
|
|
|
- let mut reserved: HashMap<(String, i16), i64> = HashMap::new();
|
|
|
- let mut spent: HashSet<(String, i16)> = HashSet::new();
|
|
|
-
|
|
|
- for chunk in ids.chunks(MAX_IDS_PER_QUERY) {
|
|
|
- let predicate = id_predicate(chunk.len(), 1);
|
|
|
-
|
|
|
- let active_sql =
|
|
|
- format!("SELECT transfer_id, idx FROM active_postings WHERE {predicate}");
|
|
|
- let mut active_q = sqlx::query(&active_sql);
|
|
|
- for id in chunk {
|
|
|
- active_q = active_q
|
|
|
- .bind(envelope_id_to_hex(&id.transfer))
|
|
|
- .bind(id.index as i16);
|
|
|
- }
|
|
|
- let active_rows = active_q
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- for row in &active_rows {
|
|
|
- active.insert(row_key(row)?);
|
|
|
- }
|
|
|
-
|
|
|
- let reserved_sql = format!(
|
|
|
- "SELECT transfer_id, idx, reservation FROM reserved_postings WHERE {predicate}"
|
|
|
- );
|
|
|
- let mut reserved_q = sqlx::query(&reserved_sql);
|
|
|
- for id in chunk {
|
|
|
- reserved_q = reserved_q
|
|
|
- .bind(envelope_id_to_hex(&id.transfer))
|
|
|
- .bind(id.index as i16);
|
|
|
- }
|
|
|
- let reserved_rows = reserved_q
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- for row in &reserved_rows {
|
|
|
- let rid: i64 = row
|
|
|
- .try_get("reservation")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- reserved.insert(row_key(row)?, rid);
|
|
|
- }
|
|
|
-
|
|
|
- let spent_sql = format!("SELECT transfer_id, idx FROM postings WHERE {predicate}");
|
|
|
- let mut spent_q = sqlx::query(&spent_sql);
|
|
|
- for id in chunk {
|
|
|
- spent_q = spent_q
|
|
|
- .bind(envelope_id_to_hex(&id.transfer))
|
|
|
- .bind(id.index as i16);
|
|
|
- }
|
|
|
- let spent_rows = spent_q
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- for row in &spent_rows {
|
|
|
- spent.insert(row_key(row)?);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // Reconstruct each id's state in input order, preserving the active >
|
|
|
- // reserved > spent > missing precedence of the original probes.
|
|
|
- let mut out = Vec::with_capacity(ids.len());
|
|
|
- for id in ids {
|
|
|
- let key = (envelope_id_to_hex(&id.transfer), id.index as i16);
|
|
|
- out.push(if active.contains(&key) {
|
|
|
- PostingState::Active
|
|
|
- } else if let Some(rid) = reserved.get(&key) {
|
|
|
- PostingState::Reserved(ReservationId::new(*rid))
|
|
|
- } else if spent.contains(&key) {
|
|
|
- PostingState::Spent
|
|
|
- } else {
|
|
|
- PostingState::Missing
|
|
|
- });
|
|
|
- }
|
|
|
- Ok(out)
|
|
|
- }
|
|
|
-
|
|
|
- async fn query_postings(&self, query: &PostingQuery) -> Result<Page<Posting>, StoreError> {
|
|
|
- let (where_clause, count_clause) = {
|
|
|
- let source = filter_source(query.filter);
|
|
|
- let mut w = String::from("WHERE owner = $1");
|
|
|
- let mut idx = 2u32;
|
|
|
- if query.sub.is_some() {
|
|
|
- w.push_str(&format!(" AND subaccount = ${idx}"));
|
|
|
- idx += 1;
|
|
|
- }
|
|
|
- if query.asset.is_some() {
|
|
|
- w.push_str(&format!(" AND asset = ${idx}"));
|
|
|
- }
|
|
|
- let c = format!("SELECT COUNT(*) as cnt FROM {source} {w}");
|
|
|
- let limit = query.limit.unwrap_or(u32::MAX);
|
|
|
- let offset = query.offset.unwrap_or(0);
|
|
|
- // Order by the posting primary key so pagination is deterministic:
|
|
|
- // without it LIMIT/OFFSET could skip or repeat rows across pages,
|
|
|
- // especially for `Live`, whose source is a `UNION ALL` with no
|
|
|
- // inherent order.
|
|
|
- w.push_str(&format!(
|
|
|
- " ORDER BY transfer_id, idx LIMIT {limit} OFFSET {offset}"
|
|
|
- ));
|
|
|
- (format!("SELECT * FROM {source} {w}"), c)
|
|
|
- };
|
|
|
-
|
|
|
- // Build count query
|
|
|
- let mut count_q = sqlx::query(&count_clause).bind(query.account);
|
|
|
- if let Some(s) = query.sub {
|
|
|
- count_q = count_q.bind(s);
|
|
|
- }
|
|
|
- if let Some(ref a) = query.asset {
|
|
|
- count_q = count_q.bind(a.0 as i32);
|
|
|
- }
|
|
|
- let count_row = count_q
|
|
|
- .fetch_one(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let total: i64 = count_row
|
|
|
- .try_get("cnt")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- // Build data query
|
|
|
- let mut data_q = sqlx::query(&where_clause).bind(query.account);
|
|
|
- if let Some(s) = query.sub {
|
|
|
- data_q = data_q.bind(s);
|
|
|
- }
|
|
|
- if let Some(ref a) = query.asset {
|
|
|
- data_q = data_q.bind(a.0 as i32);
|
|
|
- }
|
|
|
- let rows = data_q
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let items: Vec<Posting> = rows.iter().map(row_to_posting).collect::<Result<_, _>>()?;
|
|
|
- Ok(Page {
|
|
|
- items,
|
|
|
- total: total as u64,
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- async fn reserve_postings(
|
|
|
- &self,
|
|
|
- ids: &[PostingId],
|
|
|
- reservation: ReservationId,
|
|
|
- ) -> Result<u64, StoreError> {
|
|
|
- // Dumb instruction over the whole id set, in two statements: copy the
|
|
|
- // currently-active rows into the reserved index (sourced from
|
|
|
- // `active_postings`, so only active ids move), then delete those same
|
|
|
- // ids from `active_postings`. The DELETE's affected count is the number
|
|
|
- // claimed, and by active/reserved disjointness it equals the INSERT's
|
|
|
- // row count. Concurrent reserves serialize on the reserved-index primary
|
|
|
- // key, so exactly one wins each contended id.
|
|
|
- if ids.is_empty() {
|
|
|
- return Ok(0);
|
|
|
- }
|
|
|
- let mut tx = self
|
|
|
- .pool
|
|
|
- .begin()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- // Chunked so a large id set stays under the bind-parameter limit; all
|
|
|
- // chunks share one transaction so the whole claim is atomic.
|
|
|
- let mut claimed: u64 = 0;
|
|
|
- for chunk in ids.chunks(MAX_IDS_PER_QUERY) {
|
|
|
- // Reservation is $1; each id pair follows starting at $2.
|
|
|
- let insert_sql = format!(
|
|
|
- "INSERT INTO reserved_postings (transfer_id, idx, owner, subaccount, asset, value, reservation) \
|
|
|
- SELECT transfer_id, idx, owner, subaccount, asset, value, $1 FROM active_postings WHERE {} \
|
|
|
- ON CONFLICT (transfer_id, idx) DO NOTHING",
|
|
|
- id_predicate(chunk.len(), 2)
|
|
|
- );
|
|
|
- let mut insert_q = sqlx::query(&insert_sql).bind(reservation.0);
|
|
|
- for id in chunk {
|
|
|
- insert_q = insert_q
|
|
|
- .bind(envelope_id_to_hex(&id.transfer))
|
|
|
- .bind(id.index as i16);
|
|
|
- }
|
|
|
- insert_q
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let delete_sql = format!(
|
|
|
- "DELETE FROM active_postings WHERE {}",
|
|
|
- id_predicate(chunk.len(), 1)
|
|
|
- );
|
|
|
- let mut delete_q = sqlx::query(&delete_sql);
|
|
|
- for id in chunk {
|
|
|
- delete_q = delete_q
|
|
|
- .bind(envelope_id_to_hex(&id.transfer))
|
|
|
- .bind(id.index as i16);
|
|
|
- }
|
|
|
- let del = delete_q
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- claimed += del.rows_affected();
|
|
|
- }
|
|
|
-
|
|
|
- tx.commit()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(claimed)
|
|
|
- }
|
|
|
-
|
|
|
- async fn release_postings(
|
|
|
- &self,
|
|
|
- ids: &[PostingId],
|
|
|
- reservation: ReservationId,
|
|
|
- ) -> Result<u64, StoreError> {
|
|
|
- // Dumb instruction over the whole id set: copy the rows reserved by
|
|
|
- // `reservation` back into the active index, then delete them from the
|
|
|
- // reserved index. The DELETE's affected count is the number released; an
|
|
|
- // id already active or reserved by another saga does not match.
|
|
|
- if ids.is_empty() {
|
|
|
- return Ok(0);
|
|
|
- }
|
|
|
- let mut tx = self
|
|
|
- .pool
|
|
|
- .begin()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- // Chunked so a large id set stays under the bind-parameter limit; all
|
|
|
- // chunks share one transaction.
|
|
|
- let mut released: u64 = 0;
|
|
|
- for chunk in ids.chunks(MAX_IDS_PER_QUERY) {
|
|
|
- // Reservation is $1; each id pair follows starting at $2.
|
|
|
- let insert_sql = format!(
|
|
|
- "INSERT INTO active_postings (transfer_id, idx, owner, subaccount, asset, value) \
|
|
|
- SELECT transfer_id, idx, owner, subaccount, asset, value FROM reserved_postings \
|
|
|
- WHERE ({}) AND reservation = $1 ON CONFLICT (transfer_id, idx) DO NOTHING",
|
|
|
- id_predicate(chunk.len(), 2)
|
|
|
- );
|
|
|
- let mut insert_q = sqlx::query(&insert_sql).bind(reservation.0);
|
|
|
- for id in chunk {
|
|
|
- insert_q = insert_q
|
|
|
- .bind(envelope_id_to_hex(&id.transfer))
|
|
|
- .bind(id.index as i16);
|
|
|
- }
|
|
|
- insert_q
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let delete_sql = format!(
|
|
|
- "DELETE FROM reserved_postings WHERE ({}) AND reservation = $1",
|
|
|
- id_predicate(chunk.len(), 2)
|
|
|
- );
|
|
|
- let mut delete_q = sqlx::query(&delete_sql).bind(reservation.0);
|
|
|
- for id in chunk {
|
|
|
- delete_q = delete_q
|
|
|
- .bind(envelope_id_to_hex(&id.transfer))
|
|
|
- .bind(id.index as i16);
|
|
|
- }
|
|
|
- let del = delete_q
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- released += del.rows_affected();
|
|
|
- }
|
|
|
-
|
|
|
- tx.commit()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(released)
|
|
|
- }
|
|
|
-
|
|
|
- async fn deactivate_postings(
|
|
|
- &self,
|
|
|
- ids: &[PostingId],
|
|
|
- reservation: Option<ReservationId>,
|
|
|
- ) -> Result<u64, StoreError> {
|
|
|
- // Dumb instruction over the whole id set: a DELETE removes the ids from
|
|
|
- // an index so they become spent (present only in the immutable table).
|
|
|
- // `rows_affected` is the count; the caller interprets a shortfall.
|
|
|
- // Chunked under one transaction so a large id set stays within the
|
|
|
- // bind-parameter limit while the removal stays atomic.
|
|
|
- if ids.is_empty() {
|
|
|
- return Ok(0);
|
|
|
- }
|
|
|
- let mut tx = self
|
|
|
- .pool
|
|
|
- .begin()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let mut removed: u64 = 0;
|
|
|
- for chunk in ids.chunks(MAX_IDS_PER_QUERY) {
|
|
|
- let (sql, rid) = match reservation {
|
|
|
- // Raw path: remove from the active index.
|
|
|
- None => (
|
|
|
- format!(
|
|
|
- "DELETE FROM active_postings WHERE {}",
|
|
|
- id_predicate(chunk.len(), 1)
|
|
|
- ),
|
|
|
- None,
|
|
|
- ),
|
|
|
- // Saga path: remove only the rows reserved by `rid`.
|
|
|
- Some(rid) => (
|
|
|
- format!(
|
|
|
- "DELETE FROM reserved_postings WHERE ({}) AND reservation = $1",
|
|
|
- id_predicate(chunk.len(), 2)
|
|
|
- ),
|
|
|
- Some(rid),
|
|
|
- ),
|
|
|
- };
|
|
|
- let mut q = sqlx::query(&sql);
|
|
|
- if let Some(rid) = rid {
|
|
|
- q = q.bind(rid.0);
|
|
|
- }
|
|
|
- for id in chunk {
|
|
|
- q = q
|
|
|
- .bind(envelope_id_to_hex(&id.transfer))
|
|
|
- .bind(id.index as i16);
|
|
|
- }
|
|
|
- let res = q
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- removed += res.rows_affected();
|
|
|
- }
|
|
|
- tx.commit()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(removed)
|
|
|
- }
|
|
|
-
|
|
|
- async fn insert_postings(&self, postings: &[Posting]) -> Result<u64, StoreError> {
|
|
|
- // Dumb instruction: insert each posting into the immutable table and, only
|
|
|
- // when the row was newly inserted, add its id to the active index. Return
|
|
|
- // the count of immutable rows inserted. The newness gate stops a replayed
|
|
|
- // finalize from re-activating a since-spent posting.
|
|
|
- let mut tx = self
|
|
|
- .pool
|
|
|
- .begin()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let mut inserted: u64 = 0;
|
|
|
- for posting in postings {
|
|
|
- let hex = envelope_id_to_hex(&posting.id.transfer);
|
|
|
- let res = sqlx::query(
|
|
|
- "INSERT INTO postings (transfer_id, idx, owner, subaccount, asset, value) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (transfer_id, idx) DO NOTHING"
|
|
|
- )
|
|
|
- .bind(hex.clone())
|
|
|
- .bind(posting.id.index as i16)
|
|
|
- .bind(posting.owner.id)
|
|
|
- .bind(posting.owner.sub)
|
|
|
- .bind(posting.asset.0 as i32)
|
|
|
- .bind(posting.value.to_string())
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- if res.rows_affected() == 1 {
|
|
|
- // Activate a full copy so spendable reads never merge.
|
|
|
- sqlx::query(
|
|
|
- "INSERT INTO active_postings (transfer_id, idx, owner, subaccount, asset, value) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (transfer_id, idx) DO NOTHING",
|
|
|
- )
|
|
|
- .bind(hex)
|
|
|
- .bind(posting.id.index as i16)
|
|
|
- .bind(posting.owner.id)
|
|
|
- .bind(posting.owner.sub)
|
|
|
- .bind(posting.asset.0 as i32)
|
|
|
- .bind(posting.value.to_string())
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- inserted += 1;
|
|
|
- }
|
|
|
- }
|
|
|
- tx.commit()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(inserted)
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-// TransferStore
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-
|
|
|
-#[async_trait]
|
|
|
-impl TransferStore for SqlStore {
|
|
|
- async fn get_transfer(&self, id: &EnvelopeId) -> Result<Option<EnvelopeRecord>, StoreError> {
|
|
|
- let row = sqlx::query("SELECT transfer, receipt, created_at FROM transfers WHERE id = $1")
|
|
|
- .bind(envelope_id_to_hex(id))
|
|
|
- .fetch_optional(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- match row {
|
|
|
- None => Ok(None),
|
|
|
- Some(row) => {
|
|
|
- let transfer_json: String = row
|
|
|
- .try_get("transfer")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let receipt_json: String = row
|
|
|
- .try_get("receipt")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let created_at: i64 = row
|
|
|
- .try_get("created_at")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(Some(EnvelopeRecord {
|
|
|
- envelope: deserialize_json(&transfer_json)?,
|
|
|
- receipt: deserialize_json(&receipt_json)?,
|
|
|
- created_at,
|
|
|
- }))
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- async fn store_transfer(
|
|
|
- &self,
|
|
|
- record: EnvelopeRecord,
|
|
|
- involved: &[AccountId],
|
|
|
- ) -> Result<u64, StoreError> {
|
|
|
- let tid = record.receipt.transfer_id;
|
|
|
- let tid_hex = envelope_id_to_hex(&tid);
|
|
|
- let transfer_json = serialize_json(&record.envelope)?;
|
|
|
- let receipt_json = serialize_json(&record.receipt)?;
|
|
|
-
|
|
|
- let mut tx = self
|
|
|
- .pool
|
|
|
- .begin()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let res = sqlx::query("INSERT INTO transfers (id, transfer, receipt, created_at, book) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (id) DO NOTHING")
|
|
|
- .bind(&tid_hex)
|
|
|
- .bind(&transfer_json)
|
|
|
- .bind(&receipt_json)
|
|
|
- .bind(record.created_at)
|
|
|
- .bind(record.envelope.book().0)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let inserted = res.rows_affected();
|
|
|
-
|
|
|
- // Index every involved account (caller supplies the set; storage does no
|
|
|
- // computation). Idempotent so a replay is harmless.
|
|
|
- for account in involved {
|
|
|
- sqlx::query("INSERT INTO transfer_accounts (transfer_id, account_id, subaccount) VALUES ($1, $2, $3) ON CONFLICT (transfer_id, account_id, subaccount) DO NOTHING")
|
|
|
- .bind(&tid_hex)
|
|
|
- .bind(account.id)
|
|
|
- .bind(account.sub)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- }
|
|
|
-
|
|
|
- tx.commit()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(inserted)
|
|
|
- }
|
|
|
-
|
|
|
- async fn get_transfers_for_account(
|
|
|
- &self,
|
|
|
- id: i64,
|
|
|
- sub: Option<i64>,
|
|
|
- ) -> Result<Vec<EnvelopeRecord>, StoreError> {
|
|
|
- // `sub == None` spans every subaccount of `id`; `Some(s)` restricts to
|
|
|
- // one. The subaccount is matched only for equality.
|
|
|
- let mut sql = String::from(
|
|
|
- "SELECT t.id, t.transfer, t.receipt, t.created_at FROM transfers t INNER JOIN transfer_accounts ta ON t.id = ta.transfer_id WHERE ta.account_id = $1",
|
|
|
- );
|
|
|
- if sub.is_some() {
|
|
|
- sql.push_str(" AND ta.subaccount = $2");
|
|
|
- }
|
|
|
- sql.push_str(" ORDER BY t.created_at");
|
|
|
-
|
|
|
- let mut q = sqlx::query(&sql).bind(id);
|
|
|
- if let Some(s) = sub {
|
|
|
- q = q.bind(s);
|
|
|
- }
|
|
|
- let rows = q
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let mut result = Vec::with_capacity(rows.len());
|
|
|
- for row in &rows {
|
|
|
- let transfer_json: String = row
|
|
|
- .try_get("transfer")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let receipt_json: String = row
|
|
|
- .try_get("receipt")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let created_at: i64 = row
|
|
|
- .try_get("created_at")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- result.push(EnvelopeRecord {
|
|
|
- envelope: deserialize_json(&transfer_json)?,
|
|
|
- receipt: deserialize_json(&receipt_json)?,
|
|
|
- created_at,
|
|
|
- });
|
|
|
- }
|
|
|
- Ok(result)
|
|
|
- }
|
|
|
-
|
|
|
- async fn query_transfers(
|
|
|
- &self,
|
|
|
- query: &TransferQuery,
|
|
|
- ) -> Result<Page<EnvelopeRecord>, StoreError> {
|
|
|
- // Load base records, using the account join when available.
|
|
|
- let base_records = if let Some(account) = query.account {
|
|
|
- self.get_transfers_for_account(account, query.sub).await?
|
|
|
- } else {
|
|
|
- let rows = sqlx::query(
|
|
|
- "SELECT transfer, receipt, created_at FROM transfers ORDER BY created_at",
|
|
|
- )
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let mut records = Vec::with_capacity(rows.len());
|
|
|
- for row in &rows {
|
|
|
- let transfer_json: String = row
|
|
|
- .try_get("transfer")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let receipt_json: String = row
|
|
|
- .try_get("receipt")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let created_at: i64 = row
|
|
|
- .try_get("created_at")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- records.push(EnvelopeRecord {
|
|
|
- envelope: deserialize_json(&transfer_json)?,
|
|
|
- receipt: deserialize_json(&receipt_json)?,
|
|
|
- created_at,
|
|
|
- });
|
|
|
- }
|
|
|
- records
|
|
|
- };
|
|
|
-
|
|
|
- // The account/subaccount narrowing happened in the load above; the
|
|
|
- // shared filter covers the time-window and book predicates, then the
|
|
|
- // shared page cut applies `offset`/`limit`.
|
|
|
- Ok(paginate(
|
|
|
- filter_transfers(base_records, query),
|
|
|
- query.offset,
|
|
|
- query.limit,
|
|
|
- ))
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-// SagaStore
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-
|
|
|
-#[async_trait]
|
|
|
-impl SagaStore for SqlStore {
|
|
|
- async fn save_saga(&self, id: &i64, data: Vec<u8>) -> Result<(), StoreError> {
|
|
|
- sqlx::query(
|
|
|
- "INSERT INTO sagas (id, data) VALUES ($1, $2) \
|
|
|
- ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data",
|
|
|
- )
|
|
|
- .bind(*id)
|
|
|
- .bind(to_hex(&data))
|
|
|
- .execute(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(())
|
|
|
- }
|
|
|
-
|
|
|
- async fn list_pending_sagas(&self) -> Result<Vec<(i64, Vec<u8>)>, StoreError> {
|
|
|
- let rows = sqlx::query("SELECT id, data FROM sagas")
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let mut result = Vec::with_capacity(rows.len());
|
|
|
- for row in &rows {
|
|
|
- let id: i64 = row
|
|
|
- .try_get("id")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let data_hex: String = row
|
|
|
- .try_get("data")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- result.push((id, from_hex(&data_hex)?));
|
|
|
- }
|
|
|
- Ok(result)
|
|
|
- }
|
|
|
-
|
|
|
- async fn get_saga(&self, id: &i64) -> Result<Option<Vec<u8>>, StoreError> {
|
|
|
- let row = sqlx::query("SELECT data FROM sagas WHERE id = $1")
|
|
|
- .bind(*id)
|
|
|
- .fetch_optional(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- match row {
|
|
|
- Some(row) => {
|
|
|
- let data_hex: String = row
|
|
|
- .try_get("data")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(Some(from_hex(&data_hex)?))
|
|
|
- }
|
|
|
- None => Ok(None),
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- async fn delete_saga(&self, id: &i64) -> Result<(), StoreError> {
|
|
|
- sqlx::query("DELETE FROM sagas WHERE id = $1")
|
|
|
- .bind(*id)
|
|
|
- .execute(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(())
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-// EventStore
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-
|
|
|
-#[async_trait]
|
|
|
-impl EventStore for SqlStore {
|
|
|
- async fn append_event(&self, event: &LedgerEvent) -> Result<u64, StoreError> {
|
|
|
- let kind_str =
|
|
|
- serde_json::to_string(&event.kind).map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let data = serialize_json(event)?;
|
|
|
- let seq = self.autoid.next() as u64;
|
|
|
-
|
|
|
- // Idempotent on the dedup key: a replayed transfer or lifecycle-transition
|
|
|
- // event conflicts on `dedup_key` and returns the existing seq instead of a
|
|
|
- // duplicate row.
|
|
|
- match event_dedup_key(&event.kind) {
|
|
|
- Some(dedup_key) => {
|
|
|
- let res = sqlx::query("INSERT INTO events (seq, timestamp, kind, data, dedup_key) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (dedup_key) DO NOTHING")
|
|
|
- .bind(seq as i64)
|
|
|
- .bind(event.timestamp)
|
|
|
- .bind(&kind_str)
|
|
|
- .bind(&data)
|
|
|
- .bind(&dedup_key)
|
|
|
- .execute(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- if res.rows_affected() == 0 {
|
|
|
- let row = sqlx::query("SELECT seq FROM events WHERE dedup_key = $1")
|
|
|
- .bind(&dedup_key)
|
|
|
- .fetch_one(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let existing: i64 = row
|
|
|
- .try_get("seq")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- return Ok(existing as u64);
|
|
|
- }
|
|
|
- Ok(seq)
|
|
|
- }
|
|
|
- None => {
|
|
|
- sqlx::query(
|
|
|
- "INSERT INTO events (seq, timestamp, kind, data) VALUES ($1, $2, $3, $4)",
|
|
|
- )
|
|
|
- .bind(seq as i64)
|
|
|
- .bind(event.timestamp)
|
|
|
- .bind(&kind_str)
|
|
|
- .bind(&data)
|
|
|
- .execute(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(seq)
|
|
|
- }
|
|
|
+ dialect,
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
- async fn get_events_since(
|
|
|
- &self,
|
|
|
- after_seq: u64,
|
|
|
- limit: u32,
|
|
|
- ) -> Result<Vec<LedgerEvent>, StoreError> {
|
|
|
- let rows = sqlx::query("SELECT seq, data FROM events WHERE seq > $1 ORDER BY seq LIMIT $2")
|
|
|
- .bind(after_seq as i64)
|
|
|
- .bind(limit as i32)
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let mut events = Vec::with_capacity(rows.len());
|
|
|
- for row in &rows {
|
|
|
- let seq: i64 = row
|
|
|
- .try_get("seq")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let data_json: String = row
|
|
|
- .try_get("data")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let mut event: LedgerEvent = deserialize_json(&data_json)?;
|
|
|
- event.seq = seq as u64;
|
|
|
- events.push(event);
|
|
|
- }
|
|
|
- Ok(events)
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-// BookStore
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-
|
|
|
-#[async_trait]
|
|
|
-impl BookStore for SqlStore {
|
|
|
- async fn create_book(&self, book: Book) -> Result<u64, StoreError> {
|
|
|
- // Pessimistic locking, same shape as create_account: lock any existing
|
|
|
- // book row with `SELECT ... FOR UPDATE` inside the transaction, then
|
|
|
- // insert with `ON CONFLICT DO NOTHING` as the portable backstop.
|
|
|
- let lock = self.lock_clause().await?;
|
|
|
- let data = serialize_json(&book)?;
|
|
|
- let mut tx = self
|
|
|
- .pool
|
|
|
- .begin()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
-
|
|
|
- let existing = sqlx::query(&format!("SELECT 1 FROM books WHERE id = $1 LIMIT 1{lock}"))
|
|
|
- .bind(book.id.0)
|
|
|
- .fetch_optional(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- if existing.is_some() {
|
|
|
- return Ok(0);
|
|
|
- }
|
|
|
-
|
|
|
- let res = sqlx::query(
|
|
|
- "INSERT INTO books (id, name, data) VALUES ($1, $2, $3) ON CONFLICT (id) DO NOTHING",
|
|
|
- )
|
|
|
- .bind(book.id.0)
|
|
|
- .bind(&book.name)
|
|
|
- .bind(&data)
|
|
|
- .execute(&mut *tx)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- if res.rows_affected() == 0 {
|
|
|
- return Ok(0);
|
|
|
- }
|
|
|
-
|
|
|
- tx.commit()
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(1)
|
|
|
- }
|
|
|
-
|
|
|
- async fn get_book(&self, id: &BookId) -> Result<Book, StoreError> {
|
|
|
- let row = sqlx::query("SELECT data FROM books WHERE id = $1")
|
|
|
- .bind(id.0)
|
|
|
- .fetch_optional(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?
|
|
|
- .ok_or_else(|| StoreError::NotFound(format!("book {id:?}")))?;
|
|
|
- let data: String = row
|
|
|
- .try_get("data")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- deserialize_json(&data)
|
|
|
- }
|
|
|
-
|
|
|
- async fn list_books(&self) -> Result<Vec<Book>, StoreError> {
|
|
|
- let rows = sqlx::query("SELECT data FROM books")
|
|
|
- .fetch_all(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- rows.iter()
|
|
|
- .map(|row| {
|
|
|
- let data: String = row
|
|
|
- .try_get("data")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- deserialize_json(&data)
|
|
|
- })
|
|
|
- .collect()
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-// BalanceProjectionStore
|
|
|
-// ---------------------------------------------------------------------------
|
|
|
-
|
|
|
-#[async_trait]
|
|
|
-impl BalanceProjectionStore for SqlStore {
|
|
|
- async fn append_balance_projection(
|
|
|
- &self,
|
|
|
- account: &AccountId,
|
|
|
- asset: &AssetId,
|
|
|
- balance: Cent,
|
|
|
- watermark: i64,
|
|
|
- ) -> Result<(), StoreError> {
|
|
|
- // Append-only: mint a fresh monotonic id and insert a new cache point.
|
|
|
- let id = self.autoid.next();
|
|
|
- sqlx::query(
|
|
|
- "INSERT INTO balance_projection (id, account, subaccount, asset, balance, watermark) \
|
|
|
- VALUES ($1, $2, $3, $4, $5, $6)",
|
|
|
- )
|
|
|
- .bind(id)
|
|
|
- .bind(account.id)
|
|
|
- .bind(account.sub)
|
|
|
- .bind(asset.0 as i32)
|
|
|
- .bind(balance.to_string())
|
|
|
- .bind(watermark)
|
|
|
- .execute(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(())
|
|
|
- }
|
|
|
-
|
|
|
- async fn get_closest_balance_projection(
|
|
|
- &self,
|
|
|
- account: &AccountId,
|
|
|
- asset: &AssetId,
|
|
|
- as_of: i64,
|
|
|
- ) -> Result<Option<BalanceProjection>, StoreError> {
|
|
|
- // Closest at or before `as_of`: the largest watermark not exceeding it,
|
|
|
- // tie-broken by highest id. Row selection, not an aggregate over values.
|
|
|
- let row = sqlx::query(
|
|
|
- "SELECT id, balance, watermark FROM balance_projection \
|
|
|
- WHERE account = $1 AND subaccount = $2 AND asset = $3 AND watermark <= $4 \
|
|
|
- ORDER BY watermark DESC, id DESC LIMIT 1",
|
|
|
- )
|
|
|
- .bind(account.id)
|
|
|
- .bind(account.sub)
|
|
|
- .bind(asset.0 as i32)
|
|
|
- .bind(as_of)
|
|
|
- .fetch_optional(&self.pool)
|
|
|
- .await
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let Some(row) = row else {
|
|
|
- return Ok(None);
|
|
|
- };
|
|
|
- let id: i64 = row
|
|
|
- .try_get("id")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let balance: String = row
|
|
|
- .try_get("balance")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- let watermark: i64 = row
|
|
|
- .try_get("watermark")
|
|
|
- .map_err(|e| StoreError::Internal(e.to_string()))?;
|
|
|
- Ok(Some(BalanceProjection {
|
|
|
- id,
|
|
|
- account: *account,
|
|
|
- asset: *asset,
|
|
|
- balance: Cent::from_str(&balance).map_err(|e| StoreError::Internal(e.to_string()))?,
|
|
|
- watermark,
|
|
|
- }))
|
|
|
- }
|
|
|
}
|