//! Writing a plan into the database (SPEC §10.8): one `BEGIN IMMEDIATE` transaction, the account found or created, //! rows whose `import_key` exists skipped, every other row validated with the server's own push checks //! (`kvitt_server::sync::check`), inserted with a new `server_version`, and audited through [`AuditWriter`] with //! `source` `import`. Nothing goes through a push queue: the importer has none. use std::collections::BTreeMap; use chrono::{DateTime, Utc}; use kvitt_domain::{BalanceExpense, BalanceSettlement, compute_balances}; use kvitt_server::api::accounts::{self, name_key}; use kvitt_server::api::rows::{self, Origin}; use kvitt_server::audit::{AuditEvent, AuditWriter}; use kvitt_server::db as server_db; use kvitt_server::error::AppError; use kvitt_server::sync::check::{self, Fail}; use kvitt_server::time::format_ts; use kvitt_server::wire::{ Allocation, AuditAction, AuditSource, EntityType, PushedExpense, PushedSettlement, Split, SplitMode, User, }; use sqlx::SqliteConnection; use uuid::Uuid; use crate::parse::RowProblem; use crate::plan::{PlannedBody, PlannedExpense, PlannedRow, PlannedSettlement, PlannedSplit}; /// Default ratio of a created account: 50/50 (SPEC §10.8). pub const DEFAULT_SHARE_BP: i64 = 5000; /// What writing produced (before the caller commits or rolls back). #[derive(Debug, Default)] pub struct Written { pub account_id: String, pub account_created: bool, pub expenses_created: usize, pub settlements_created: usize, pub skipped_duplicates: usize, /// Rows the server's checks rejected; non-empty means the caller must roll back. pub rejected: Vec, } /// The household context of a run. pub struct Household<'a> { pub users: &'a [User], pub owner: Uuid, } fn uuid(text: &str) -> Result { rows::uuid(text) } /// Finds the non-deleted account with this name (case-insensitive, SPEC §14.1 `name_key`), else creates it with /// the default 50/50 ratio, created by the owner. async fn find_or_create_account( conn: &mut SqliteConnection, name: &str, household: &Household<'_>, now_ts: &str, ) -> Result<(String, bool), AppError> { let key = name_key(name); let existing: Option = sqlx::query_scalar("SELECT id FROM accounts WHERE name_key = ? AND deleted_at IS NULL") .bind(&key) .fetch_optional(&mut *conn) .await?; if let Some(id) = existing { return Ok((id, false)); } let id = Uuid::now_v7(); let version = server_db::next_server_version(conn).await?; accounts::insert_account(conn, id, name, household.owner, now_ts, version).await?; let shares: Vec<(Uuid, i64)> = household .users .iter() .map(|user| (user.id, DEFAULT_SHARE_BP)) .collect(); accounts::replace_shares(conn, id, &shares).await?; Ok((id.to_string(), true)) } /// The id of each non-deleted default category, by key. async fn default_categories( conn: &mut SqliteConnection, ) -> Result, AppError> { let pairs: Vec<(String, String)> = sqlx::query_as( "SELECT default_key, id FROM categories WHERE default_key IS NOT NULL AND deleted_at IS NULL", ) .fetch_all(&mut *conn) .await?; Ok(pairs.into_iter().collect()) } async fn key_exists(conn: &mut SqliteConnection, key: &str) -> Result { let exists: bool = sqlx::query_scalar( "SELECT EXISTS (SELECT 1 FROM expenses WHERE import_key = ?1) \ OR EXISTS (SELECT 1 FROM settlements WHERE import_key = ?1)", ) .bind(key) .fetch_one(&mut *conn) .await?; Ok(exists) } fn pushed_expense( e: &PlannedExpense, users: &[User], account_id: Uuid, category_id: Uuid, now_ts: &str, ) -> Result { let (split_mode, splits) = match &e.split { PlannedSplit::Exact(shares) => ( SplitMode::Exact, shares .iter() .map(|(user, share)| { Ok(Split { user_id: uuid(user)?, share_ore: *share, share_bp: None, }) }) .collect::, AppError>>()?, ), PlannedSplit::Full(bearer) => (SplitMode::Full, full_splits(e.amount_ore, bearer, users)?), }; Ok(PushedExpense { id: Uuid::now_v7(), account_id, category_id, amount_ore: e.amount_ore, description: e.description.clone(), notes: e.notes.clone(), expense_date: e.date.clone(), payer_user_id: uuid(&e.payer_user_id)?, split_mode, splits, created_at: now_ts.to_owned(), updated_at: now_ts.to_owned(), deleted_at: None, }) } /// The splits of mode `full` (SPEC §9.1: bearer 10000 bp and the whole amount, the other user 0). fn full_splits(amount_ore: i64, bearer: &str, users: &[User]) -> Result, AppError> { users .iter() .map(|u| { let bears = u.id.to_string() == bearer; Ok(Split { user_id: u.id, share_ore: if bears { amount_ore } else { 0 }, share_bp: Some(if bears { 10_000 } else { 0 }), }) }) .collect() } fn pushed_settlement( s: &PlannedSettlement, account_id: Uuid, now_ts: &str, ) -> Result { Ok(PushedSettlement { id: Uuid::now_v7(), from_user_id: uuid(&s.from_user_id)?, to_user_id: uuid(&s.to_user_id)?, amount_ore: s.amount_ore, settlement_date: s.date.clone(), note: s.note.clone(), allocations: vec![Allocation { account_id, amount_ore: s.amount_ore, }], created_at: now_ts.to_owned(), updated_at: now_ts.to_owned(), deleted_at: None, }) } /// The server-owned columns of an imported row: created by the owner, no device, a new `server_version`. async fn origin<'a>( conn: &mut SqliteConnection, import_key: &'a str, owner: Uuid, now_ts: &'a str, ) -> Result, AppError> { Ok(Origin { user_id: owner, device_id: None, import_key: Some(import_key), received_at: now_ts, server_version: server_db::next_server_version(conn).await?, }) } async fn audit_create( conn: &mut SqliteConnection, audit: &dyn AuditWriter, entity_type: EntityType, entity_id: Uuid, after: serde_json::Value, owner: Uuid, now_ts: &str, ) -> Result<(), AppError> { audit .write( conn, AuditEvent { entity_type, entity_id, action: AuditAction::Create, source: AuditSource::Import, actor_user_id: owner, actor_device_id: None, client_at: now_ts.to_owned(), server_at: now_ts.to_owned(), before: None, after, }, ) .await?; Ok(()) } /// Turns a server check failure into a row problem, or passes a real error through. fn rejected(line: u64, fail: Fail) -> Result { match fail { Fail::Reject(r) => Ok(RowProblem { line, reason: format!("{}: {}", rows::token_of(&r.reason)?, r.message), }), Fail::Internal(e) => Err(e), } } /// Writes the planned rows on `conn` (inside the caller's `BEGIN IMMEDIATE` transaction). pub async fn write( conn: &mut SqliteConnection, account_name: &str, planned: &[PlannedRow], household: &Household<'_>, audit: &dyn AuditWriter, now: DateTime, ) -> Result { let now_ts = format_ts(now); let (account_id, account_created) = find_or_create_account(conn, account_name, household, &now_ts).await?; let account_uuid = uuid(&account_id)?; let categories = default_categories(conn).await?; let owner = household.owner; let mut written = Written { account_id, account_created, ..Written::default() }; for row in planned { if key_exists(conn, &row.import_key).await? { written.skipped_duplicates += 1; continue; } match &row.body { PlannedBody::Expense(e) => { let category_id = categories.get(&e.category_key).ok_or_else(|| { AppError::internal(format!("default category {} is missing", e.category_key)) })?; let pushed = pushed_expense( e, household.users, account_uuid, uuid(category_id)?, &now_ts, )?; let checked = match check::expense(conn, &pushed, household.users, now).await { Ok(checked) => checked, Err(fail) => { written.rejected.push(rejected(row.line, fail)?); continue; } }; let origin = origin(conn, &row.import_key, owner, &now_ts).await?; rows::insert_expense(conn, &checked, None, origin).await?; let stored = rows::expense(conn, checked.id) .await? .ok_or_else(|| AppError::internal("imported expense vanished"))?; let after = serde_json::to_value(&stored).map_err(AppError::internal)?; audit_create( conn, audit, EntityType::Expense, checked.id, after, owner, &now_ts, ) .await?; written.expenses_created += 1; } PlannedBody::Settlement(s) => { let pushed = pushed_settlement(s, account_uuid, &now_ts)?; let checked = match check::settlement(conn, &pushed, household.users, now).await { Ok(checked) => checked, Err(fail) => { written.rejected.push(rejected(row.line, fail)?); continue; } }; let origin = origin(conn, &row.import_key, owner, &now_ts).await?; rows::insert_settlement(conn, &checked, origin).await?; let stored = rows::settlement(conn, checked.id) .await? .ok_or_else(|| AppError::internal("imported settlement vanished"))?; let after = serde_json::to_value(&stored).map_err(AppError::internal)?; audit_create( conn, audit, EntityType::Settlement, checked.id, after, owner, &now_ts, ) .await?; written.settlements_created += 1; } } } Ok(written) } /// The account's balance per user over its non-deleted expenses and settlement allocations (SPEC §9.3), computed /// with `kvitt_domain::compute_balances`. pub async fn account_balance( conn: &mut SqliteConnection, account_id: &str, user_ids: &[String], ) -> Result, AppError> { let heads: Vec<(String, i64, String)> = sqlx::query_as( "SELECT id, amount_ore, payer_user_id FROM expenses WHERE account_id = ? AND deleted_at IS NULL", ) .bind(account_id) .fetch_all(&mut *conn) .await?; let splits: Vec<(String, String, i64)> = sqlx::query_as( "SELECT s.expense_id, s.user_id, s.share_ore FROM expense_splits s JOIN expenses e ON e.id = s.expense_id \ WHERE e.account_id = ? AND e.deleted_at IS NULL", ) .bind(account_id) .fetch_all(&mut *conn) .await?; let mut shares: BTreeMap> = BTreeMap::new(); for (expense_id, user_id, share) in splits { shares.entry(expense_id).or_default().insert(user_id, share); } let expenses: Vec = heads .into_iter() .map(|(id, amount_ore, payer_user_id)| BalanceExpense { account_id: account_id.to_owned(), amount_ore, payer_user_id, shares: shares.remove(&id).unwrap_or_default(), }) .collect(); let allocations: Vec<(String, String, i64)> = sqlx::query_as( "SELECT s.from_user_id, s.to_user_id, a.amount_ore FROM settlement_allocations a \ JOIN settlements s ON s.id = a.settlement_id WHERE a.account_id = ? AND s.deleted_at IS NULL", ) .bind(account_id) .fetch_all(&mut *conn) .await?; let settlements: Vec = allocations .into_iter() .map(|(from_user_id, to_user_id, amount)| BalanceSettlement { from_user_id, to_user_id, allocations: BTreeMap::from([(account_id.to_owned(), amount)]), }) .collect(); let balances = compute_balances(user_ids, &[account_id.to_owned()], &expenses, &settlements); Ok(balances .per_account .get(account_id) .cloned() .unwrap_or_default()) }