diff --git a/server/crates/kvitt-import/src/db.rs b/server/crates/kvitt-import/src/db.rs index f0980b0..1b25404 100644 --- a/server/crates/kvitt-import/src/db.rs +++ b/server/crates/kvitt-import/src/db.rs @@ -7,12 +7,12 @@ use std::collections::BTreeMap; use chrono::{DateTime, Utc}; use kvitt_domain::{BalanceExpense, BalanceSettlement, compute_balances}; -use kvitt_server::api::{accounts::name_key, rows}; +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::sync::rows as sync_rows; use kvitt_server::time::format_ts; use kvitt_server::wire::{ Allocation, AuditAction, AuditSource, EntityType, PushedExpense, PushedSettlement, Split, @@ -66,32 +66,16 @@ async fn find_or_create_account( if let Some(id) = existing { return Ok((id, false)); } - let id = Uuid::now_v7().to_string(); + let id = Uuid::now_v7(); let version = server_db::next_server_version(conn).await?; - sqlx::query( - "INSERT INTO accounts (id, name, name_key, created_by_user_id, created_at, updated_at, archived_at, \ - deleted_at, server_version) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?)", - ) - .bind(&id) - .bind(name) - .bind(&key) - .bind(household.owner.to_string()) - .bind(now_ts) - .bind(now_ts) - .bind(version) - .execute(&mut *conn) - .await?; - for user in household.users { - sqlx::query( - "INSERT INTO account_default_shares (account_id, user_id, share_bp) VALUES (?, ?, ?)", - ) - .bind(&id) - .bind(user.id.to_string()) - .bind(DEFAULT_SHARE_BP) - .execute(&mut *conn) - .await?; - } - Ok((id, true)) + 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. @@ -194,91 +178,20 @@ fn pushed_settlement( }) } -async fn insert_expense( +/// 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, - p: &PushedExpense, - import_key: &str, + import_key: &'a str, owner: Uuid, - now_ts: &str, -) -> Result<(), AppError> { - let version = server_db::next_server_version(conn).await?; - sqlx::query( - "INSERT INTO expenses (id, account_id, category_id, amount_ore, description, notes, expense_date, \ - payer_user_id, split_mode, recurring_template_id, recurring_period, import_key, created_by_user_id, \ - updated_by_user_id, updated_by_device_id, created_at, updated_at, deleted_at, server_received_at, \ - server_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, ?, NULL, ?, ?, NULL, ?, ?)", - ) - .bind(p.id.to_string()) - .bind(p.account_id.to_string()) - .bind(p.category_id.to_string()) - .bind(p.amount_ore) - .bind(&p.description) - .bind(&p.notes) - .bind(&p.expense_date) - .bind(p.payer_user_id.to_string()) - .bind(rows::token_of(&p.split_mode)?) - .bind(import_key) - .bind(owner.to_string()) - .bind(owner.to_string()) - .bind(now_ts) - .bind(now_ts) - .bind(now_ts) - .bind(version) - .execute(&mut *conn) - .await?; - for split in &p.splits { - sqlx::query( - "INSERT INTO expense_splits (expense_id, user_id, share_ore, share_bp) VALUES (?, ?, ?, ?)", - ) - .bind(p.id.to_string()) - .bind(split.user_id.to_string()) - .bind(split.share_ore) - .bind(split.share_bp) - .execute(&mut *conn) - .await?; - } - Ok(()) -} - -async fn insert_settlement( - conn: &mut SqliteConnection, - p: &PushedSettlement, - import_key: &str, - owner: Uuid, - now_ts: &str, -) -> Result<(), AppError> { - let version = server_db::next_server_version(conn).await?; - sqlx::query( - "INSERT INTO settlements (id, from_user_id, to_user_id, amount_ore, settlement_date, note, import_key, \ - created_by_user_id, updated_by_user_id, updated_by_device_id, created_at, updated_at, deleted_at, \ - server_received_at, server_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, NULL, ?, ?)", - ) - .bind(p.id.to_string()) - .bind(p.from_user_id.to_string()) - .bind(p.to_user_id.to_string()) - .bind(p.amount_ore) - .bind(&p.settlement_date) - .bind(&p.note) - .bind(import_key) - .bind(owner.to_string()) - .bind(owner.to_string()) - .bind(now_ts) - .bind(now_ts) - .bind(now_ts) - .bind(version) - .execute(&mut *conn) - .await?; - for allocation in &p.allocations { - sqlx::query( - "INSERT INTO settlement_allocations (settlement_id, account_id, amount_ore) VALUES (?, ?, ?)", - ) - .bind(p.id.to_string()) - .bind(allocation.account_id.to_string()) - .bind(allocation.amount_ore) - .execute(&mut *conn) - .await?; - } - Ok(()) + 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( @@ -365,7 +278,8 @@ pub async fn write( continue; } }; - insert_expense(conn, &checked, &row.import_key, owner, &now_ts).await?; + 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"))?; @@ -391,8 +305,9 @@ pub async fn write( continue; } }; - insert_settlement(conn, &checked, &row.import_key, owner, &now_ts).await?; - let stored = sync_rows::settlement(conn, checked.id) + 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)?; diff --git a/server/crates/kvitt-server/src/api/accounts.rs b/server/crates/kvitt-server/src/api/accounts.rs index db34c72..ec8c92a 100644 --- a/server/crates/kvitt-server/src/api/accounts.rs +++ b/server/crates/kvitt-server/src/api/accounts.rs @@ -51,7 +51,8 @@ async fn check_name_free( } } -async fn replace_shares( +/// Replaces the account's `account_default_shares` rows. +pub async fn replace_shares( conn: &mut SqliteConnection, account_id: Uuid, shares: &[(Uuid, i64)], @@ -73,6 +74,32 @@ async fn replace_shares( Ok(()) } +/// Inserts a live, unarchived account row (without its default shares, see [`replace_shares`]); `name` is the +/// trimmed name. +pub async fn insert_account( + conn: &mut SqliteConnection, + id: Uuid, + name: &str, + created_by: Uuid, + now_ts: &str, + server_version: i64, +) -> Result<(), AppError> { + sqlx::query( + "INSERT INTO accounts (id, name, name_key, created_by_user_id, created_at, updated_at, archived_at, \ + deleted_at, server_version) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?)", + ) + .bind(id.to_string()) + .bind(name) + .bind(name_key(name)) + .bind(created_by.to_string()) + .bind(now_ts) + .bind(now_ts) + .bind(server_version) + .execute(&mut *conn) + .await?; + Ok(()) +} + async fn reload(conn: &mut SqliteConnection, id: Uuid) -> Result { rows::account(conn, id) .await? @@ -107,19 +134,7 @@ pub async fn create_account( let id = Uuid::now_v7(); let now = state.clock.now_ts(); let version = db::next_server_version(&mut tx).await?; - sqlx::query( - "INSERT INTO accounts (id, name, name_key, created_by_user_id, created_at, updated_at, archived_at, \ - deleted_at, server_version) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?)", - ) - .bind(id.to_string()) - .bind(&name) - .bind(&key) - .bind(device.user_id.to_string()) - .bind(&now) - .bind(&now) - .bind(version) - .execute(&mut *tx) - .await?; + insert_account(&mut tx, id, &name, device.user_id, &now, version).await?; replace_shares(&mut tx, id, &shares).await?; let account = reload(&mut tx, id).await?; tx.commit().await?; diff --git a/server/crates/kvitt-server/src/api/recurring.rs b/server/crates/kvitt-server/src/api/recurring.rs index c5ab0eb..254886d 100644 --- a/server/crates/kvitt-server/src/api/recurring.rs +++ b/server/crates/kvitt-server/src/api/recurring.rs @@ -1,7 +1,7 @@ //! Recurring templates: E-28 to E-30 (FR-9) and the generation of due periods (SPEC §9.5, D-015). //! //! [`generate_due_periods`] is the one function that creates the expenses of a template's due periods. E-28 and E-29 -//! call it before responding; M4's scheduler calls it for every template (or [`generate_all_due`]). +//! call it before responding; the daily scheduler (`tasks::scheduler`) calls [`generate_all_due`]. use std::collections::BTreeMap; @@ -13,7 +13,8 @@ use sqlx::SqliteConnection; use uuid::Uuid; use crate::AppState; -use crate::api::{ApiJson, IdPath, JsonBody, rows, validate}; +use crate::api::rows::{self, Origin}; +use crate::api::{ApiJson, IdPath, JsonBody, validate}; use crate::audit::AuditEvent; use crate::auth::AuthDevice; use crate::db; @@ -23,7 +24,8 @@ use crate::push::{PushMessage, PushType}; use crate::time::{STOCKHOLM, format_ts}; use crate::wire::{ AuditAction, AuditSource, CreateRecurringTemplateRequest, EntityType, - PatchRecurringTemplateRequest, RecurringTemplate, SplitMode, TemplateSplit, User, UuidSchema, + PatchRecurringTemplateRequest, PushedExpense, RecurringTemplate, Split, SplitMode, + TemplateSplit, User, UuidSchema, }; /// Local time of day at which a period's expense is created (SPEC §9.5, AC-9.2). @@ -56,12 +58,12 @@ pub struct GeneratedExpense { pub period: String, } -/// The expense splits of a generated expense: `(user_id, share_ore, share_bp)`. +/// The expense splits of a generated expense. async fn resolve_splits( conn: &mut SqliteConnection, template: &RecurringTemplate, users: &[User], -) -> Result)>, AppError> { +) -> Result, AppError> { let ids = rows::user_ids(users); let payer = template.payer_user_id.to_string(); let bp: BTreeMap = match template.split_mode { @@ -116,7 +118,13 @@ async fn resolve_splits( })?; return shares .into_iter() - .map(|(user, ore)| Ok((rows::uuid(&user)?, ore, None))) + .map(|(user, share_ore)| { + Ok(Split { + user_id: rows::uuid(&user)?, + share_ore, + share_bp: None, + }) + }) .collect(); } }; @@ -130,13 +138,84 @@ async fn resolve_splits( .map_err(|e| AppError::internal(format!("stored template split is invalid: {e}")))?; shares .into_iter() - .map(|(user, ore)| { + .map(|(user, share_ore)| { let share_bp = bp.get(&user).map(|v| i64::from(*v)); - Ok((rows::uuid(&user)?, ore, share_bp)) + Ok(Split { + user_id: rows::uuid(&user)?, + share_ore, + share_bp, + }) }) .collect() } +/// Creates the expense of one due `period` of `template` inside the caller's transaction, with its audit entry +/// (see [`generate_due_periods`]). Returns the new expense id. +async fn create_period_expense( + state: &AppState, + conn: &mut SqliteConnection, + template: &RecurringTemplate, + period: Period, + day: u8, + now_ts: &str, +) -> Result { + let users = rows::complete_household(conn).await?; + let splits = resolve_splits(conn, template, &users).await?; + let period_text = period.to_string(); + let expense = PushedExpense { + id: Uuid::now_v7(), + account_id: template.account_id, + category_id: template.category_id, + amount_ore: template.amount_ore, + description: template.description.clone(), + notes: template.notes.clone(), + expense_date: period.clamped_date(day).map_err(AppError::internal)?, + payer_user_id: template.payer_user_id, + split_mode: template.split_mode, + splits, + created_at: now_ts.to_owned(), + updated_at: now_ts.to_owned(), + deleted_at: None, + }; + let version = db::next_server_version(conn).await?; + rows::insert_expense( + conn, + &expense, + Some((template.id, &period_text)), + Origin { + user_id: template.created_by_user_id, + device_id: None, + import_key: None, + received_at: now_ts, + server_version: version, + }, + ) + .await?; + let stored = rows::expense(conn, expense.id) + .await? + .ok_or_else(|| AppError::internal("generated expense vanished"))?; + let after = serde_json::to_value(&stored).map_err(AppError::internal)?; + state + .audit + .write( + conn, + AuditEvent { + entity_type: EntityType::Expense, + entity_id: expense.id, + action: AuditAction::Create, + source: AuditSource::Recurring, + actor_user_id: template.created_by_user_id, + actor_device_id: None, + client_at: now_ts.to_owned(), + server_at: now_ts.to_owned(), + before: None, + after, + }, + ) + .await?; + Ok(expense.id) +} + /// Generates every due period of one template (SPEC §9.5), oldest first, each in its own `BEGIN IMMEDIATE` /// transaction: the expense (server UUIDv7, `expense_date` the clamped date, fields copied from the template, /// created and updated by the template's creator, `updated_by_device_id` null), its splits (mode `default` resolves @@ -183,72 +262,11 @@ pub async fn generate_due_periods( .fetch_one(&mut *tx) .await?; let now_ts = format_ts(now); - let mut created = None; - if !exists { - let users = rows::complete_household(&mut tx).await?; - let splits = resolve_splits(&mut tx, &template, &users).await?; - let expense_id = Uuid::now_v7(); - let version = db::next_server_version(&mut tx).await?; - sqlx::query( - "INSERT INTO expenses (id, account_id, category_id, amount_ore, description, notes, expense_date, \ - payer_user_id, split_mode, recurring_template_id, recurring_period, import_key, created_by_user_id, \ - updated_by_user_id, updated_by_device_id, created_at, updated_at, deleted_at, server_received_at, \ - server_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, NULL, ?, ?, NULL, ?, ?)", - ) - .bind(expense_id.to_string()) - .bind(template.account_id.to_string()) - .bind(template.category_id.to_string()) - .bind(template.amount_ore) - .bind(&template.description) - .bind(&template.notes) - .bind(period.clamped_date(day).map_err(AppError::internal)?) - .bind(template.payer_user_id.to_string()) - .bind(rows::token_of(&template.split_mode)?) - .bind(template_id.to_string()) - .bind(&period_text) - .bind(template.created_by_user_id.to_string()) - .bind(template.created_by_user_id.to_string()) - .bind(&now_ts) - .bind(&now_ts) - .bind(&now_ts) - .bind(version) - .execute(&mut *tx) - .await?; - for (user_id, share_ore, share_bp) in &splits { - sqlx::query( - "INSERT INTO expense_splits (expense_id, user_id, share_ore, share_bp) VALUES (?, ?, ?, ?)", - ) - .bind(expense_id.to_string()) - .bind(user_id.to_string()) - .bind(share_ore) - .bind(share_bp) - .execute(&mut *tx) - .await?; - } - let expense = rows::expense(&mut tx, expense_id) - .await? - .ok_or_else(|| AppError::internal("generated expense vanished"))?; - let after = serde_json::to_value(&expense).map_err(AppError::internal)?; - state - .audit - .write( - &mut tx, - AuditEvent { - entity_type: EntityType::Expense, - entity_id: expense_id, - action: AuditAction::Create, - source: AuditSource::Recurring, - actor_user_id: template.created_by_user_id, - actor_device_id: None, - client_at: now_ts.clone(), - server_at: now_ts.clone(), - before: None, - after, - }, - ) - .await?; - created = Some(expense_id); - } + let created = if exists { + None + } else { + Some(create_period_expense(state, &mut tx, &template, period, day, &now_ts).await?) + }; let version = db::next_server_version(&mut tx).await?; sqlx::query( "UPDATE recurring_templates SET last_generated_period = ?, updated_at = ?, server_version = ? WHERE id = ?", diff --git a/server/crates/kvitt-server/src/api/rows.rs b/server/crates/kvitt-server/src/api/rows.rs index 7e1bb37..f7ca378 100644 --- a/server/crates/kvitt-server/src/api/rows.rs +++ b/server/crates/kvitt-server/src/api/rows.rs @@ -1,4 +1,5 @@ -//! Reading stored rows into their wire form (SPEC §14.3.3), shared by the M3 handlers and the recurring generation. +//! Stored rows (SPEC §14.1): reading them into their wire form (SPEC §14.3.3) and the inserts that more than one +//! writer shares (REST handlers, recurring generation, sync push, the importer). //! //! Every function takes a connection, so it works on the pool (via `acquire`) and inside a write transaction. @@ -11,8 +12,9 @@ use uuid::Uuid; use crate::error::{AppError, ErrorCode}; use crate::wire::{ - Account, Category, Device, Expense, RecurringTemplate, Role, Share, Split, SplitMode, - TemplateSplit, User, + Account, Allocation, AuditChange, AuditEntry, Category, Device, Expense, PushedExpense, + PushedSettlement, RecurringTemplate, Role, Settlement, Share, Split, SplitMode, TemplateSplit, + User, }; /// Parses a stored UUID. @@ -25,7 +27,7 @@ fn opt_uuid(text: Option) -> Result, AppError> { } /// Parses a stored snake_case enum token through its serde form. -fn token(text: &str) -> Result { +pub fn token(text: &str) -> Result { serde_json::from_value(serde_json::Value::String(text.to_owned())) .map_err(|e| AppError::internal(format!("stored token {text:?}: {e}"))) } @@ -546,3 +548,257 @@ pub fn bp_shares( .map_err(|e| fail(e.token()))?; Ok(bp) } + +// --------------------------------------------------------------------------------------------------------------- +// Settlements +// --------------------------------------------------------------------------------------------------------------- + +#[derive(FromRow)] +struct SettlementRow { + id: String, + from_user_id: String, + to_user_id: String, + amount_ore: i64, + settlement_date: String, + note: String, + created_by_user_id: String, + updated_by_user_id: String, + created_at: String, + updated_at: String, + deleted_at: Option, + server_version: i64, +} + +/// One settlement (tombstones included), with its allocations sorted by `account_id`. +pub async fn settlement( + conn: &mut SqliteConnection, + id: Uuid, +) -> Result, AppError> { + let Some(row) = sqlx::query_as::<_, SettlementRow>( + "SELECT id, from_user_id, to_user_id, amount_ore, settlement_date, note, created_by_user_id, \ + updated_by_user_id, created_at, updated_at, deleted_at, server_version FROM settlements WHERE id = ?", + ) + .bind(id.to_string()) + .fetch_optional(&mut *conn) + .await? + else { + return Ok(None); + }; + let allocations: Vec<(String, i64)> = sqlx::query_as( + "SELECT account_id, amount_ore FROM settlement_allocations WHERE settlement_id = ? ORDER BY account_id", + ) + .bind(id.to_string()) + .fetch_all(&mut *conn) + .await?; + let mut settlement = Settlement { + id: uuid(&row.id)?, + from_user_id: uuid(&row.from_user_id)?, + to_user_id: uuid(&row.to_user_id)?, + amount_ore: row.amount_ore, + settlement_date: row.settlement_date, + note: row.note, + allocations: allocations + .into_iter() + .map(|(account_id, amount_ore)| { + Ok(Allocation { + account_id: uuid(&account_id)?, + amount_ore, + }) + }) + .collect::>()?, + created_by_user_id: uuid(&row.created_by_user_id)?, + updated_by_user_id: uuid(&row.updated_by_user_id)?, + created_at: row.created_at, + updated_at: row.updated_at, + deleted_at: row.deleted_at, + server_version: row.server_version, + }; + settlement.sort(); + Ok(Some(settlement)) +} + +// --------------------------------------------------------------------------------------------------------------- +// Audit entries and the epoch +// --------------------------------------------------------------------------------------------------------------- + +#[derive(FromRow)] +struct AuditRow { + id: String, + entity_type: String, + entity_id: String, + action: String, + source: String, + actor_user_id: String, + client_at: String, + server_at: String, + changes: String, + server_version: i64, +} + +/// One audit entry in its wire form (SPEC §14.3.3; `actor_device_id` is not on the wire). +pub async fn audit_entry( + conn: &mut SqliteConnection, + id: Uuid, +) -> Result, AppError> { + let Some(row) = sqlx::query_as::<_, AuditRow>( + "SELECT id, entity_type, entity_id, action, source, actor_user_id, client_at, server_at, changes, \ + server_version FROM audit_log WHERE id = ?", + ) + .bind(id.to_string()) + .fetch_optional(&mut *conn) + .await? + else { + return Ok(None); + }; + let changes: Vec = serde_json::from_str(&row.changes) + .map_err(|e| AppError::internal(format!("stored audit changes: {e}")))?; + Ok(Some(AuditEntry { + id: uuid(&row.id)?, + entity_type: token(&row.entity_type)?, + entity_id: uuid(&row.entity_id)?, + action: token(&row.action)?, + source: token(&row.source)?, + actor_user_id: uuid(&row.actor_user_id)?, + client_at: row.client_at, + server_at: row.server_at, + changes, + server_version: row.server_version, + })) +} + +/// `meta.sync_epoch`. +pub async fn epoch(conn: &mut SqliteConnection) -> Result { + let value: Option = + sqlx::query_scalar("SELECT value FROM meta WHERE key = 'sync_epoch'") + .fetch_optional(&mut *conn) + .await?; + value.ok_or_else(|| AppError::internal("meta.sync_epoch is missing")) +} + +// --------------------------------------------------------------------------------------------------------------- +// Inserts shared by several writers +// --------------------------------------------------------------------------------------------------------------- + +/// The server-owned columns of a new expense or settlement row. +#[derive(Debug, Clone, Copy)] +pub struct Origin<'a> { + /// `created_by_user_id` and `updated_by_user_id`. + pub user_id: Uuid, + /// `updated_by_device_id`: the pushing device; `None` for recurring generation and the importer. + pub device_id: Option, + /// `import_key` (the importer only). + pub import_key: Option<&'a str>, + /// `server_received_at`: server now. + pub received_at: &'a str, + /// The row's `server_version`, taken by the caller in its write transaction. + pub server_version: i64, +} + +/// Inserts an expense and its splits (a tombstone when `deleted_at` is set). `recurring` is the generating template +/// and period (`recurring_template_id`, `recurring_period`). +pub async fn insert_expense( + conn: &mut SqliteConnection, + expense: &PushedExpense, + recurring: Option<(Uuid, &str)>, + origin: Origin<'_>, +) -> Result<(), AppError> { + sqlx::query( + "INSERT INTO expenses (id, account_id, category_id, amount_ore, description, notes, expense_date, \ + payer_user_id, split_mode, recurring_template_id, recurring_period, import_key, created_by_user_id, \ + updated_by_user_id, updated_by_device_id, created_at, updated_at, deleted_at, server_received_at, \ + server_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(expense.id.to_string()) + .bind(expense.account_id.to_string()) + .bind(expense.category_id.to_string()) + .bind(expense.amount_ore) + .bind(&expense.description) + .bind(&expense.notes) + .bind(&expense.expense_date) + .bind(expense.payer_user_id.to_string()) + .bind(token_of(&expense.split_mode)?) + .bind(recurring.map(|(template_id, _)| template_id.to_string())) + .bind(recurring.map(|(_, period)| period)) + .bind(origin.import_key) + .bind(origin.user_id.to_string()) + .bind(origin.user_id.to_string()) + .bind(origin.device_id.map(|id| id.to_string())) + .bind(&expense.created_at) + .bind(&expense.updated_at) + .bind(&expense.deleted_at) + .bind(origin.received_at) + .bind(origin.server_version) + .execute(&mut *conn) + .await?; + insert_expense_splits(conn, expense.id, &expense.splits).await +} + +/// Inserts the `expense_splits` rows of an expense. +pub async fn insert_expense_splits( + conn: &mut SqliteConnection, + expense_id: Uuid, + splits: &[Split], +) -> Result<(), AppError> { + for split in splits { + sqlx::query( + "INSERT INTO expense_splits (expense_id, user_id, share_ore, share_bp) VALUES (?, ?, ?, ?)", + ) + .bind(expense_id.to_string()) + .bind(split.user_id.to_string()) + .bind(split.share_ore) + .bind(split.share_bp) + .execute(&mut *conn) + .await?; + } + Ok(()) +} + +/// Inserts a settlement and its allocations (a tombstone when `deleted_at` is set). +pub async fn insert_settlement( + conn: &mut SqliteConnection, + settlement: &PushedSettlement, + origin: Origin<'_>, +) -> Result<(), AppError> { + sqlx::query( + "INSERT INTO settlements (id, from_user_id, to_user_id, amount_ore, settlement_date, note, import_key, \ + created_by_user_id, updated_by_user_id, updated_by_device_id, created_at, updated_at, deleted_at, \ + server_received_at, server_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(settlement.id.to_string()) + .bind(settlement.from_user_id.to_string()) + .bind(settlement.to_user_id.to_string()) + .bind(settlement.amount_ore) + .bind(&settlement.settlement_date) + .bind(&settlement.note) + .bind(origin.import_key) + .bind(origin.user_id.to_string()) + .bind(origin.user_id.to_string()) + .bind(origin.device_id.map(|id| id.to_string())) + .bind(&settlement.created_at) + .bind(&settlement.updated_at) + .bind(&settlement.deleted_at) + .bind(origin.received_at) + .bind(origin.server_version) + .execute(&mut *conn) + .await?; + insert_allocations(conn, settlement.id, &settlement.allocations).await +} + +/// Inserts the `settlement_allocations` rows of a settlement. +pub async fn insert_allocations( + conn: &mut SqliteConnection, + settlement_id: Uuid, + allocations: &[Allocation], +) -> Result<(), AppError> { + for allocation in allocations { + sqlx::query( + "INSERT INTO settlement_allocations (settlement_id, account_id, amount_ore) VALUES (?, ?, ?)", + ) + .bind(settlement_id.to_string()) + .bind(allocation.account_id.to_string()) + .bind(allocation.amount_ore) + .execute(&mut *conn) + .await?; + } + Ok(()) +} diff --git a/server/crates/kvitt-server/src/sync/mod.rs b/server/crates/kvitt-server/src/sync/mod.rs index 3aa22b3..8820e85 100644 --- a/server/crates/kvitt-server/src/sync/mod.rs +++ b/server/crates/kvitt-server/src/sync/mod.rs @@ -5,7 +5,6 @@ pub mod check; pub mod pull; pub mod push; -pub mod rows; use chrono::{DateTime, Duration, Utc}; diff --git a/server/crates/kvitt-server/src/sync/pull.rs b/server/crates/kvitt-server/src/sync/pull.rs index 5e7b718..9d6e4d2 100644 --- a/server/crates/kvitt-server/src/sync/pull.rs +++ b/server/crates/kvitt-server/src/sync/pull.rs @@ -7,8 +7,6 @@ use crate::error::{AppError, ErrorCode}; use crate::time::format_ts; use crate::wire::SyncPullResponse; -use super::rows as sync_rows; - /// Rows per page when `limit` is absent (SPEC §14.4.9). pub const DEFAULT_LIMIT: i64 = 500; /// Largest `limit` (SPEC §9.6). @@ -47,7 +45,7 @@ pub async fn pull(state: &AppState, since: i64, limit: i64) -> Result = sqlx::query_as(ROW_STREAM) .bind(since) @@ -103,12 +101,12 @@ pub async fn pull(state: &AppState, since: i64, limit: i64) -> Result page.settlements.push( - sync_rows::settlement(&mut tx, id) + rows::settlement(&mut tx, id) .await? .ok_or_else(|| missing("settlements"))?, ), "audit_log" => page.audit_entries.push( - sync_rows::audit_entry(&mut tx, id) + rows::audit_entry(&mut tx, id) .await? .ok_or_else(|| missing("audit_log"))?, ), diff --git a/server/crates/kvitt-server/src/sync/push.rs b/server/crates/kvitt-server/src/sync/push.rs index 0ef0448..8a5c0fd 100644 --- a/server/crates/kvitt-server/src/sync/push.rs +++ b/server/crates/kvitt-server/src/sync/push.rs @@ -12,7 +12,7 @@ use sqlx::{Acquire, SqliteConnection}; use uuid::Uuid; use crate::AppState; -use crate::api::rows; +use crate::api::rows::{self, Origin}; use crate::audit::AuditEvent; use crate::auth::AuthDevice; use crate::db; @@ -26,7 +26,6 @@ use crate::wire::{ }; use super::check::{self, Fail, Rejection}; -use super::rows as sync_rows; /// Mutations per `POST /sync` (SPEC §14.4.9). pub const MAX_MUTATIONS: usize = 100; @@ -241,32 +240,7 @@ impl Kind for ExpenseKind { version: i64, now_ts: &str, ) -> Result<(), AppError> { - sqlx::query( - "INSERT INTO expenses (id, account_id, category_id, amount_ore, description, notes, expense_date, \ - payer_user_id, split_mode, recurring_template_id, recurring_period, import_key, created_by_user_id, \ - updated_by_user_id, updated_by_device_id, created_at, updated_at, deleted_at, server_received_at, \ - server_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL, ?, ?, ?, ?, ?, ?, ?, ?)", - ) - .bind(p.id.to_string()) - .bind(p.account_id.to_string()) - .bind(p.category_id.to_string()) - .bind(p.amount_ore) - .bind(&p.description) - .bind(&p.notes) - .bind(&p.expense_date) - .bind(p.payer_user_id.to_string()) - .bind(rows::token_of(&p.split_mode)?) - .bind(caller.user_id.to_string()) - .bind(caller.user_id.to_string()) - .bind(caller.device_id.to_string()) - .bind(&p.created_at) - .bind(&p.updated_at) - .bind(&p.deleted_at) - .bind(now_ts) - .bind(version) - .execute(&mut *conn) - .await?; - insert_splits(conn, p).await + rows::insert_expense(conn, p, None, origin(caller, version, now_ts)).await } async fn update( @@ -299,23 +273,19 @@ impl Kind for ExpenseKind { .bind(p.id.to_string()) .execute(&mut *conn) .await?; - insert_splits(conn, p).await + rows::insert_expense_splits(conn, p.id, &p.splits).await } } -async fn insert_splits(conn: &mut SqliteConnection, p: &PushedExpense) -> Result<(), AppError> { - for split in &p.splits { - sqlx::query( - "INSERT INTO expense_splits (expense_id, user_id, share_ore, share_bp) VALUES (?, ?, ?, ?)", - ) - .bind(p.id.to_string()) - .bind(split.user_id.to_string()) - .bind(split.share_ore) - .bind(split.share_bp) - .execute(&mut *conn) - .await?; +/// The server-owned columns of a row the caller pushes. +fn origin<'a>(caller: &AuthDevice, version: i64, now_ts: &'a str) -> Origin<'a> { + Origin { + user_id: caller.user_id, + device_id: Some(caller.device_id), + import_key: None, + received_at: now_ts, + server_version: version, } - Ok(()) } impl Kind for SettlementKind { @@ -345,7 +315,7 @@ impl Kind for SettlementKind { } async fn load(conn: &mut SqliteConnection, id: Uuid) -> Result, AppError> { - sync_rows::settlement(conn, id).await + rows::settlement(conn, id).await } async fn check( @@ -364,28 +334,7 @@ impl Kind for SettlementKind { version: i64, now_ts: &str, ) -> Result<(), AppError> { - sqlx::query( - "INSERT INTO settlements (id, from_user_id, to_user_id, amount_ore, settlement_date, note, import_key, \ - created_by_user_id, updated_by_user_id, updated_by_device_id, created_at, updated_at, deleted_at, \ - server_received_at, server_version) VALUES (?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?)", - ) - .bind(p.id.to_string()) - .bind(p.from_user_id.to_string()) - .bind(p.to_user_id.to_string()) - .bind(p.amount_ore) - .bind(&p.settlement_date) - .bind(&p.note) - .bind(caller.user_id.to_string()) - .bind(caller.user_id.to_string()) - .bind(caller.device_id.to_string()) - .bind(&p.created_at) - .bind(&p.updated_at) - .bind(&p.deleted_at) - .bind(now_ts) - .bind(version) - .execute(&mut *conn) - .await?; - insert_allocations(conn, p).await + rows::insert_settlement(conn, p, origin(caller, version, now_ts)).await } async fn update( @@ -415,27 +364,10 @@ impl Kind for SettlementKind { .bind(p.id.to_string()) .execute(&mut *conn) .await?; - insert_allocations(conn, p).await + rows::insert_allocations(conn, p.id, &p.allocations).await } } -async fn insert_allocations( - conn: &mut SqliteConnection, - p: &PushedSettlement, -) -> Result<(), AppError> { - for allocation in &p.allocations { - sqlx::query( - "INSERT INTO settlement_allocations (settlement_id, account_id, amount_ore) VALUES (?, ?, ?)", - ) - .bind(p.id.to_string()) - .bind(allocation.account_id.to_string()) - .bind(allocation.amount_ore) - .execute(&mut *conn) - .await?; - } - Ok(()) -} - /// The outcome of one mutation: its result and the push it causes, if any. struct Processed { result: MutationResult, diff --git a/server/crates/kvitt-server/src/sync/rows.rs b/server/crates/kvitt-server/src/sync/rows.rs deleted file mode 100644 index 535589e..0000000 --- a/server/crates/kvitt-server/src/sync/rows.rs +++ /dev/null @@ -1,160 +0,0 @@ -//! Loading the rows sync needs that `api::rows` does not cover: settlements, audit entries and the epoch. - -use sqlx::{FromRow, SqliteConnection}; -use uuid::Uuid; - -use crate::api::rows::uuid; -use crate::error::AppError; -use crate::wire::{ - Allocation, AuditAction, AuditChange, AuditEntry, AuditSource, EntityType, Settlement, -}; - -#[derive(FromRow)] -struct SettlementRow { - id: String, - from_user_id: String, - to_user_id: String, - amount_ore: i64, - settlement_date: String, - note: String, - created_by_user_id: String, - updated_by_user_id: String, - created_at: String, - updated_at: String, - deleted_at: Option, - server_version: i64, -} - -/// One settlement (tombstones included), with its allocations sorted by `account_id`. -pub async fn settlement( - conn: &mut SqliteConnection, - id: Uuid, -) -> Result, AppError> { - let Some(row) = sqlx::query_as::<_, SettlementRow>( - "SELECT id, from_user_id, to_user_id, amount_ore, settlement_date, note, created_by_user_id, \ - updated_by_user_id, created_at, updated_at, deleted_at, server_version FROM settlements WHERE id = ?", - ) - .bind(id.to_string()) - .fetch_optional(&mut *conn) - .await? - else { - return Ok(None); - }; - let allocations: Vec<(String, i64)> = sqlx::query_as( - "SELECT account_id, amount_ore FROM settlement_allocations WHERE settlement_id = ? ORDER BY account_id", - ) - .bind(id.to_string()) - .fetch_all(&mut *conn) - .await?; - let mut settlement = Settlement { - id: uuid(&row.id)?, - from_user_id: uuid(&row.from_user_id)?, - to_user_id: uuid(&row.to_user_id)?, - amount_ore: row.amount_ore, - settlement_date: row.settlement_date, - note: row.note, - allocations: allocations - .into_iter() - .map(|(account_id, amount_ore)| { - Ok(Allocation { - account_id: uuid(&account_id)?, - amount_ore, - }) - }) - .collect::>()?, - created_by_user_id: uuid(&row.created_by_user_id)?, - updated_by_user_id: uuid(&row.updated_by_user_id)?, - created_at: row.created_at, - updated_at: row.updated_at, - deleted_at: row.deleted_at, - server_version: row.server_version, - }; - settlement.sort(); - Ok(Some(settlement)) -} - -#[derive(FromRow)] -struct AuditRow { - id: String, - entity_type: String, - entity_id: String, - action: String, - source: String, - actor_user_id: String, - client_at: String, - server_at: String, - changes: String, - server_version: i64, -} - -fn entity_type(text: &str) -> Result { - match text { - "expense" => Ok(EntityType::Expense), - "settlement" => Ok(EntityType::Settlement), - other => Err(AppError::internal(format!( - "unknown audit entity_type {other:?}" - ))), - } -} - -fn action(text: &str) -> Result { - match text { - "create" => Ok(AuditAction::Create), - "update" => Ok(AuditAction::Update), - "delete" => Ok(AuditAction::Delete), - other => Err(AppError::internal(format!( - "unknown audit action {other:?}" - ))), - } -} - -fn source(text: &str) -> Result { - match text { - "app" => Ok(AuditSource::App), - "recurring" => Ok(AuditSource::Recurring), - "import" => Ok(AuditSource::Import), - other => Err(AppError::internal(format!( - "unknown audit source {other:?}" - ))), - } -} - -/// One audit entry in its wire form (SPEC §14.3.3; `actor_device_id` is not on the wire). -pub async fn audit_entry( - conn: &mut SqliteConnection, - id: Uuid, -) -> Result, AppError> { - let Some(row) = sqlx::query_as::<_, AuditRow>( - "SELECT id, entity_type, entity_id, action, source, actor_user_id, client_at, server_at, changes, \ - server_version FROM audit_log WHERE id = ?", - ) - .bind(id.to_string()) - .fetch_optional(&mut *conn) - .await? - else { - return Ok(None); - }; - let changes: Vec = serde_json::from_str(&row.changes) - .map_err(|e| AppError::internal(format!("stored audit changes: {e}")))?; - Ok(Some(AuditEntry { - id: uuid(&row.id)?, - entity_type: entity_type(&row.entity_type)?, - entity_id: uuid(&row.entity_id)?, - action: action(&row.action)?, - source: source(&row.source)?, - actor_user_id: uuid(&row.actor_user_id)?, - client_at: row.client_at, - server_at: row.server_at, - changes, - server_version: row.server_version, - })) -} - -/// `meta.sync_epoch`. -pub async fn epoch(conn: &mut SqliteConnection) -> Result { - let value: Option = - sqlx::query_scalar("SELECT value FROM meta WHERE key = 'sync_epoch'") - .fetch_optional(&mut *conn) - .await?; - value.ok_or_else(|| AppError::internal("meta.sync_epoch is missing")) -}