jacobh2/kvitt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Server overhaul: one place for stored-row reads and the shared inserts

Browse Source 
api::rows now also loads settlements, audit entries and the epoch (sync::rows is gone; audit enums parse through
rows::token) and owns the expense, split, settlement and allocation inserts that sync push, recurring generation and
the importer each had their own copy of (rows::Origin carries the server-owned columns). accounts::insert_account and
replace_shares are shared with the importer. Recurring generation moves the per-period expense into
create_period_expense. No SQL, wire or log change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TQd3pEP9UeN2uifbHknQ7B
This commit is contained in:
Jacob Hagstedt
2026-10-18 22:28:51 +02:00
parent a0259ab68d
commit 8103dae2be
8 changed files with 426 additions and 453 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/crates/kvitt-import/src/db.rs
+28 -113
View File
@@ -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<Origin<'a>, 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)?;
server/crates/kvitt-server/src/api/accounts.rs
+29 -14
View File
@@ -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<Account, AppError> {
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?;
server/crates/kvitt-server/src/api/recurring.rs
+92 -74
View File
@@ -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<Vec<(Uuid, i64, Option<i64>)>, AppError> {
) -> Result<Vec<Split>, AppError> {
let ids = rows::user_ids(users);
let payer = template.payer_user_id.to_string();
let bp: BTreeMap<String, u32> = 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<Uuid, AppError> {
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 = ?",
server/crates/kvitt-server/src/api/rows.rs
+260 -4
View File
@@ -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<String>) -> Result<Option<Uuid>, AppError> {
}
/// Parses a stored snake_case enum token through its serde form.
fn token<T: DeserializeOwned>(text: &str) -> Result<T, AppError> {
pub fn token<T: DeserializeOwned>(text: &str) -> Result<T, AppError> {
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<String>,
server_version: i64,
}
/// One settlement (tombstones included), with its allocations sorted by `account_id`.
pub async fn settlement(
conn: &mut SqliteConnection,
id: Uuid,
) -> Result<Option<Settlement>, 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::<Result<_, AppError>>()?,
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<Option<AuditEntry>, 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<AuditChange> = 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<String, AppError> {
let value: Option<String> =
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<Uuid>,
/// `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(())
}
server/crates/kvitt-server/src/sync/mod.rs
-1
View File
@@ -5,7 +5,6 @@
pub mod check;
pub mod pull;
pub mod push;
pub mod rows;
use chrono::{DateTime, Duration, Utc};
server/crates/kvitt-server/src/sync/pull.rs
+3 -5
View File
@@ -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<SyncPullRe
)
.with_field("since"));
}
let epoch = sync_rows::epoch(&mut tx).await?;
let epoch = rows::epoch(&mut tx).await?;
// One row more than the page tells whether more exist.
let mut stream: Vec<(i64, String, String)> = sqlx::query_as(ROW_STREAM)
.bind(since)
@@ -103,12 +101,12 @@ pub async fn pull(state: &AppState, since: i64, limit: i64) -> Result<SyncPullRe
.ok_or_else(|| missing("expenses"))?,
),
"settlements" => 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"))?,
),
server/crates/kvitt-server/src/sync/push.rs
+14 -82
View File
@@ -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<Option<Settlement>, 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,
server/crates/kvitt-server/src/sync/rows.rs
-160
View File
@@ -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<String>,
server_version: i64,
}
/// One settlement (tombstones included), with its allocations sorted by `account_id`.
pub async fn settlement(
conn: &mut SqliteConnection,
id: Uuid,
) -> Result<Option<Settlement>, 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::<Result<_, AppError>>()?,
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<EntityType, AppError> {
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<AuditAction, AppError> {
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<AuditSource, AppError> {
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<Option<AuditEntry>, 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<AuditChange> = 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<String, AppError> {
let value: Option<String> =
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"))
}