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

//! `kvitt-import`: imports Splitwise CSV exports, one file per group, into a Kvitt account (SPEC §10.8, FR-17).
//!
//! The run: read and check the file ([`parse`]), resolve the `--user` arguments against the household, build the
//! rows ([`plan`]), then in one `BEGIN IMMEDIATE` transaction find or create the account, skip rows whose
//! `import_key` exists, validate the rest with the server's own push checks, insert and audit them ([`db`]), and
//! reconcile the account balance with the file's totals row. The transaction commits only when every row is valid,
//! the balance matches and this is not a dry run; otherwise it rolls back and nothing is written.

// No unwrap/expect outside tests (CLAUDE.md server rule); integration tests are separate crates and may use them.
#![cfg_attr(not(test), deny(clippy::unwrap_used, clippy::expect_used))]

pub mod amount;
pub mod categories;
pub mod cli;
pub mod db;
pub mod parse;
pub mod plan;

use std::collections::{BTreeMap, BTreeSet};
use std::fmt::Write as _;
use std::path::PathBuf;

use chrono::{DateTime, Utc};
use kvitt_server::api::{rows, validate};
use kvitt_server::audit::AuditWriter;
use kvitt_server::error::AppError;
use kvitt_server::wire::{Role, User};
use sqlx::SqliteConnection;
use uuid::Uuid;

use crate::amount::{Rate, format_ore};
use crate::categories::CategoryTable;
use crate::parse::{FileError, Language, RowProblem};

/// Exit status: success.
pub const EXIT_OK: u8 = 0;
/// Exit status: database or file system error.
pub const EXIT_ERROR: u8 = 1;
/// Exit status: bad command line (clap's own usage errors use 2 as well).
pub const EXIT_USAGE: u8 = 2;
/// Exit status: the file cannot be imported (header, invalid rows, non-SEK rows without `--fx`); nothing written.
pub const EXIT_INVALID: u8 = 3;
/// Exit status: the balance does not reconcile with the file's totals row; nothing written.
pub const EXIT_MISMATCH: u8 = 4;

/// One `--user` argument: the Splitwise member name and the Kvitt user (an id, or a display name to resolve).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UserArg {
    pub splitwise_name: String,
    pub kvitt_user: String,
}

/// A whole run's inputs.
#[derive(Debug, Clone)]
pub struct Options {
    pub db: PathBuf,
    pub file: PathBuf,
    pub account: String,
    pub users: Vec<UserArg>,
    pub fx: BTreeMap<String, Rate>,
    pub dry_run: bool,
}

/// Why a run failed.
#[derive(Debug)]
pub enum ImportError {
    /// Bad arguments or unresolvable users.
    Usage(String),
    /// The header is not one of the two variants or names other members.
    Header(String),
    /// Rows that cannot be imported, with their line numbers.
    InvalidRows(Vec<RowProblem>),
    /// Non-SEK rows without a `--fx` rate: currency → rows.
    MissingFx(BTreeMap<String, usize>),
    /// Database or I/O.
    Io(String),
}

impl ImportError {
    pub fn exit_code(&self) -> u8 {
        match self {
            Self::Usage(_) => EXIT_USAGE,
            Self::Header(_) | Self::InvalidRows(_) | Self::MissingFx(_) => EXIT_INVALID,
            Self::Io(_) => EXIT_ERROR,
        }
    }
}

impl std::fmt::Display for ImportError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Usage(m) => write!(f, "usage: {m}"),
            Self::Header(m) => write!(f, "header: {m}; nothing written"),
            Self::InvalidRows(problems) => {
                writeln!(f, "{} invalid rows; nothing written:", problems.len())?;
                for p in problems {
                    writeln!(f, "  line {}: {}", p.line, p.reason)?;
                }
                Ok(())
            }
            Self::MissingFx(currencies) => {
                write!(f, "rows not in SEK and no --fx rate; nothing written:")?;
                for (currency, count) in currencies {
                    write!(f, " {currency} ({count} rows)")?;
                }
                Ok(())
            }
            Self::Io(m) => write!(f, "error: {m}"),
        }
    }
}

impl From<AppError> for ImportError {
    fn from(e: AppError) -> Self {
        Self::Io(format!("database: {} ({})", e.message, e.code))
    }
}

impl From<sqlx::Error> for ImportError {
    fn from(e: sqlx::Error) -> Self {
        Self::Io(format!("database: {e}"))
    }
}

/// One member's line of the reconciliation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberBalance {
    pub splitwise_name: String,
    pub user_id: String,
    /// The Splitwise totals row value (converted to SEK öre).
    pub splitwise_ore: i64,
    /// `net(u)` of the Kvitt account after the import (SPEC §9.3).
    pub kvitt_ore: i64,
}

/// The reconciliation of the account balance with the file's totals row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Reconciliation {
    /// Per member, in header order; empty without a totals row.
    pub members: Vec<MemberBalance>,
    /// Whether the file had a totals row.
    pub has_totals: bool,
    pub matched: bool,
    /// Whether `--fx` conversion took place (rounding per value can then make an exact match impossible, so a
    /// mismatch is reported but does not fail the run).
    pub converted: bool,
}

impl Reconciliation {
    /// Whether the run may commit: an exact match, or a mismatch caused only by `--fx` rounding.
    pub fn accepted(&self) -> bool {
        self.has_totals && (self.matched || self.converted)
    }
}

/// What a run did (or, in a dry run, would do).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Report {
    pub dry_run: bool,
    pub committed: bool,
    pub language: Language,
    pub account_name: String,
    pub account_id: String,
    pub account_created: bool,
    /// Data rows (expenses and payments); blank and totals rows excluded.
    pub rows_read: usize,
    pub blank_rows: usize,
    pub totals_rows: usize,
    pub expenses_created: usize,
    pub settlements_created: usize,
    pub skipped_duplicates: usize,
    /// Unmapped Splitwise category → rows (imported as `other`).
    pub unmapped_categories: BTreeMap<String, usize>,
    pub non_sek_rows: usize,
    pub reconciliation: Reconciliation,
}

impl Report {
    pub fn unmapped_rows(&self) -> usize {
        self.unmapped_categories.values().sum()
    }

    /// The process exit status for this report.
    pub fn exit_code(&self) -> u8 {
        if self.reconciliation.accepted() {
            EXIT_OK
        } else {
            EXIT_MISMATCH
        }
    }
}

impl std::fmt::Display for Report {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut out = String::new();
        let verb = if self.dry_run {
            "would create"
        } else {
            "created"
        };
        let _ = writeln!(
            out,
            "kvitt-import summary{}",
            if self.dry_run {
                " (dry run, nothing written)"
            } else {
                ""
            }
        );
        let _ = writeln!(out, "  header variant:      {}", self.language.as_str());
        let _ = writeln!(
            out,
            "  account:             {} ({}{})",
            self.account_name,
            self.account_id,
            if self.account_created {
                ", new"
            } else {
                ", existing"
            }
        );
        let _ = writeln!(out, "  rows read:           {}", self.rows_read);
        let _ = writeln!(out, "  totals rows:         {}", self.totals_rows);
        let _ = writeln!(out, "  expenses {verb}: {}", self.expenses_created);
        let _ = writeln!(out, "  settlements {verb}: {}", self.settlements_created);
        let _ = writeln!(out, "  skipped duplicates:  {}", self.skipped_duplicates);
        let _ = writeln!(out, "  non-SEK rows:        {}", self.non_sek_rows);
        let _ = writeln!(out, "  unknown categories (imported as other):");
        if self.unmapped_categories.is_empty() {
            let _ = writeln!(out, "    none");
        }
        for (name, count) in &self.unmapped_categories {
            let _ = writeln!(out, "    {name}: {count} rows");
        }
        if self.unmapped_rows() > 0 {
            let _ = writeln!(
                out,
                "  WARNING: {} rows with an unmapped category",
                self.unmapped_rows()
            );
        }
        let r = &self.reconciliation;
        let _ = writeln!(
            out,
            "  reconciliation (account balance vs the Splitwise totals row, SEK):"
        );
        if !r.has_totals {
            let _ = writeln!(out, "    MISMATCH: the file has no totals row");
        }
        for m in &r.members {
            let _ = writeln!(
                out,
                "    {}: splitwise {}, kvitt {}",
                m.splitwise_name,
                format_ore(m.splitwise_ore),
                format_ore(m.kvitt_ore)
            );
        }
        if r.has_totals {
            let verdict = match (r.matched, r.converted) {
                (true, _) => "match",
                (false, true) => "MISMATCH after --fx rounding (accepted)",
                (false, false) => "MISMATCH",
            };
            let _ = writeln!(out, "    result: {verdict}");
        }
        let status = if self.committed {
            "committed"
        } else if self.dry_run {
            "rolled back (dry run)"
        } else {
            "rolled back, nothing written"
        };
        let _ = write!(out, "  transaction:         {status}");
        f.write_str(&out)
    }
}

/// Parses `NAME=VALUE` (the value after the last `=`, so names may not contain `=` but nothing else is special).
pub fn split_assignment(text: &str) -> Result<(String, String), String> {
    let (name, value) = text
        .rsplit_once('=')
        .ok_or_else(|| format!("{text:?} is not NAME=VALUE"))?;
    let (name, value) = (name.trim(), value.trim());
    if name.is_empty() || value.is_empty() {
        return Err(format!("{text:?} is not NAME=VALUE"));
    }
    Ok((name.to_owned(), value.to_owned()))
}

/// Opens an existing database (never creates one) and checks its schema version.
pub async fn open(path: &std::path::Path) -> Result<kvitt_server::db::Db, ImportError> {
    if !path.is_file() {
        return Err(ImportError::Io(format!(
            "no database at {}",
            path.display()
        )));
    }
    let options = kvitt_server::db::pool::options(path).create_if_missing(false);
    let pool = sqlx::sqlite::SqlitePoolOptions::new()
        .max_connections(1)
        .connect_with(options)
        .await?;
    let version: Option<String> =
        sqlx::query_scalar("SELECT value FROM meta WHERE key = 'schema_version'")
            .fetch_optional(&pool)
            .await
            .map_err(|e| ImportError::Io(format!("not a Kvitt database: {e}")))?;
    let expected = kvitt_server::db::SCHEMA_VERSION.to_string();
    if version.as_deref() != Some(expected.as_str()) {
        return Err(ImportError::Io(format!(
            "schema version {version:?}, expected {expected}; run `kvitt-server migrate` first"
        )));
    }
    Ok(pool)
}

/// Resolves one `--user` value: a user id, or else a display name that exactly one user has.
fn resolve_user(value: &str, users: &[User]) -> Result<Uuid, ImportError> {
    if let Ok(id) = Uuid::parse_str(value) {
        return users
            .iter()
            .find(|u| u.id == id)
            .map(|u| u.id)
            .ok_or_else(|| ImportError::Usage(format!("no user with id {id}")));
    }
    let matches: Vec<&User> = users.iter().filter(|u| u.display_name == value).collect();
    match matches[..] {
        [user] => Ok(user.id),
        [] => Err(ImportError::Usage(format!(
            "no user with id or display name {value:?}"
        ))),
        _ => Err(ImportError::Usage(format!(
            "display name {value:?} is ambiguous; give the user id"
        ))),
    }
}

/// The household, the owner and the user id of each member column (header order).
async fn resolve_household(
    conn: &mut SqliteConnection,
    options: &Options,
    members: &[String],
) -> Result<(Vec<User>, Uuid, Vec<String>), ImportError> {
    let users = rows::complete_household(conn)
        .await
        .map_err(|e| ImportError::Usage(format!("household: {}", e.message)))?;
    let owner = users
        .iter()
        .find(|u| u.role == Role::Admin)
        .map(|u| u.id)
        .ok_or_else(|| ImportError::Usage("the household has no admin (owner)".into()))?;
    let mut by_name: BTreeMap<&str, Uuid> = BTreeMap::new();
    for arg in &options.users {
        let id = resolve_user(&arg.kvitt_user, &users)?;
        by_name.insert(arg.splitwise_name.as_str(), id);
    }
    let ids: BTreeSet<Uuid> = by_name.values().copied().collect();
    let household: BTreeSet<Uuid> = users.iter().map(|u| u.id).collect();
    if ids != household {
        return Err(ImportError::Usage(
            "the two --user arguments must name the two household users, one each".into(),
        ));
    }
    let member_ids = members
        .iter()
        .map(|m| {
            by_name
                .get(m.as_str())
                .map(Uuid::to_string)
                .ok_or_else(|| ImportError::Header(format!("member column {m:?} has no --user")))
        })
        .collect::<Result<Vec<_>, _>>()?;
    Ok((users, owner, member_ids))
}

/// Checks the command line against the file header: two `--user` arguments with distinct names, the same set of
/// names as the header's member columns.
fn check_users(options: &Options, members: &[String]) -> Result<(), ImportError> {
    let names: BTreeSet<&str> = options
        .users
        .iter()
        .map(|u| u.splitwise_name.as_str())
        .collect();
    if options.users.len() != 2 || names.len() != 2 {
        return Err(ImportError::Usage(
            "give exactly two --user arguments with distinct Splitwise names".into(),
        ));
    }
    let header: BTreeSet<&str> = members.iter().map(String::as_str).collect();
    if names != header {
        return Err(ImportError::Header(format!(
            "the member columns {members:?} do not match the --user names {:?}",
            names.into_iter().collect::<Vec<_>>()
        )));
    }
    Ok(())
}

/// Runs one import. `now` is the import time (`created_at`, `updated_at`, audit times; dates are checked against
/// it). Returns the report whether or not it committed (see [`Report::committed`]); errors write nothing.
pub async fn run(
    options: &Options,
    now: DateTime<Utc>,
    audit: &dyn AuditWriter,
) -> Result<Report, ImportError> {
    let account_name = validate::text("account", &options.account, 1, validate::ACCOUNT_NAME_MAX)
        .map_err(|e| ImportError::Usage(e.message))?;
    let table = CategoryTable::committed().map_err(ImportError::Io)?;
    let bytes = std::fs::read(&options.file)
        .map_err(|e| ImportError::Io(format!("{}: {e}", options.file.display())))?;
    let file = parse::parse(&bytes).map_err(|e| match e {
        FileError::Header(m) => ImportError::Header(m),
        FileError::Csv(m) => ImportError::InvalidRows(vec![RowProblem { line: 0, reason: m }]),
    })?;
    check_users(options, &file.header.members)?;

    let pool = open(&options.db).await?;
    let mut conn = pool.acquire().await?;
    let (users, owner, member_ids) =
        resolve_household(&mut conn, options, &file.header.members).await?;
    // An existing account keeps its stored name, which is also the name in the import key, so a re-run with the
    // name in another case still finds its rows.
    let existing: Option<String> =
        sqlx::query_scalar("SELECT name FROM accounts WHERE name_key = ? AND deleted_at IS NULL")
            .bind(kvitt_server::api::accounts::name_key(&account_name))
            .fetch_optional(&mut *conn)
            .await?;
    let account_name = existing.unwrap_or(account_name);
    drop(conn);

    let owner_text = owner.to_string();
    let plan = plan::build(
        &file,
        &plan::PlanContext {
            account_name: &account_name,
            member_user_ids: &member_ids,
            owner_user_id: &owner_text,
            fx: &options.fx,
            categories: &table,
        },
    );
    if !plan.missing_fx.is_empty() {
        return Err(ImportError::MissingFx(plan.missing_fx));
    }
    if !plan.problems.is_empty() {
        return Err(ImportError::InvalidRows(plan.problems));
    }

    let mut tx = kvitt_server::db::begin_immediate(&pool).await?;
    let household = db::Household {
        users: &users,
        owner,
    };
    let written = db::write(&mut tx, &account_name, &plan.rows, &household, audit, now).await?;
    if !written.rejected.is_empty() {
        tx.rollback().await?;
        return Err(ImportError::InvalidRows(written.rejected));
    }
    let balance =
        db::account_balance(&mut tx, &written.account_id, &rows::user_ids(&users)).await?;
    let reconciliation = match &plan.expected_totals {
        Some(expected) => {
            let members: Vec<MemberBalance> = file
                .header
                .members
                .iter()
                .zip(&member_ids)
                .map(|(name, id)| MemberBalance {
                    splitwise_name: name.clone(),
                    user_id: id.clone(),
                    splitwise_ore: expected.get(id).copied().unwrap_or(0),
                    kvitt_ore: balance.get(id).copied().unwrap_or(0),
                })
                .collect();
            let matched = members.iter().all(|m| m.splitwise_ore == m.kvitt_ore);
            Reconciliation {
                members,
                has_totals: true,
                matched,
                converted: plan.converted,
            }
        }
        None => Reconciliation {
            members: Vec::new(),
            has_totals: false,
            matched: false,
            converted: plan.converted,
        },
    };
    let commit = !options.dry_run && reconciliation.accepted();
    if commit {
        tx.commit().await?;
    } else {
        tx.rollback().await?;
    }
    pool.close().await;
    Ok(Report {
        dry_run: options.dry_run,
        committed: commit,
        language: file.header.language,
        account_name,
        account_id: written.account_id,
        account_created: written.account_created,
        rows_read: file.rows.len(),
        blank_rows: file.blank_rows,
        totals_rows: file.totals.len(),
        expenses_created: written.expenses_created,
        settlements_created: written.settlements_created,
        skipped_duplicates: written.skipped_duplicates,
        unmapped_categories: plan.unmapped,
        non_sek_rows: plan.non_sek_rows,
        reconciliation,
    })
}