kvitt/server/crates/kvitt-import/src/db.rs

//! Writing a plan into the database (SPEC §10.8): one `BEGIN IMMEDIATE` transaction, the account found or created,
//! rows whose `import_key` exists skipped, every other row validated with the server's own push checks
//! (`kvitt_server::sync::check`), inserted with a new `server_version`, and audited through [`AuditWriter`] with
//! `source` `import`. Nothing goes through a push queue: the importer has none.

use std::collections::BTreeMap;

use chrono::{DateTime, Utc};
use kvitt_domain::{BalanceExpense, BalanceSettlement, compute_balances};
use kvitt_server::api::{accounts::name_key, rows};
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,
    SplitMode, User,
};
use sqlx::SqliteConnection;
use uuid::Uuid;

use crate::parse::RowProblem;
use crate::plan::{PlannedBody, PlannedExpense, PlannedRow, PlannedSettlement, PlannedSplit};

/// Default ratio of a created account: 50/50 (SPEC §10.8).
pub const DEFAULT_SHARE_BP: i64 = 5000;

/// What writing produced (before the caller commits or rolls back).
#[derive(Debug, Default)]
pub struct Written {
    pub account_id: String,
    pub account_created: bool,
    pub expenses_created: usize,
    pub settlements_created: usize,
    pub skipped_duplicates: usize,
    /// Rows the server's checks rejected; non-empty means the caller must roll back.
    pub rejected: Vec<RowProblem>,
}

/// The household context of a run.
pub struct Household<'a> {
    pub users: &'a [User],
    pub owner: Uuid,
}

fn uuid(text: &str) -> Result<Uuid, AppError> {
    rows::uuid(text)
}

/// Finds the non-deleted account with this name (case-insensitive, SPEC §14.1 `name_key`), else creates it with
/// the default 50/50 ratio, created by the owner.
async fn find_or_create_account(
    conn: &mut SqliteConnection,
    name: &str,
    household: &Household<'_>,
    now_ts: &str,
) -> Result<(String, bool), AppError> {
    let key = name_key(name);
    let existing: Option<String> =
        sqlx::query_scalar("SELECT id FROM accounts WHERE name_key = ? AND deleted_at IS NULL")
            .bind(&key)
            .fetch_optional(&mut *conn)
            .await?;
    if let Some(id) = existing {
        return Ok((id, false));
    }
    let id = Uuid::now_v7().to_string();
    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))
}

/// The id of each non-deleted default category, by key.
async fn default_categories(
    conn: &mut SqliteConnection,
) -> Result<BTreeMap<String, String>, AppError> {
    let pairs: Vec<(String, String)> = sqlx::query_as(
        "SELECT default_key, id FROM categories WHERE default_key IS NOT NULL AND deleted_at IS NULL",
    )
    .fetch_all(&mut *conn)
    .await?;
    Ok(pairs.into_iter().collect())
}

async fn key_exists(conn: &mut SqliteConnection, key: &str) -> Result<bool, AppError> {
    let exists: bool = sqlx::query_scalar(
        "SELECT EXISTS (SELECT 1 FROM expenses WHERE import_key = ?1) \
         OR EXISTS (SELECT 1 FROM settlements WHERE import_key = ?1)",
    )
    .bind(key)
    .fetch_one(&mut *conn)
    .await?;
    Ok(exists)
}

fn pushed_expense(
    e: &PlannedExpense,
    users: &[User],
    account_id: Uuid,
    category_id: Uuid,
    now_ts: &str,
) -> Result<PushedExpense, AppError> {
    let (split_mode, splits) = match &e.split {
        PlannedSplit::Exact(shares) => (
            SplitMode::Exact,
            shares
                .iter()
                .map(|(user, share)| {
                    Ok(Split {
                        user_id: uuid(user)?,
                        share_ore: *share,
                        share_bp: None,
                    })
                })
                .collect::<Result<Vec<_>, AppError>>()?,
        ),
        PlannedSplit::Full(bearer) => (SplitMode::Full, full_splits(e.amount_ore, bearer, users)?),
    };
    Ok(PushedExpense {
        id: Uuid::now_v7(),
        account_id,
        category_id,
        amount_ore: e.amount_ore,
        description: e.description.clone(),
        notes: e.notes.clone(),
        expense_date: e.date.clone(),
        payer_user_id: uuid(&e.payer_user_id)?,
        split_mode,
        splits,
        created_at: now_ts.to_owned(),
        updated_at: now_ts.to_owned(),
        deleted_at: None,
    })
}

/// The splits of mode `full` (SPEC §9.1: bearer 10000 bp and the whole amount, the other user 0).
fn full_splits(amount_ore: i64, bearer: &str, users: &[User]) -> Result<Vec<Split>, AppError> {
    users
        .iter()
        .map(|u| {
            let bears = u.id.to_string() == bearer;
            Ok(Split {
                user_id: u.id,
                share_ore: if bears { amount_ore } else { 0 },
                share_bp: Some(if bears { 10_000 } else { 0 }),
            })
        })
        .collect()
}

fn pushed_settlement(
    s: &PlannedSettlement,
    account_id: Uuid,
    now_ts: &str,
) -> Result<PushedSettlement, AppError> {
    Ok(PushedSettlement {
        id: Uuid::now_v7(),
        from_user_id: uuid(&s.from_user_id)?,
        to_user_id: uuid(&s.to_user_id)?,
        amount_ore: s.amount_ore,
        settlement_date: s.date.clone(),
        note: s.note.clone(),
        allocations: vec![Allocation {
            account_id,
            amount_ore: s.amount_ore,
        }],
        created_at: now_ts.to_owned(),
        updated_at: now_ts.to_owned(),
        deleted_at: None,
    })
}

async fn insert_expense(
    conn: &mut SqliteConnection,
    p: &PushedExpense,
    import_key: &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(())
}

async fn audit_create(
    conn: &mut SqliteConnection,
    audit: &dyn AuditWriter,
    entity_type: EntityType,
    entity_id: Uuid,
    after: serde_json::Value,
    owner: Uuid,
    now_ts: &str,
) -> Result<(), AppError> {
    audit
        .write(
            conn,
            AuditEvent {
                entity_type,
                entity_id,
                action: AuditAction::Create,
                source: AuditSource::Import,
                actor_user_id: owner,
                actor_device_id: None,
                client_at: now_ts.to_owned(),
                server_at: now_ts.to_owned(),
                before: None,
                after,
            },
        )
        .await?;
    Ok(())
}

/// Turns a server check failure into a row problem, or passes a real error through.
fn rejected(line: u64, fail: Fail) -> Result<RowProblem, AppError> {
    match fail {
        Fail::Reject(r) => Ok(RowProblem {
            line,
            reason: format!("{}: {}", rows::token_of(&r.reason)?, r.message),
        }),
        Fail::Internal(e) => Err(e),
    }
}

/// Writes the planned rows on `conn` (inside the caller's `BEGIN IMMEDIATE` transaction).
pub async fn write(
    conn: &mut SqliteConnection,
    account_name: &str,
    planned: &[PlannedRow],
    household: &Household<'_>,
    audit: &dyn AuditWriter,
    now: DateTime<Utc>,
) -> Result<Written, AppError> {
    let now_ts = format_ts(now);
    let (account_id, account_created) =
        find_or_create_account(conn, account_name, household, &now_ts).await?;
    let account_uuid = uuid(&account_id)?;
    let categories = default_categories(conn).await?;
    let owner = household.owner;
    let mut written = Written {
        account_id,
        account_created,
        ..Written::default()
    };
    for row in planned {
        if key_exists(conn, &row.import_key).await? {
            written.skipped_duplicates += 1;
            continue;
        }
        match &row.body {
            PlannedBody::Expense(e) => {
                let category_id = categories.get(&e.category_key).ok_or_else(|| {
                    AppError::internal(format!("default category {} is missing", e.category_key))
                })?;
                let pushed = pushed_expense(
                    e,
                    household.users,
                    account_uuid,
                    uuid(category_id)?,
                    &now_ts,
                )?;
                let checked = match check::expense(conn, &pushed, household.users, now).await {
                    Ok(checked) => checked,
                    Err(fail) => {
                        written.rejected.push(rejected(row.line, fail)?);
                        continue;
                    }
                };
                insert_expense(conn, &checked, &row.import_key, owner, &now_ts).await?;
                let stored = rows::expense(conn, checked.id)
                    .await?
                    .ok_or_else(|| AppError::internal("imported expense vanished"))?;
                let after = serde_json::to_value(&stored).map_err(AppError::internal)?;
                audit_create(
                    conn,
                    audit,
                    EntityType::Expense,
                    checked.id,
                    after,
                    owner,
                    &now_ts,
                )
                .await?;
                written.expenses_created += 1;
            }
            PlannedBody::Settlement(s) => {
                let pushed = pushed_settlement(s, account_uuid, &now_ts)?;
                let checked = match check::settlement(conn, &pushed, household.users, now).await {
                    Ok(checked) => checked,
                    Err(fail) => {
                        written.rejected.push(rejected(row.line, fail)?);
                        continue;
                    }
                };
                insert_settlement(conn, &checked, &row.import_key, owner, &now_ts).await?;
                let stored = sync_rows::settlement(conn, checked.id)
                    .await?
                    .ok_or_else(|| AppError::internal("imported settlement vanished"))?;
                let after = serde_json::to_value(&stored).map_err(AppError::internal)?;
                audit_create(
                    conn,
                    audit,
                    EntityType::Settlement,
                    checked.id,
                    after,
                    owner,
                    &now_ts,
                )
                .await?;
                written.settlements_created += 1;
            }
        }
    }
    Ok(written)
}

/// The account's balance per user over its non-deleted expenses and settlement allocations (SPEC §9.3), computed
/// with `kvitt_domain::compute_balances`.
pub async fn account_balance(
    conn: &mut SqliteConnection,
    account_id: &str,
    user_ids: &[String],
) -> Result<BTreeMap<String, i64>, AppError> {
    let heads: Vec<(String, i64, String)> = sqlx::query_as(
        "SELECT id, amount_ore, payer_user_id FROM expenses WHERE account_id = ? AND deleted_at IS NULL",
    )
    .bind(account_id)
    .fetch_all(&mut *conn)
    .await?;
    let splits: Vec<(String, String, i64)> = sqlx::query_as(
        "SELECT s.expense_id, s.user_id, s.share_ore FROM expense_splits s JOIN expenses e ON e.id = s.expense_id \
         WHERE e.account_id = ? AND e.deleted_at IS NULL",
    )
    .bind(account_id)
    .fetch_all(&mut *conn)
    .await?;
    let mut shares: BTreeMap<String, BTreeMap<String, i64>> = BTreeMap::new();
    for (expense_id, user_id, share) in splits {
        shares.entry(expense_id).or_default().insert(user_id, share);
    }
    let expenses: Vec<BalanceExpense> = heads
        .into_iter()
        .map(|(id, amount_ore, payer_user_id)| BalanceExpense {
            account_id: account_id.to_owned(),
            amount_ore,
            payer_user_id,
            shares: shares.remove(&id).unwrap_or_default(),
        })
        .collect();
    let allocations: Vec<(String, String, i64)> = sqlx::query_as(
        "SELECT s.from_user_id, s.to_user_id, a.amount_ore FROM settlement_allocations a \
         JOIN settlements s ON s.id = a.settlement_id WHERE a.account_id = ? AND s.deleted_at IS NULL",
    )
    .bind(account_id)
    .fetch_all(&mut *conn)
    .await?;
    let settlements: Vec<BalanceSettlement> = allocations
        .into_iter()
        .map(|(from_user_id, to_user_id, amount)| BalanceSettlement {
            from_user_id,
            to_user_id,
            allocations: BTreeMap::from([(account_id.to_owned(), amount)]),
        })
        .collect();
    let balances = compute_balances(user_ids, &[account_id.to_owned()], &expenses, &settlements);
    Ok(balances
        .per_account
        .get(account_id)
        .cloned()
        .unwrap_or_default())
}