//! 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; the daily scheduler (`tasks::scheduler`) calls [`generate_all_due`]. use std::collections::BTreeMap; use axum::extract::State; use axum::http::StatusCode; use chrono::{DateTime, NaiveDate, NaiveTime, TimeZone, Utc}; use kvitt_domain::{Period, SplitSpec}; use sqlx::SqliteConnection; use uuid::Uuid; use crate::AppState; use crate::api::rows::{self, Origin}; use crate::api::{ApiJson, IdPath, JsonBody, validate}; use crate::audit::AuditEvent; use crate::auth::AuthDevice; use crate::db; use crate::error::AppError; use crate::openapi::responses::{BadRequest, Conflict, NotFound, Unprocessable}; use crate::push::{PushMessage, PushType}; use crate::time::{STOCKHOLM, format_ts}; use crate::wire::{ AuditAction, AuditSource, CreateRecurringTemplateRequest, EntityType, 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). pub const GENERATION_TIME: NaiveTime = match NaiveTime::from_hms_opt(6, 0, 0) { Some(t) => t, None => NaiveTime::MIN, }; // --------------------------------------------------------------------------------------------------------------- // Generation (SPEC §9.5) // --------------------------------------------------------------------------------------------------------------- /// The occurrence instant of `period` for `day_of_month`: the clamped date at 06:00 Europe/Stockholm. pub fn occurrence_instant(period: Period, day_of_month: u8) -> Result, AppError> { let date_text = period .clamped_date(day_of_month) .map_err(AppError::internal)?; let date = NaiveDate::parse_from_str(&date_text, "%Y-%m-%d").map_err(AppError::internal)?; let local = STOCKHOLM .from_local_datetime(&date.and_time(GENERATION_TIME)) .earliest() .ok_or_else(|| AppError::internal("06:00 does not exist in Europe/Stockholm"))?; Ok(local.with_timezone(&Utc)) } /// One expense created by [`generate_due_periods`]. #[derive(Debug, Clone, PartialEq, Eq)] pub struct GeneratedExpense { pub expense_id: Uuid, pub period: String, } /// The expense splits of a generated expense (also E-35's `my_share` forecast of a template). pub(crate) async fn resolve_splits( conn: &mut SqliteConnection, template: &RecurringTemplate, users: &[User], ) -> Result, AppError> { let ids = rows::user_ids(users); let payer = template.payer_user_id.to_string(); let bp: BTreeMap = match template.split_mode { // Mode default resolves the account's current default ratio at generation time. SplitMode::Default => { let account = rows::account(conn, template.account_id) .await? .ok_or_else(|| AppError::internal("template account is missing"))?; account .default_shares .iter() .map(|s| { Ok(( s.user_id.to_string(), u32::try_from(s.share_bp).map_err(AppError::internal)?, )) }) .collect::>()? } SplitMode::Ratio | SplitMode::Full => template .splits .iter() .map(|s| { let bp = s .share_bp .ok_or_else(|| AppError::internal("template split without share_bp"))?; Ok(( s.user_id.to_string(), u32::try_from(bp).map_err(AppError::internal)?, )) }) .collect::>()?, SplitMode::Exact => { let ore: BTreeMap = template .splits .iter() .map(|s| { let ore = s .share_ore .ok_or_else(|| AppError::internal("template split without share_ore"))?; Ok((s.user_id.to_string(), ore)) }) .collect::>()?; let shares = kvitt_domain::compute_split( template.amount_ore, &payer, &ids, &SplitSpec::Exact { ore }, ) .map_err(|e| { AppError::internal(format!("stored exact template split is invalid: {e}")) })?; return shares .into_iter() .map(|(user, share_ore)| { Ok(Split { user_id: rows::uuid(&user)?, share_ore, share_bp: None, }) }) .collect(); } }; // Ratio rule for default, ratio and full (full is ratio with the bearer at 10000 bp, SPEC §14.5.2). let shares = kvitt_domain::compute_split( template.amount_ore, &payer, &ids, &SplitSpec::Ratio { bp: bp.clone() }, ) .map_err(|e| AppError::internal(format!("stored template split is invalid: {e}")))?; shares .into_iter() .map(|(user, share_ore)| { let share_bp = bp.get(&user).map(|v| i64::from(*v)); 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 /// the account's current default ratio; other modes use the template's splits, `ratio`/`full` recomputed by the ratio /// rule), an audit `create` with `source` `recurring` through [`crate::audit::AuditWriter`], and the template's /// `last_generated_period` (new `server_version`). After each commit it enqueues `recurring_expense_created`. /// /// A period whose expense already exists (the unique index on `(recurring_template_id, recurring_period)`, a /// deleted generated expense included) is skipped silently and only `last_generated_period` advances. Deleted or /// paused templates generate nothing. "Now" is `state.clock`. pub async fn generate_due_periods( state: &AppState, template_id: Uuid, ) -> Result, AppError> { let now = state.clock.now(); let mut generated = Vec::new(); loop { let mut tx = db::begin_immediate(&state.db).await?; let Some(template) = rows::template(&mut tx, template_id).await? else { break; }; if template.deleted_at.is_some() || template.paused { break; } let Some(period) = kvitt_domain::next_candidate_period( &template.start_period, template.end_period.as_deref(), template.last_generated_period.as_deref(), ) .map_err(AppError::internal)? else { break; }; let day = u8::try_from(template.day_of_month).map_err(AppError::internal)?; if occurrence_instant(period, day)? > now { break; } let period_text = period.to_string(); let exists: bool = sqlx::query_scalar( "SELECT EXISTS (SELECT 1 FROM expenses WHERE recurring_template_id = ? AND recurring_period = ?)", ) .bind(template_id.to_string()) .bind(&period_text) .fetch_one(&mut *tx) .await?; let now_ts = format_ts(now); 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 = ?", ) .bind(&period_text) .bind(&now_ts) .bind(version) .bind(template_id.to_string()) .execute(&mut *tx) .await?; tx.commit().await?; if let Some(expense_id) = created { state.push.enqueue(PushMessage { push_type: PushType::RecurringExpenseCreated, entity_id: Some(expense_id), actor_user_id: template.created_by_user_id, count: 1, }); generated.push(GeneratedExpense { expense_id, period: period_text, }); } } Ok(generated) } /// [`generate_due_periods`] for every non-deleted, non-paused template, in id order. Returns the number of /// expenses created. pub async fn generate_all_due(state: &AppState) -> Result { let ids: Vec = sqlx::query_scalar( "SELECT id FROM recurring_templates WHERE deleted_at IS NULL AND paused = 0 ORDER BY id", ) .fetch_all(&state.db) .await?; let mut created = 0; for id in ids { created += generate_due_periods(state, rows::uuid(&id)?).await?.len(); } Ok(created) } /// The latest period whose occurrence instant is before `now` (the resume rule of SPEC §9.5). pub fn latest_past_period(day_of_month: u8, now: DateTime) -> Result { let current = validate::current_period(now)?; if occurrence_instant(current, day_of_month)? < now { Ok(current) } else { validate::months_before(current, 1).ok_or_else(|| AppError::internal("no month before")) } } // --------------------------------------------------------------------------------------------------------------- // Validation and storage of template fields // --------------------------------------------------------------------------------------------------------------- /// A stored template split: `(user_id, share_bp, share_ore)`. type StoredSplit = (Uuid, Option, Option); /// Every client-writable field of a template, after merging a PATCH. struct Fields { account_id: Uuid, category_id: Uuid, amount_ore: i64, description: String, notes: String, payer_user_id: Uuid, split_mode: SplitMode, splits: Vec, day_of_month: i64, start_period: String, end_period: Option, paused: bool, } /// The validated template splits as stored: `(user_id, share_bp, share_ore)` (SPEC §9.1, §14.1). fn validate_splits(fields: &Fields, users: &[User]) -> Result, AppError> { let fail = |message: &str| AppError::validation("splits", message); match fields.split_mode { SplitMode::Default => { if fields.splits.is_empty() { Ok(Vec::new()) } else { Err(fail("splits must be empty for mode default")) } } SplitMode::Ratio | SplitMode::Full => { let mut entries = Vec::new(); for split in &fields.splits { match (split.share_bp, split.share_ore) { (Some(bp), None) => entries.push((split.user_id, bp)), _ => return Err(fail("modes ratio and full take share_bp only")), } } rows::bp_shares("splits", &entries, users)?; if fields.split_mode == SplitMode::Full { let mut values: Vec = entries.iter().map(|(_, bp)| *bp).collect(); values.sort_unstable(); if values != [0, 10_000] { return Err(fail( "mode full needs one user at 10000 bp and the other at 0", )); } } Ok(entries .into_iter() .map(|(u, bp)| (u, Some(bp), None)) .collect()) } SplitMode::Exact => { let mut ore = BTreeMap::new(); for split in &fields.splits { match (split.share_bp, split.share_ore) { (None, Some(value)) => { ore.insert(split.user_id.to_string(), value); } _ => return Err(fail("mode exact takes share_ore only")), } } if ore.len() != fields.splits.len() { return Err(fail(kvitt_domain::SplitError::ShareUsersMismatch.token())); } kvitt_domain::compute_split( fields.amount_ore, &fields.payer_user_id.to_string(), &rows::user_ids(users), &SplitSpec::Exact { ore }, ) .map_err(|e| fail(e.token()))?; Ok(fields .splits .iter() .map(|s| (s.user_id, None, s.share_ore)) .collect()) } } } /// Validates `fields` (SPEC §9.6, §9.1, E-28). `start_changed` applies the 12-month backfill limit, which binds only /// a `start_period` the client sends. async fn validate_fields( conn: &mut SqliteConnection, fields: &mut Fields, start_changed: bool, now: DateTime, ) -> Result, AppError> { let users = rows::household(conn).await?; let account = rows::account(conn, fields.account_id).await?; if account.is_none_or(|a| a.deleted_at.is_some()) { return Err(AppError::validation( "account_id", "account_id refers to no account", )); } let category = rows::category(conn, fields.category_id).await?; if category.is_none_or(|c| c.deleted_at.is_some()) { return Err(AppError::validation( "category_id", "category_id refers to no category", )); } validate::amount("amount_ore", fields.amount_ore)?; fields.description = validate::text( "description", &fields.description, 1, validate::DESCRIPTION_MAX, )?; fields.notes = validate::notes("notes", &fields.notes)?; if !users.iter().any(|u| u.id == fields.payer_user_id) { return Err(AppError::validation( "payer_user_id", kvitt_domain::SplitError::PayerUnknown.token(), )); } let splits = validate_splits(fields, &users)?; validate::day_of_month(fields.day_of_month)?; let start = if start_changed { validate::start_period(&fields.start_period, now)? } else { validate::period("start_period", &fields.start_period)? }; if let Some(end) = &fields.end_period && validate::period("end_period", end)? < start { return Err(AppError::validation( "end_period", "end_period must not be before start_period", )); } Ok(splits) } async fn replace_splits( conn: &mut SqliteConnection, template_id: Uuid, splits: &[StoredSplit], ) -> Result<(), AppError> { sqlx::query("DELETE FROM recurring_template_splits WHERE template_id = ?") .bind(template_id.to_string()) .execute(&mut *conn) .await?; for (user_id, share_bp, share_ore) in splits { sqlx::query( "INSERT INTO recurring_template_splits (template_id, user_id, share_bp, share_ore) VALUES (?, ?, ?, ?)", ) .bind(template_id.to_string()) .bind(user_id.to_string()) .bind(share_bp) .bind(share_ore) .execute(&mut *conn) .await?; } Ok(()) } /// Generates what is due, then reads the template back in wire form. async fn generate_and_reload( state: &AppState, template_id: Uuid, ) -> Result { generate_due_periods(state, template_id).await?; let mut conn = state.db.acquire().await?; rows::template(&mut conn, template_id) .await? .ok_or_else(|| AppError::internal("template vanished")) } #[utoipa::path( post, path = "/recurring-templates", operation_id = "createRecurringTemplate", summary = "E-28 Create recurring template", description = "If a period is already due (SPEC §9.5), the server generates it before responding.", request_body(content = CreateRecurringTemplateRequest, content_type = "application/json; charset=utf-8"), responses( (status = 201, description = "The new template.", body = RecurringTemplate, content_type = "application/json; charset=utf-8"), (status = 400, response = BadRequest), (status = 409, response = Conflict), (status = 422, response = Unprocessable), ) )] pub async fn create_recurring_template( device: AuthDevice, State(state): State, ApiJson(body): ApiJson, ) -> Result<(StatusCode, JsonBody), AppError> { let now = state.clock.now(); let mut tx = db::begin_immediate(&state.db).await?; rows::complete_household(&mut tx).await?; let mut fields = Fields { account_id: body.account_id, category_id: body.category_id, amount_ore: body.amount_ore, description: body.description, notes: body.notes, payer_user_id: body.payer_user_id, split_mode: body.split_mode, splits: body.splits, day_of_month: body.day_of_month, start_period: body.start_period, end_period: body.end_period, paused: body.paused, }; let splits = validate_fields(&mut tx, &mut fields, true, now).await?; let id = Uuid::now_v7(); let now_ts = format_ts(now); let version = db::next_server_version(&mut tx).await?; sqlx::query( "INSERT INTO recurring_templates (id, account_id, category_id, amount_ore, description, notes, payer_user_id, \ split_mode, day_of_month, start_period, end_period, paused, last_generated_period, created_by_user_id, \ updated_by_user_id, created_at, updated_at, deleted_at, server_version) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, NULL, ?)", ) .bind(id.to_string()) .bind(fields.account_id.to_string()) .bind(fields.category_id.to_string()) .bind(fields.amount_ore) .bind(&fields.description) .bind(&fields.notes) .bind(fields.payer_user_id.to_string()) .bind(rows::token_of(&fields.split_mode)?) .bind(fields.day_of_month) .bind(&fields.start_period) .bind(&fields.end_period) .bind(i64::from(fields.paused)) .bind(device.user_id.to_string()) .bind(device.user_id.to_string()) .bind(&now_ts) .bind(&now_ts) .bind(version) .execute(&mut *tx) .await?; replace_splits(&mut tx, id, &splits).await?; tx.commit().await?; let template = generate_and_reload(&state, id).await?; Ok((StatusCode::CREATED, JsonBody(template))) } #[utoipa::path( patch, path = "/recurring-templates/{template_id}", operation_id = "patchRecurringTemplate", summary = "E-29 Edit, pause, resume", description = "Any subset of the E-28 fields; `split_mode` and `splits` must be sent together.", params( ("template_id" = UuidSchema, Path, description = "Not a UUID is 404 `not_found`."), ), request_body(content = PatchRecurringTemplateRequest, content_type = "application/json; charset=utf-8"), responses( (status = 200, description = "The updated template.", body = RecurringTemplate, content_type = "application/json; charset=utf-8"), (status = 400, response = BadRequest), (status = 404, response = NotFound), (status = 422, response = Unprocessable), ) )] pub async fn patch_recurring_template( device: AuthDevice, State(state): State, IdPath(template_id): IdPath, ApiJson(body): ApiJson, ) -> Result, AppError> { if body == PatchRecurringTemplateRequest::default() { return Err(AppError::invalid_request("PATCH needs at least one field")); } if body.split_mode.is_some() != body.splits.is_some() { return Err(AppError::invalid_request( "split_mode and splits must be sent together", )); } let now = state.clock.now(); let mut tx = db::begin_immediate(&state.db).await?; let current = rows::live_template(&mut tx, template_id).await?; let start_changed = body.start_period.is_some(); let mut fields = Fields { account_id: body.account_id.unwrap_or(current.account_id), category_id: body.category_id.unwrap_or(current.category_id), amount_ore: body.amount_ore.unwrap_or(current.amount_ore), description: body.description.unwrap_or(current.description), notes: body.notes.unwrap_or(current.notes), payer_user_id: body.payer_user_id.unwrap_or(current.payer_user_id), split_mode: body.split_mode.unwrap_or(current.split_mode), splits: body.splits.unwrap_or(current.splits), day_of_month: body.day_of_month.unwrap_or(current.day_of_month), start_period: body.start_period.unwrap_or(current.start_period), end_period: body.end_period.unwrap_or(current.end_period), paused: body.paused.unwrap_or(current.paused), }; let splits = validate_fields(&mut tx, &mut fields, start_changed, now).await?; // Resume does not backfill (SPEC §9.5): last_generated_period jumps to the latest period whose occurrence is // already past, when that is later than the current value (and not before start_period). let mut last_generated = current.last_generated_period.clone(); if current.paused && !fields.paused { let day = validate::day_of_month(fields.day_of_month)?; let latest = latest_past_period(day, now)?; let start = validate::period("start_period", &fields.start_period)?; let current_last = last_generated .as_deref() .map(|p| validate::period("last_generated_period", p)) .transpose()?; if latest >= start && current_last.is_none_or(|last| latest > last) { last_generated = Some(latest.to_string()); } } let version = db::next_server_version(&mut tx).await?; sqlx::query( "UPDATE recurring_templates SET account_id = ?, category_id = ?, amount_ore = ?, description = ?, notes = ?, \ payer_user_id = ?, split_mode = ?, day_of_month = ?, start_period = ?, end_period = ?, paused = ?, \ last_generated_period = ?, updated_by_user_id = ?, updated_at = ?, server_version = ? WHERE id = ?", ) .bind(fields.account_id.to_string()) .bind(fields.category_id.to_string()) .bind(fields.amount_ore) .bind(&fields.description) .bind(&fields.notes) .bind(fields.payer_user_id.to_string()) .bind(rows::token_of(&fields.split_mode)?) .bind(fields.day_of_month) .bind(&fields.start_period) .bind(&fields.end_period) .bind(i64::from(fields.paused)) .bind(&last_generated) .bind(device.user_id.to_string()) .bind(format_ts(now)) .bind(version) .bind(template_id.to_string()) .execute(&mut *tx) .await?; replace_splits(&mut tx, template_id, &splits).await?; tx.commit().await?; Ok(JsonBody(generate_and_reload(&state, template_id).await?)) } #[utoipa::path( delete, path = "/recurring-templates/{template_id}", operation_id = "deleteRecurringTemplate", summary = "E-30 Soft delete recurring template", params( ("template_id" = UuidSchema, Path, description = "Not a UUID is 404 `not_found`."), ), responses( (status = 200, description = "The tombstone.", body = RecurringTemplate, content_type = "application/json; charset=utf-8"), (status = 404, response = NotFound), ) )] pub async fn delete_recurring_template( device: AuthDevice, State(state): State, IdPath(template_id): IdPath, ) -> Result, AppError> { let mut tx = db::begin_immediate(&state.db).await?; rows::live_template(&mut tx, template_id).await?; let now = state.clock.now_ts(); let version = db::next_server_version(&mut tx).await?; sqlx::query( "UPDATE recurring_templates SET deleted_at = ?, updated_at = ?, updated_by_user_id = ?, server_version = ? \ WHERE id = ?", ) .bind(&now) .bind(&now) .bind(device.user_id.to_string()) .bind(version) .bind(template_id.to_string()) .execute(&mut *tx) .await?; let template = rows::template(&mut tx, template_id) .await? .ok_or_else(|| AppError::internal("template vanished"))?; tx.commit().await?; Ok(JsonBody(template)) }