use std::collections::BTreeMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Mutex;
use time::Date;
use crate::account::{AccountId, AccountRecord};
use crate::balance::{Balance, BalanceKey, TrialBalance};
use crate::checkpoint::Checkpoint;
use crate::clearing::{Clearing, ClearingId, OpenItem};
use crate::entry::{Balanced, Entry, EntryId};
use crate::hash::Hash;
use crate::journal::{Journal, JournalError, LogIndex, NotSequenced, Recorded};
use crate::merkle::{ConsistencyProof, InclusionProof, TreeHead};
use crate::money::Currency;
use crate::period::{LedgerId, Period, PeriodId, PeriodState};
use crate::posting::Layer;
use crate::seal::Seal;
pub mod conformance;
#[cfg(feature = "postgres")]
#[cfg_attr(docsrs, doc(cfg(feature = "postgres")))]
pub mod postgres;
#[cfg(feature = "sqlite")]
#[cfg_attr(docsrs, doc(cfg(feature = "sqlite")))]
pub mod sqlite;
#[cfg(feature = "iceberg")]
#[cfg_attr(docsrs, doc(cfg(feature = "iceberg")))]
pub mod iceberg;
pub const DEFAULT_PAGE_SIZE: usize = 256;
pub const MAX_PAGE_SIZE: usize = 4096;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EntryBatch<const P: u8> {
entries: Vec<Entry<Balanced, P>>,
}
impl<const P: u8> EntryBatch<P> {
pub fn new(entries: Vec<Entry<Balanced, P>>) -> Result<Self, BatchError> {
if entries.is_empty() {
return Err(BatchError::Empty);
}
Ok(Self { entries })
}
#[must_use]
pub fn single(entry: Entry<Balanced, P>) -> Self {
Self {
entries: vec![entry],
}
}
#[must_use]
pub fn entries(&self) -> &[Entry<Balanced, P>] {
&self.entries
}
#[must_use]
pub fn len(&self) -> usize {
self.entries.len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum BatchError {
#[error("an entry batch must hold at least one entry")]
Empty,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Cursor {
pub after: Option<LogIndex>,
pub limit: usize,
}
impl Default for Cursor {
fn default() -> Self {
Self {
after: None,
limit: DEFAULT_PAGE_SIZE,
}
}
}
impl Cursor {
#[must_use]
pub fn start() -> Self {
Self::default()
}
#[must_use]
pub fn after(index: LogIndex) -> Self {
Self {
after: Some(index),
limit: DEFAULT_PAGE_SIZE,
}
}
#[must_use]
pub fn with_limit(mut self, limit: usize) -> Self {
self.limit = limit;
self
}
#[must_use]
pub fn effective_limit(&self) -> usize {
self.limit.clamp(1, MAX_PAGE_SIZE)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoredEntry<const P: u8> {
pub index: Option<LogIndex>,
pub entry: Entry<Balanced, P>,
pub content_hash: Hash,
}
impl<const P: u8> StoredEntry<P> {
pub fn require_index(&self) -> Result<LogIndex, NotSequenced> {
self.index.ok_or(NotSequenced {
id: self.entry.id(),
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Page<const P: u8> {
pub records: Vec<StoredEntry<P>>,
pub next: Option<Cursor>,
}
impl<const P: u8> Page<P> {
#[must_use]
pub fn is_empty(&self) -> bool {
self.records.is_empty()
}
}
pub trait LedgerStore<const P: u8>: Send + Sync {
type Error: std::error::Error + Send + Sync + 'static;
fn ledger(&self) -> &LedgerId;
fn append(
&self,
batch: &EntryBatch<P>,
) -> impl Future<Output = Result<Vec<Recorded>, Self::Error>> + Send;
fn register_account(
&self,
record: &AccountRecord,
) -> impl Future<Output = Result<(), Self::Error>> + Send;
fn accounts(&self) -> impl Future<Output = Result<Vec<AccountRecord>, Self::Error>> + Send;
fn get(
&self,
id: EntryId,
) -> impl Future<Output = Result<Option<StoredEntry<P>>, Self::Error>> + Send;
fn page(&self, cursor: Cursor) -> impl Future<Output = Result<Page<P>, Self::Error>> + Send;
fn head(&self) -> impl Future<Output = Result<TreeHead, Self::Error>> + Send;
fn sequence(&self) -> impl Future<Output = Result<u64, Self::Error>> + Send {
async { Ok(0) }
}
fn len(&self) -> impl Future<Output = Result<u64, Self::Error>> + Send;
fn is_empty(&self) -> impl Future<Output = Result<bool, Self::Error>> + Send {
async { self.len().await.map(|n| n == 0) }
}
fn balance(
&self,
key: BalanceKey,
size: Option<u64>,
) -> impl Future<Output = Result<Balance<P>, Self::Error>> + Send;
fn trial_balance(
&self,
size: Option<u64>,
) -> impl Future<Output = Result<TrialBalance<P>, Self::Error>> + Send;
fn prove_inclusion(
&self,
index: LogIndex,
) -> impl Future<Output = Result<InclusionProof, Self::Error>> + Send;
fn prove_consistency(
&self,
old_size: u64,
) -> impl Future<Output = Result<ConsistencyProof, Self::Error>> + Send;
fn define_period(
&self,
period: &Period,
) -> impl Future<Output = Result<(), Self::Error>> + Send;
fn transition_period(
&self,
period: &PeriodId,
to: PeriodState,
) -> impl Future<Output = Result<(), Self::Error>> + Send;
fn periods(&self) -> impl Future<Output = Result<Vec<Period>, Self::Error>> + Send;
fn seal_period(
&self,
period: &PeriodId,
) -> impl Future<Output = Result<Seal, Self::Error>> + Send;
fn seals(&self) -> impl Future<Output = Result<Vec<Seal>, Self::Error>> + Send;
fn clear(&self, clearing: Clearing<P>) -> impl Future<Output = Result<(), Self::Error>> + Send;
fn reset_clearing(
&self,
id: ClearingId,
on: Date,
) -> impl Future<Output = Result<(), Self::Error>> + Send;
fn open_items(
&self,
key: BalanceKey,
) -> impl Future<Output = Result<Vec<OpenItem<P>>, Self::Error>> + Send;
fn balances(
&self,
accounts: &[AccountId],
currency: Currency,
layer: Layer,
size: Option<u64>,
) -> impl Future<Output = Result<BTreeMap<AccountId, Balance<P>>, Self::Error>> + Send;
fn statement(
&self,
key: BalanceKey,
cursor: Cursor,
) -> impl Future<Output = Result<StatementPage<P>, Self::Error>> + Send;
fn save_checkpoint(
&self,
checkpoint: &Checkpoint<P>,
) -> impl Future<Output = Result<(), Self::Error>> + Send;
fn load_checkpoint(
&self,
key: BalanceKey,
) -> impl Future<Output = Result<Option<Checkpoint<P>>, Self::Error>> + Send;
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StatementLine<const P: u8> {
pub index: LogIndex,
pub posting: crate::clearing::PostingRef,
pub booking_date: Date,
pub direction: crate::posting::Direction,
pub amount: crate::money::Amount<P>,
pub running: Balance<P>,
pub kind: Option<crate::dimensions::Label>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StatementPage<const P: u8> {
pub lines: Vec<StatementLine<P>>,
pub next: Option<Cursor>,
}
pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
pub trait DynLedgerStore<const P: u8>: Send + Sync {
type Error: std::error::Error + Send + Sync + 'static;
fn append_boxed<'a>(
&'a self,
batch: &'a EntryBatch<P>,
) -> BoxFuture<'a, Result<Vec<Recorded>, Self::Error>>;
fn get_boxed(&self, id: EntryId) -> BoxFuture<'_, Result<Option<StoredEntry<P>>, Self::Error>>;
fn page_boxed(&self, cursor: Cursor) -> BoxFuture<'_, Result<Page<P>, Self::Error>>;
fn head_boxed(&self) -> BoxFuture<'_, Result<TreeHead, Self::Error>>;
fn balance_boxed(
&self,
key: BalanceKey,
size: Option<u64>,
) -> BoxFuture<'_, Result<Balance<P>, Self::Error>>;
}
impl<const P: u8, S> DynLedgerStore<P> for S
where
S: LedgerStore<P>,
{
type Error = S::Error;
fn append_boxed<'a>(
&'a self,
batch: &'a EntryBatch<P>,
) -> BoxFuture<'a, Result<Vec<Recorded>, Self::Error>> {
Box::pin(self.append(batch))
}
fn get_boxed(&self, id: EntryId) -> BoxFuture<'_, Result<Option<StoredEntry<P>>, Self::Error>> {
Box::pin(self.get(id))
}
fn page_boxed(&self, cursor: Cursor) -> BoxFuture<'_, Result<Page<P>, Self::Error>> {
Box::pin(self.page(cursor))
}
fn head_boxed(&self) -> BoxFuture<'_, Result<TreeHead, Self::Error>> {
Box::pin(self.head())
}
fn balance_boxed(
&self,
key: BalanceKey,
size: Option<u64>,
) -> BoxFuture<'_, Result<Balance<P>, Self::Error>> {
Box::pin(self.balance(key, size))
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum MemoryStoreError {
#[error(transparent)]
Journal(#[from] JournalError),
#[error(transparent)]
Account(#[from] crate::account::AccountError),
#[error(transparent)]
Period(#[from] crate::period::PeriodError),
#[error(transparent)]
Proof(#[from] crate::merkle::ProofError),
#[error(transparent)]
Money(#[from] crate::money::MoneyError),
}
#[derive(Debug)]
pub struct MemoryStore<const P: u8> {
ledger: LedgerId,
inner: Mutex<Journal<P>>,
checkpoints: Mutex<BTreeMap<BalanceKey, Checkpoint<P>>>,
}
impl<const P: u8> MemoryStore<P> {
#[must_use]
pub fn new(ledger: LedgerId) -> Self {
Self {
inner: Mutex::new(Journal::new(ledger.clone())),
ledger,
checkpoints: Mutex::new(BTreeMap::new()),
}
}
#[must_use]
pub fn from_journal(journal: Journal<P>) -> Self {
Self {
ledger: journal.ledger().clone(),
inner: Mutex::new(journal),
checkpoints: Mutex::new(BTreeMap::new()),
}
}
fn with<R>(&self, f: impl FnOnce(&Journal<P>) -> R) -> R {
let guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
f(&guard)
}
fn with_mut<R>(&self, f: impl FnOnce(&mut Journal<P>) -> R) -> R {
let mut guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
f(&mut guard)
}
#[must_use]
pub fn snapshot(&self) -> Journal<P> {
self.with(Clone::clone)
}
}
impl<const P: u8> LedgerStore<P> for MemoryStore<P> {
type Error = MemoryStoreError;
fn ledger(&self) -> &LedgerId {
&self.ledger
}
fn register_account(
&self,
record: &AccountRecord,
) -> impl Future<Output = Result<(), Self::Error>> + Send {
let record = record.clone();
let result = self.with_mut(|journal| {
journal.accounts_mut().restore(record)?;
Ok(())
});
async move { result }
}
fn accounts(&self) -> impl Future<Output = Result<Vec<AccountRecord>, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.account_records()));
async move { result }
}
fn append(
&self,
batch: &EntryBatch<P>,
) -> impl Future<Output = Result<Vec<Recorded>, Self::Error>> + Send {
let result =
self.with_mut(|journal| Ok(journal.record_batch(batch.entries().iter().cloned())?));
async move { result }
}
fn get(
&self,
id: EntryId,
) -> impl Future<Output = Result<Option<StoredEntry<P>>, Self::Error>> + Send {
let result = self.with(|journal| {
let found = journal.index_of(id).and_then(|index| {
journal.at(index).map(|entry| StoredEntry {
index: Some(index),
entry: entry.clone(),
content_hash: entry.content_hash(),
})
});
Ok(found)
});
async move { result }
}
fn page(&self, cursor: Cursor) -> impl Future<Output = Result<Page<P>, Self::Error>> + Send {
let result = self.with(|journal| {
let start = cursor.after.map_or(0usize, |i| {
usize::try_from(i.get().saturating_add(1)).unwrap_or(usize::MAX)
});
let limit = cursor.effective_limit();
let mut records = Vec::new();
for (offset, entry) in journal.entries().iter().skip(start).take(limit).enumerate() {
let index = start.saturating_add(offset);
records.push(StoredEntry {
index: Some(LogIndex::new(index as u64)),
entry: entry.clone(),
content_hash: entry.content_hash(),
});
}
let next = records
.last()
.filter(|_| start.saturating_add(records.len()) < journal.len())
.and_then(|r| r.index)
.map(|index| Cursor {
after: Some(index),
limit: cursor.limit,
});
Ok(Page { records, next })
});
async move { result }
}
fn head(&self) -> impl Future<Output = Result<TreeHead, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.head()));
async move { result }
}
fn len(&self) -> impl Future<Output = Result<u64, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.len() as u64));
async move { result }
}
fn balance(
&self,
key: BalanceKey,
size: Option<u64>,
) -> impl Future<Output = Result<Balance<P>, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.balance(&key, size)?));
async move { result }
}
fn trial_balance(
&self,
size: Option<u64>,
) -> impl Future<Output = Result<TrialBalance<P>, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.trial_balance(size)?));
async move { result }
}
fn prove_inclusion(
&self,
index: LogIndex,
) -> impl Future<Output = Result<InclusionProof, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.prove_inclusion(index)?));
async move { result }
}
fn prove_consistency(
&self,
old_size: u64,
) -> impl Future<Output = Result<ConsistencyProof, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.prove_consistency(old_size)?));
async move { result }
}
fn define_period(
&self,
period: &Period,
) -> impl Future<Output = Result<(), Self::Error>> + Send {
let period = period.clone();
let result = self.with_mut(|journal| {
journal.calendar_mut().ensure(period)?;
Ok(())
});
async move { result }
}
fn transition_period(
&self,
period: &PeriodId,
to: PeriodState,
) -> impl Future<Output = Result<(), Self::Error>> + Send {
let result = self.with_mut(|journal| Ok(journal.transition_period(period, to)?));
async move { result }
}
fn periods(&self) -> impl Future<Output = Result<Vec<Period>, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.calendar().iter().cloned().collect()));
async move { result }
}
fn seal_period(
&self,
period: &PeriodId,
) -> impl Future<Output = Result<Seal, Self::Error>> + Send {
let result = self.with_mut(|journal| Ok(journal.seal_period(period)?));
async move { result }
}
fn seals(&self) -> impl Future<Output = Result<Vec<Seal>, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.seals().seals().to_vec()));
async move { result }
}
fn clear(&self, clearing: Clearing<P>) -> impl Future<Output = Result<(), Self::Error>> + Send {
let result = self.with_mut(|journal| Ok(journal.clear(clearing)?));
async move { result }
}
fn reset_clearing(
&self,
id: ClearingId,
on: Date,
) -> impl Future<Output = Result<(), Self::Error>> + Send {
let result = self.with_mut(|journal| Ok(journal.reset_clearing(id, on)?));
async move { result }
}
fn open_items(
&self,
key: BalanceKey,
) -> impl Future<Output = Result<Vec<OpenItem<P>>, Self::Error>> + Send {
let result = self.with(|journal| Ok(journal.open_items(&key)?));
async move { result }
}
fn balances(
&self,
accounts: &[AccountId],
currency: Currency,
layer: Layer,
size: Option<u64>,
) -> impl Future<Output = Result<BTreeMap<AccountId, Balance<P>>, Self::Error>> + Send {
let wanted: Vec<AccountId> = accounts.to_vec();
let result = self.with(|journal| {
let tb = journal.trial_balance(size)?;
let mut out = BTreeMap::new();
for account in wanted {
let key = BalanceKey {
account,
currency,
layer,
};
if let Some(balance) = tb.get(&key) {
out.insert(account, *balance);
}
}
Ok(out)
});
async move { result }
}
fn statement(
&self,
key: BalanceKey,
cursor: Cursor,
) -> impl Future<Output = Result<StatementPage<P>, Self::Error>> + Send {
let result = self.with(|journal| {
let all = journal.statement(&key)?;
let start = cursor.after.map_or(0usize, |after| {
all.iter()
.position(|line| line.index.get() > after.get())
.unwrap_or(all.len())
});
let limit = cursor.effective_limit();
let lines: Vec<StatementLine<P>> = all
.iter()
.skip(start)
.take(limit)
.map(|l| StatementLine {
index: l.index,
posting: l.posting,
booking_date: l.booking_date,
direction: l.direction,
amount: l.amount,
running: l.running,
kind: l.kind.clone(),
})
.collect();
let next = lines
.last()
.filter(|_| start.saturating_add(lines.len()) < all.len())
.map(|l| Cursor {
after: Some(l.index),
limit: cursor.limit,
});
Ok(StatementPage { lines, next })
});
async move { result }
}
fn save_checkpoint(
&self,
checkpoint: &Checkpoint<P>,
) -> impl Future<Output = Result<(), Self::Error>> + Send {
let checkpoint = *checkpoint;
let mut guard = self.checkpoints.lock().unwrap_or_else(|e| e.into_inner());
guard.insert(checkpoint.key, checkpoint);
async move { Ok(()) }
}
fn load_checkpoint(
&self,
key: BalanceKey,
) -> impl Future<Output = Result<Option<Checkpoint<P>>, Self::Error>> + Send {
let guard = self.checkpoints.lock().unwrap_or_else(|e| e.into_inner());
let found = guard.get(&key).copied();
async move { Ok(found) }
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::entry::Draft;
use crate::storage::conformance::test_ledger;
#[test]
fn a_batch_needs_at_least_one_entry() {
let empty: Vec<Entry<Balanced, 2>> = Vec::new();
assert_eq!(EntryBatch::new(empty), Err(BatchError::Empty));
}
#[test]
fn cursor_limits_are_clamped() {
assert_eq!(Cursor::start().effective_limit(), DEFAULT_PAGE_SIZE);
assert_eq!(Cursor::start().with_limit(0).effective_limit(), 1);
assert_eq!(
Cursor::start().with_limit(usize::MAX).effective_limit(),
MAX_PAGE_SIZE
);
}
#[test]
fn the_memory_store_is_usable_behind_dyn() {
let store: Box<dyn DynLedgerStore<2, Error = MemoryStoreError>> =
Box::new(MemoryStore::<2>::new(test_ledger()));
let head = conformance::block_on(store.head_boxed()).expect("reads");
assert_eq!(head.size, 0);
}
#[test]
fn drafts_are_not_storable() {
fn _accepts_only_balanced<const P: u8>(_: Vec<Entry<Balanced, P>>) {}
let _ = std::marker::PhantomData::<Entry<Draft, 2>>;
}
}