wanderbooks/crates/ledger-store/src/store.rs

//! [`Store`]: the write path, the SQLite projection, and maintenance.
//!
//! Write-path order for every mutation is **validate → journal → mutate →
//! apply**. Journaling before mutating means a failed write leaves nothing
//! behind: the journal is the source of truth, the in-memory [`Ledger`] is
//! the validation authority, and SQLite is the queryable projection that
//! the next open heals via `last_applied_seq`.

use crate::blob::BlobStore;
use crate::error::StoreError;
use crate::journal::{JournalLine, JournalWriter, read_and_verify_journal};
use crate::schema;
use chrono::{Months, NaiveDate};
use ledger_core::{
    AccountId, AccountType, AssetId, AssetStatus, Currency, EntryId, EntryStatus, Ledger,
    LedgerError, Money, NewAsset, Side, reports,
};
use rusqlite::{Connection, OptionalExtension, params};
use rust_decimal::Decimal;
use serde_json::{Value, json};
use std::path::{Path, PathBuf};

/// File names inside the data directory.
pub const JOURNAL_FILE: &str = "journal.jsonl";
/// File names inside the data directory.
pub const DB_FILE: &str = "wanderbooks.db";
const BLOBS_DIR: &str = "blobs";

/// One row of an entry listing.
#[derive(Clone, Debug)]
pub struct EntryRow {
    /// Entry id.
    pub id: u64,
    /// ISO date.
    pub date: String,
    /// Description.
    pub description: String,
    /// Id of the reversed entry, if this is a reversal.
    pub reversal_of: Option<u64>,
}

/// One row of the event log.
#[derive(Clone, Debug)]
pub struct EventRow {
    /// Event seq (1-based, per projection).
    pub seq: u64,
    /// Journal line type.
    pub event_type: String,
    /// Event payload.
    pub payload: Value,
    /// UTC timestamp.
    pub created_at: String,
}

/// The outcome of posting one month of depreciation for one asset.
#[derive(Clone, Debug)]
pub struct DepreciationRun {
    /// Asset id.
    pub asset_id: u64,
    /// Calendar month, "YYYY-MM".
    pub period: String,
    /// Amount posted, exact cents.
    pub amount: Decimal,
    /// The posted entry's id.
    pub entry_id: u64,
    /// True when this period was already posted (idempotent no-op).
    pub already_posted: bool,
}

/// The outcome of disposing an asset, computed without posting anything.
/// The MCP layer previews this; [`Store::dispose_asset`] posts it.
#[derive(Clone, Debug)]
pub struct DisposalPreview {
    /// Asset id.
    pub asset_id: u64,
    /// Disposal date.
    pub date: chrono::NaiveDate,
    /// Proceeds.
    pub proceeds: Money,
    /// Accumulated depreciation that will be cleared: already posted plus
    /// the final partial period.
    pub accumulated: Decimal,
    /// Depreciation auto-posted for the final partial period. Zero when the
    /// disposal month was already posted.
    pub final_depreciation: Decimal,
    /// Book value at disposal.
    pub book_value: Decimal,
    /// proceeds − book_value. Positive is a gain.
    pub gain_loss: Decimal,
    /// The disposal month, "YYYY-MM".
    pub period: String,
}

/// Attachment metadata (bytes live in the blob store).
#[derive(Clone, Debug)]
pub struct Attachment {
    /// Attachment id.
    pub id: u64,
    /// SHA-256 of the bytes.
    pub blob_sha: String,
    /// Linked entry, if any.
    pub entry_id: Option<u64>,
    /// `receipts`, `formation`, `legal`, `tax`, `accounting`, `other`.
    pub category: String,
    /// Original file name.
    pub file_name: String,
    /// MIME type.
    pub mime_type: String,
    /// Size in bytes.
    pub size_bytes: u64,
    /// Optional description.
    pub description: Option<String>,
    /// UTC timestamp when filed.
    pub created_at: String,
}

/// The store: one writer, one data directory.
///
/// `Store` is `Send` but not `Sync` (it owns a `rusqlite::Connection`).
/// There is exactly one writer per data directory.
pub struct Store {
    data_dir: PathBuf,
    db_path: PathBuf,
    journal: JournalWriter,
    conn: Connection,
    blobs: BlobStore,
    ledger: Ledger,
    base_currency: Currency,
}

/// One filing of bytes into the blob store and journal.
struct AttachRequest<'a> {
    line_type: &'a str,
    category: &'a str,
    entry_id: Option<EntryId>,
    file_name: &'a str,
    mime_type: &'a str,
    bytes: &'a [u8],
    description: Option<&'a str>,
}

impl Store {
    // ------------------------------------------------------------------
    // Open / create
    // ------------------------------------------------------------------

    /// Create a new store. Fails if one already exists.
    pub fn create(data_dir: &Path, base_currency: Currency) -> Result<Self, StoreError> {
        let journal_path = data_dir.join(JOURNAL_FILE);
        if journal_path.exists() {
            return Err(StoreError::AlreadyExists(data_dir.display().to_string()));
        }
        std::fs::create_dir_all(data_dir)?;
        let db_path = data_dir.join(DB_FILE);
        let conn = Self::open_conn(&db_path)?;
        Self::migrate(&conn)?;
        Self::write_meta(&conn, "base_currency", base_currency.code())?;
        Self::write_meta(&conn, "last_applied_seq", "0")?;
        let mut store = Self {
            data_dir: data_dir.to_path_buf(),
            db_path,
            journal: JournalWriter::open(&journal_path)?,
            conn,
            blobs: BlobStore::open(&data_dir.join(BLOBS_DIR))?,
            ledger: Ledger::new(base_currency),
            base_currency,
        };
        let line = store.journal.append(
            "store.created",
            json!({ "base_currency": base_currency.code() }),
        )?;
        apply_line_to_sqlite(&store.conn, &line)?;
        Ok(store)
    }

    /// Open an existing store, verifying the journal chain and catching the
    /// SQLite projection up to the journal head. A missing database file is
    /// healed by full replay: the journal is the source of truth.
    pub fn open(data_dir: &Path) -> Result<Self, StoreError> {
        let journal_path = data_dir.join(JOURNAL_FILE);
        if !journal_path.exists() {
            return Err(StoreError::NotFound(data_dir.display().to_string()));
        }
        // Single verified scan; the writer reuses it.
        let lines = read_and_verify_journal(&journal_path)?;
        let journal = JournalWriter::from_lines(&journal_path, &lines)?;
        let db_path = data_dir.join(DB_FILE);
        let mut fresh_db = !db_path.exists();
        if !fresh_db && !Self::db_file_healthy(&db_path) {
            // The projection is corrupt. The journal is the source of truth,
            // so quarantine the damaged file (for forensics) and rebuild.
            let stamp = chrono::Utc::now().format("%Y%m%dT%H%M%SZ").to_string();
            let quarantine = db_path.with_extension(format!("corrupt.{stamp}"));
            std::fs::rename(&db_path, &quarantine).map_err(StoreError::Io)?;
            // WAL/SHM are derived state; drop them so they can't contaminate
            // the rebuild.
            let _ = std::fs::remove_file(db_path.with_extension("db-wal"));
            let _ = std::fs::remove_file(db_path.with_extension("db-shm"));
            eprintln!(
                "wanderbooks: sqlite projection failed integrity_check; quarantined to {} and rebuilding from journal",
                quarantine.display()
            );
            fresh_db = true;
        }
        let conn = Self::open_conn(&db_path)?;
        Self::migrate(&conn)?;
        let base_currency = if fresh_db {
            // Heal from the journal: currency comes from store.created.
            let first = lines.first().ok_or_else(|| {
                StoreError::JournalCorrupt("journal has no store.created line".into())
            })?;
            if first.line_type != "store.created" {
                return Err(StoreError::JournalCorrupt(
                    "journal's first line is not store.created".into(),
                ));
            }
            let code = first.payload["base_currency"].as_str().ok_or_else(|| {
                StoreError::JournalCorrupt("store.created without base_currency".into())
            })?;
            let cur = Currency::new(code)
                .map_err(|_| StoreError::JournalCorrupt(format!("bad currency {code}")))?;
            Self::write_meta(&conn, "base_currency", cur.code())?;
            Self::write_meta(&conn, "last_applied_seq", "0")?;
            cur
        } else {
            let cur = Self::read_currency(&conn)?;
            // Cross-check the projection against the journal.
            if let Some(first) = lines.first()
                && first.line_type == "store.created"
            {
                let jc = first.payload["base_currency"].as_str().unwrap_or("");
                if jc != cur.code() {
                    return Err(StoreError::JournalCorrupt(format!(
                        "base currency mismatch: journal says {jc}, sqlite says {}",
                        cur.code()
                    )));
                }
            }
            cur
        };
        let mut store = Self {
            data_dir: data_dir.to_path_buf(),
            db_path,
            journal,
            conn,
            blobs: BlobStore::open(&data_dir.join(BLOBS_DIR))?,
            ledger: Ledger::new(base_currency),
            base_currency,
        };
        store.catch_up(&lines)?;
        Ok(store)
    }

    fn open_conn(db_path: &Path) -> Result<Connection, StoreError> {
        let conn = Connection::open(db_path)?;
        conn.execute_batch(
            "PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA foreign_keys=ON;",
        )?;
        Ok(conn)
    }

    /// True when the SQLite file at `db_path` passes `integrity_check`.
    /// Any failure (missing file, not a database, corruption) counts as
    /// unhealthy. Opens a throwaway connection without touching pragmas.
    fn db_file_healthy(db_path: &Path) -> bool {
        let conn = match Connection::open(db_path) {
            Ok(c) => c,
            Err(_) => return false,
        };
        let check: Result<String, _> = conn.query_row("PRAGMA integrity_check", [], |r| r.get(0));
        matches!(check, Ok(s) if s == "ok")
    }

    fn migrate(conn: &Connection) -> Result<(), StoreError> {
        let mut v = Self::current_version(conn)?;
        if v > schema::SCHEMA_VERSION {
            return Err(StoreError::SchemaTooNew(v, schema::SCHEMA_VERSION));
        }
        while v < schema::SCHEMA_VERSION {
            conn.execute_batch(schema::MIGRATIONS[v as usize])?;
            v += 1;
            Self::write_meta(conn, "schema_version", &v.to_string())?;
        }
        Ok(())
    }

    fn current_version(conn: &Connection) -> Result<u32, StoreError> {
        let has: i64 = conn.query_row(
            "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='meta'",
            [],
            |r| r.get(0),
        )?;
        if has == 0 {
            return Ok(0);
        }
        let v: Option<String> = conn
            .query_row(
                "SELECT value FROM meta WHERE key='schema_version'",
                [],
                |r| r.get(0),
            )
            .optional()?;
        Ok(v.and_then(|s| s.parse().ok()).unwrap_or(0))
    }

    fn write_meta(conn: &Connection, key: &str, value: &str) -> Result<(), StoreError> {
        conn.execute(
            "INSERT INTO meta(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
            params![key, value],
        )?;
        Ok(())
    }

    fn read_currency(conn: &Connection) -> Result<Currency, StoreError> {
        let code: String = conn.query_row(
            "SELECT value FROM meta WHERE key='base_currency'",
            [],
            |r| r.get(0),
        )?;
        Currency::new(&code)
            .map_err(|_| StoreError::JournalCorrupt(format!("bad base_currency in meta: {code}")))
    }

    /// Replay the journal: hydrate the in-memory ledger fully, and apply to
    /// SQLite every line past `last_applied_seq`.
    fn catch_up(&mut self, lines: &[JournalLine]) -> Result<(), StoreError> {
        let last_applied: u64 = self
            .conn
            .query_row(
                "SELECT value FROM meta WHERE key='last_applied_seq'",
                [],
                |r| r.get::<_, String>(0),
            )
            .optional()?
            .and_then(|s| s.parse().ok())
            .unwrap_or(0);
        for line in lines {
            replay_line_into_ledger(&mut self.ledger, line, self.base_currency)?;
            if line.seq > last_applied {
                apply_line_to_sqlite(&self.conn, line)?;
            }
        }
        Ok(())
    }

    // ------------------------------------------------------------------
    // Accessors
    // ------------------------------------------------------------------

    /// The live ledger: validation authority and report source.
    pub fn ledger(&self) -> &Ledger {
        &self.ledger
    }

    /// Base currency of this store.
    pub fn base_currency(&self) -> Currency {
        self.base_currency
    }

    /// Data directory.
    pub fn data_dir(&self) -> &Path {
        &self.data_dir
    }

    // ------------------------------------------------------------------
    // Mutations: validate → journal → mutate → apply
    // ------------------------------------------------------------------

    fn append_journal(
        &mut self,
        line_type: &str,
        payload: Value,
    ) -> Result<JournalLine, StoreError> {
        self.journal.append(line_type, payload)
    }

    /// Add an account to the chart.
    pub fn add_account(
        &mut self,
        code: &str,
        name: &str,
        account_type: AccountType,
        parent: Option<AccountId>,
        placeholder: bool,
    ) -> Result<AccountId, StoreError> {
        // Pure pre-validation replicating ChartOfAccounts::add, so the
        // journal write below cannot be followed by a failed mutation.
        let chart = self.ledger.chart();
        if code.is_empty() {
            return Err(LedgerError::EmptyAccountCode.into());
        }
        if chart.find_by_code(code).is_some() {
            return Err(LedgerError::DuplicateAccountCode(code.to_string()).into());
        }
        if let Some(p) = parent
            && chart.get(p).is_none()
        {
            return Err(LedgerError::UnknownParentAccount(p).into());
        }
        let id = self.ledger.next_account_id();
        let line = self.append_journal(
            "account.added",
            json!({
                "id": id,
                "code": code,
                "name": name,
                "account_type": serde_json::to_value(account_type)?,
                "parent": parent,
                "placeholder": placeholder,
                "currency": self.base_currency.code(),
            }),
        )?;
        let live = self
            .ledger
            .add_account(code, name, account_type, parent, placeholder)?;
        debug_assert_eq!(live, id, "id allocator diverged from journal");
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(live)
    }

    /// Start a draft entry. Drafts are journaled as created (for id
    /// determinism) but never touch the projection until posted.
    pub fn create_entry(
        &mut self,
        date: chrono::NaiveDate,
        description: &str,
    ) -> Result<EntryId, StoreError> {
        let id = self.ledger.next_entry_id();
        let line = self.append_journal(
            "entry.created",
            json!({ "id": id, "date": date.to_string(), "description": description }),
        )?;
        let live = self.ledger.create_entry(date, description);
        debug_assert_eq!(live, id, "id allocator diverged from journal");
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(live)
    }

    /// Add a line to a draft entry. Drafts are durable: the line is
    /// journaled as `entry.line_added`, so drafts (and their receipts)
    /// survive restarts. Drafts never touch the SQLite projection until
    /// posted.
    pub fn add_line(
        &mut self,
        entry: EntryId,
        account: AccountId,
        side: Side,
        amount: Money,
    ) -> Result<(), StoreError> {
        self.ledger.add_line(entry, account, side, amount)?;
        let line = self.append_journal(
            "entry.line_added",
            json!({
                "id": entry,
                "account": account,
                "side": match side {
                    Side::Debit => "debit",
                    Side::Credit => "credit",
                },
                "amount": amount.amount.to_string(),
                "currency": amount.currency.code(),
            }),
        )?;
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(())
    }

    /// Validate, journal, and post a draft entry.
    pub fn post_entry(&mut self, entry: EntryId) -> Result<(), StoreError> {
        let snap = self
            .ledger
            .get_entry(entry)
            .ok_or(LedgerError::UnknownEntry(entry))?
            .clone();
        if snap.status() == EntryStatus::Posted {
            return Err(LedgerError::EntryAlreadyPosted(entry).into());
        }
        // Pure validation before anything is journaled or mutated.
        snap.validate(self.ledger.chart())?;
        let line = self.append_journal(
            "entry.posted",
            json!({
                "id": entry,
                "date": snap.date().to_string(),
                "description": snap.description(),
                "reversal_of": snap.reversal_of(),
                "lines": snap.lines().iter().map(journal_line_json).collect::<Vec<_>>(),
            }),
        )?;
        self.ledger.post_entry(entry)?;
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(())
    }

    /// Post a reversing entry for a posted entry.
    pub fn reverse_entry(
        &mut self,
        entry: EntryId,
        date: chrono::NaiveDate,
        description: &str,
    ) -> Result<EntryId, StoreError> {
        let orig = self
            .ledger
            .get_entry(entry)
            .ok_or(LedgerError::UnknownEntry(entry))?
            .clone();
        if orig.status() != EntryStatus::Posted {
            return Err(LedgerError::EntryNotPosted(entry).into());
        }
        let new_id = self.ledger.next_entry_id();
        let flipped: Vec<Value> = orig.lines().iter().map(flip_line_json).collect();
        // Reversals journal a single entry.reversed line (no separate
        // entry.created): replay calls Ledger::reverse_entry, which allocates
        // exactly once, so the id stays deterministic.
        let reversed = self.append_journal(
            "entry.reversed",
            json!({
                "id": new_id,
                "reverses": entry,
                "date": date.to_string(),
                "description": description,
                "lines": flipped,
            }),
        )?;
        let live = self.ledger.reverse_entry(entry, date, description)?;
        debug_assert_eq!(live, new_id, "id allocator diverged from journal");
        apply_line_to_sqlite(&self.conn, &reversed)?;
        Ok(live)
    }

    // ------------------------------------------------------------------
    // Fixed assets
    // ------------------------------------------------------------------

    /// Register a fixed asset. Journaled as `asset.registered`.
    pub fn register_asset(&mut self, new: NewAsset) -> Result<AssetId, StoreError> {
        // Pure pre-validation so the journal write cannot be followed by a
        // failed mutation.
        self.ledger.validate_new_asset(&new)?;
        let id = self.ledger.next_asset_id();
        let line = self.append_journal(
            "asset.registered",
            json!({
                "id": id,
                "description": new.description,
                "purchase_date": new.purchase_date.to_string(),
                "cost": { "amount": new.cost.amount.to_string(), "currency": new.cost.currency.code() },
                "salvage_value": { "amount": new.salvage_value.amount.to_string(), "currency": new.salvage_value.currency.code() },
                "useful_life_months": new.useful_life_months,
                "method": "straight_line",
                "cost_account": new.cost_account,
                "accum_depr_account": new.accum_depr_account,
                "expense_account": new.expense_account,
            }),
        )?;
        let live = self.ledger.register_asset(new)?;
        debug_assert_eq!(live, id, "id allocator diverged from journal");
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(live)
    }

    /// Was depreciation already posted for this asset and period?
    fn depreciation_posted_entry(
        &self,
        asset_id: AssetId,
        period: &str,
    ) -> Result<Option<(u64, Decimal)>, StoreError> {
        let row: Option<(i64, String)> = self
            .conn
            .query_row(
                "SELECT entry_id, amount FROM asset_depreciation WHERE asset_id = ? AND period = ?",
                params![asset_id as i64, period],
                |r| Ok((r.get(0)?, r.get(1)?)),
            )
            .optional()?;
        row.map(|(eid, amount)| {
            let amount = amount
                .parse()
                .map_err(|_| StoreError::JournalCorrupt(format!("bad amount {amount}")))?;
            Ok((eid as u64, amount))
        })
        .transpose()
    }

    /// Sum of depreciation amounts posted for an asset. Summed in Rust with
    /// exact Decimal math: amounts are TEXT in SQLite, so SQLite must not
    /// sum them (REAL would lose pennies).
    fn asset_depreciation_total(&self, asset_id: AssetId) -> Result<Decimal, StoreError> {
        let mut stmt = self
            .conn
            .prepare("SELECT amount FROM asset_depreciation WHERE asset_id = ?")?;
        let amounts = stmt.query_map(params![asset_id as i64], |r| r.get::<_, String>(0))?;
        let mut total = Decimal::ZERO;
        for a in amounts {
            let s: String = a?;
            total += s
                .parse::<Decimal>()
                .map_err(|_| StoreError::JournalCorrupt(format!("bad amount {s}")))?;
        }
        Ok(total)
    }

    /// Post one month of depreciation for one asset. Idempotent: running it
    /// twice for the same period returns the original entry.
    ///
    /// Depreciation is an allocation, not money movement, so it is exempt
    /// from the receipt policy by construction: the registered asset and its
    /// schedule are the supporting documentation.
    pub fn run_depreciation(
        &mut self,
        asset_id: AssetId,
        period: &str,
    ) -> Result<DepreciationRun, StoreError> {
        if let Some((entry_id, amount)) = self.depreciation_posted_entry(asset_id, period)? {
            return Ok(DepreciationRun {
                asset_id,
                period: period.to_string(),
                amount,
                entry_id,
                already_posted: true,
            });
        }
        let asset = self
            .ledger
            .get_asset(asset_id)
            .ok_or(LedgerError::UnknownAsset(asset_id))?
            .clone();
        if asset.status != AssetStatus::Active {
            return Err(LedgerError::AssetAlreadyDisposed(asset_id).into());
        }
        let (year, month) = parse_period(period)?;
        let period_end = last_day_of_month(year, month);
        let sched = ledger_core::depreciation_schedule(&asset, period_end);
        let line = sched.iter().find(|l| l.period == period).ok_or_else(|| {
            StoreError::JournalCorrupt(format!(
                "period {period} is outside asset {asset_id}'s depreciable life"
            ))
        })?;
        let amount = line.amount;
        debug_assert!(amount > Decimal::ZERO);

        let entry_id = self.create_entry(
            period_end,
            &format!("Depreciation — {} — {period}", asset.description),
        )?;
        self.add_line(
            entry_id,
            asset.expense_account,
            Side::Debit,
            Money::new(amount, self.base_currency),
        )?;
        self.add_line(
            entry_id,
            asset.accum_depr_account,
            Side::Credit,
            Money::new(amount, self.base_currency),
        )?;
        self.post_entry(entry_id)?;
        let dline = self.append_journal(
            "asset.depreciation_posted",
            json!({
                "asset_id": asset_id,
                "period": period,
                "entry_id": entry_id,
                "amount": amount.to_string(),
            }),
        )?;
        apply_line_to_sqlite(&self.conn, &dline)?;
        Ok(DepreciationRun {
            asset_id,
            period: period.to_string(),
            amount,
            entry_id,
            already_posted: false,
        })
    }

    /// Compute a disposal without posting. Pure reads plus the posted-vs-
    /// scheduled reconciliation that [`Store::dispose_asset`] performs.
    pub fn dispose_preview(
        &self,
        asset_id: AssetId,
        date: chrono::NaiveDate,
        proceeds: Money,
    ) -> Result<DisposalPreview, StoreError> {
        let asset = self
            .ledger
            .get_asset(asset_id)
            .ok_or(LedgerError::UnknownAsset(asset_id))?;
        if asset.status != AssetStatus::Active {
            return Err(LedgerError::AssetAlreadyDisposed(asset_id).into());
        }
        if proceeds.amount < Decimal::ZERO {
            return Err(LedgerError::NegativeProceeds(proceeds.amount).into());
        }
        if proceeds.currency != self.base_currency {
            return Err(LedgerError::CurrencyMismatch {
                expected: self.base_currency.code().to_string(),
                found: proceeds.currency.code().to_string(),
            }
            .into());
        }
        if date < asset.purchase_date {
            return Err(LedgerError::DisposalBeforePurchase(date.to_string()).into());
        }
        let period = date.format("%Y-%m").to_string();
        let posted = self.asset_depreciation_total(asset_id)?;
        let target = ledger_core::accumulated_depreciation(asset, date);
        let already = self.depreciation_posted_entry(asset_id, &period)?.is_some();
        let final_depreciation = if already {
            Decimal::ZERO
        } else {
            (target - posted).max(Decimal::ZERO)
        };
        let accumulated = posted + final_depreciation;
        let book_value = asset.cost.amount - accumulated;
        Ok(DisposalPreview {
            asset_id,
            date,
            proceeds,
            accumulated,
            final_depreciation,
            book_value,
            gain_loss: proceeds.amount - book_value,
            period,
        })
    }

    /// Dispose of an asset: auto-post any unposted depreciation through the
    /// disposal date, then post the disposal entry (clear cost and
    /// accumulated depreciation, record proceeds and gain/loss) and mark
    /// the asset disposed. Returns the disposal entry's id.
    ///
    /// The auto-post matters: the disposal entry must clear exactly the
    /// accumulated depreciation the books actually carry, not the schedule's
    /// theoretical number. If the agent already ran depreciation for the
    /// disposal month, that posted amount is the books' truth.
    pub fn dispose_asset(
        &mut self,
        asset_id: AssetId,
        date: chrono::NaiveDate,
        proceeds: Money,
        proceeds_account: AccountId,
        gain_loss_account: AccountId,
    ) -> Result<EntryId, StoreError> {
        let asset = self
            .ledger
            .get_asset(asset_id)
            .ok_or(LedgerError::UnknownAsset(asset_id))?
            .clone();
        // The receiving account holds cash (or a receivable): an asset.
        // The gain/loss account is a revenue or expense account.
        for (id, role) in [
            (proceeds_account, "proceeds"),
            (gain_loss_account, "gain/loss"),
        ] {
            let a = self
                .ledger
                .chart()
                .get(id)
                .ok_or(LedgerError::UnknownAccount(id))?;
            if a.placeholder {
                return Err(LedgerError::PlaceholderAccount(a.code.clone()).into());
            }
            let ok = match role {
                "proceeds" => a.account_type == AccountType::Asset,
                _ => {
                    a.account_type == AccountType::Revenue || a.account_type == AccountType::Expense
                }
            };
            if !ok {
                return Err(LedgerError::AssetAccountWrongType {
                    code: a.code.clone(),
                    role: "disposal",
                    expected: if role == "proceeds" {
                        "asset"
                    } else {
                        "revenue or expense"
                    },
                    found: match a.account_type {
                        AccountType::Asset => "asset",
                        AccountType::Liability => "liability",
                        AccountType::Equity => "equity",
                        AccountType::Revenue => "revenue",
                        AccountType::Expense => "expense",
                    },
                }
                .into());
            }
        }
        if proceeds.amount < Decimal::ZERO {
            return Err(LedgerError::NegativeProceeds(proceeds.amount).into());
        }
        if proceeds.currency != self.base_currency {
            return Err(LedgerError::CurrencyMismatch {
                expected: self.base_currency.code().to_string(),
                found: proceeds.currency.code().to_string(),
            }
            .into());
        }
        if asset.status != AssetStatus::Active {
            return Err(LedgerError::AssetAlreadyDisposed(asset_id).into());
        }
        if date < asset.purchase_date {
            return Err(LedgerError::DisposalBeforePurchase(date.to_string()).into());
        }
        let base = self.base_currency;

        // One source of truth for the math: the preview computes the
        // posted-vs-scheduled reconciliation; posting follows it.
        let preview = self.dispose_preview(asset_id, date, proceeds)?;
        let accumulated = preview.accumulated;
        let gain_loss = preview.gain_loss;
        if preview.final_depreciation > Decimal::ZERO {
            let diff = preview.final_depreciation;
            let eid = self.create_entry(
                date,
                &format!(
                    "Depreciation — {} — {} (final)",
                    asset.description, preview.period
                ),
            )?;
            self.add_line(
                eid,
                asset.expense_account,
                Side::Debit,
                Money::new(diff, base),
            )?;
            self.add_line(
                eid,
                asset.accum_depr_account,
                Side::Credit,
                Money::new(diff, base),
            )?;
            self.post_entry(eid)?;
            let dline = self.append_journal(
                "asset.depreciation_posted",
                json!({
                    "asset_id": asset_id,
                    "period": preview.period,
                    "entry_id": eid,
                    "amount": diff.to_string(),
                }),
            )?;
            apply_line_to_sqlite(&self.conn, &dline)?;
        }

        let entry_id = self.create_entry(
            date,
            &format!("Disposal — {} — {}", asset.description, date),
        )?;
        self.add_line(
            entry_id,
            asset.accum_depr_account,
            Side::Debit,
            Money::new(accumulated, base),
        )?;
        self.add_line(
            entry_id,
            proceeds_account,
            Side::Debit,
            Money::new(proceeds.amount, base),
        )?;
        self.add_line(
            entry_id,
            asset.cost_account,
            Side::Credit,
            Money::new(asset.cost.amount, base),
        )?;
        if gain_loss > Decimal::ZERO {
            self.add_line(
                entry_id,
                gain_loss_account,
                Side::Credit,
                Money::new(gain_loss, base),
            )?;
        } else if gain_loss < Decimal::ZERO {
            self.add_line(
                entry_id,
                gain_loss_account,
                Side::Debit,
                Money::new(-gain_loss, base),
            )?;
        }
        self.post_entry(entry_id)?;
        let line = self.append_journal(
            "asset.disposed",
            json!({
                "asset_id": asset_id,
                "date": date.to_string(),
                "proceeds": { "amount": proceeds.amount.to_string(), "currency": proceeds.currency.code() },
                "proceeds_account": proceeds_account,
                "gain_loss_account": gain_loss_account,
                "entry_id": entry_id,
            }),
        )?;
        self.ledger.dispose_asset(asset_id, date, proceeds)?;
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(entry_id)
    }

    // ------------------------------------------------------------------
    // Blobs and attachments
    // ------------------------------------------------------------------

    /// Attach a receipt to an entry (or file it unlinked with `entry_id=None`
    /// and link later). Returns `(attachment_id, sha256, deduplicated)`.
    pub fn attach_receipt(
        &mut self,
        entry_id: Option<EntryId>,
        file_name: &str,
        mime_type: &str,
        bytes: &[u8],
        description: Option<&str>,
    ) -> Result<(u64, String, bool), StoreError> {
        self.attach(AttachRequest {
            line_type: "receipt.attached",
            category: "receipts",
            entry_id,
            file_name,
            mime_type,
            bytes,
            description,
        })
    }

    /// File a document under a category (`formation`, `legal`, `tax`,
    /// `accounting`, `other`).
    pub fn file_document(
        &mut self,
        category: &str,
        file_name: &str,
        mime_type: &str,
        bytes: &[u8],
        description: Option<&str>,
    ) -> Result<(u64, String, bool), StoreError> {
        self.attach(AttachRequest {
            line_type: "document.filed",
            category,
            entry_id: None,
            file_name,
            mime_type,
            bytes,
            description,
        })
    }

    /// One filing of bytes into the blob store and journal.
    fn attach(&mut self, req: AttachRequest<'_>) -> Result<(u64, String, bool), StoreError> {
        if let Some(eid) = req.entry_id {
            self.ledger
                .get_entry(eid)
                .ok_or(LedgerError::UnknownEntry(eid))?;
        }
        // Bytes hit the content-addressed store first; an orphan blob from a
        // failed journal append is harmless (unreferenced, deduplicated).
        let (sha, dedup) = self.blobs.put(req.bytes)?;
        let id: u64 = self.conn.query_row(
            "SELECT COALESCE(MAX(id), 0) + 1 FROM attachments",
            [],
            |r| r.get(0),
        )?;
        let line = self.append_journal(
            req.line_type,
            json!({
                "attachment_id": id,
                "blob_sha": sha,
                "entry_id": req.entry_id,
                "category": req.category,
                "file_name": req.file_name,
                "mime_type": req.mime_type,
                "size_bytes": req.bytes.len() as u64,
                "description": req.description,
            }),
        )?;
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok((id, sha, dedup))
    }

    /// Link a previously filed receipt to an entry.
    pub fn link_receipt(
        &mut self,
        attachment_id: u64,
        entry_id: EntryId,
    ) -> Result<(), StoreError> {
        self.ledger
            .get_entry(entry_id)
            .ok_or(LedgerError::UnknownEntry(entry_id))?;
        let exists: i64 = self.conn.query_row(
            "SELECT COUNT(*) FROM attachments WHERE id = ?",
            params![attachment_id as i64],
            |r| r.get(0),
        )?;
        if exists == 0 {
            return Err(StoreError::UnknownAttachment(attachment_id));
        }
        let line = self.append_journal(
            "receipt.linked",
            json!({ "attachment_id": attachment_id, "entry_id": entry_id }),
        )?;
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(())
    }

    /// List attachments, optionally filtered to one entry.
    pub fn list_attachments(
        &self,
        entry_id: Option<EntryId>,
    ) -> Result<Vec<Attachment>, StoreError> {
        let mut sql = String::from(
            "SELECT a.id, a.blob_sha, a.entry_id, a.category, b.original_name, b.mime_type, \
             b.size_bytes, a.description, a.created_at FROM attachments a \
             JOIN blobs b ON b.sha256 = a.blob_sha",
        );
        if entry_id.is_some() {
            sql.push_str(" WHERE a.entry_id = ?");
        }
        sql.push_str(" ORDER BY a.id");
        let mut stmt = self.conn.prepare(&sql)?;
        let rows: Vec<Attachment> = match entry_id {
            Some(eid) => stmt
                .query_map(params![eid as i64], row_to_attachment)?
                .collect::<Result<_, _>>()?,
            None => stmt
                .query_map([], row_to_attachment)?
                .collect::<Result<_, _>>()?,
        };
        Ok(rows)
    }

    /// Read attachment bytes, verifying the SHA-256.
    pub fn attachment_bytes(&self, attachment_id: u64) -> Result<Vec<u8>, StoreError> {
        let sha: String = self.conn.query_row(
            "SELECT blob_sha FROM attachments WHERE id = ?",
            params![attachment_id as i64],
            |r| r.get(0),
        )?;
        self.blobs.get(&sha)
    }

    // ------------------------------------------------------------------
    // Queries
    // ------------------------------------------------------------------

    /// Search posted entries with optional filters. Newest first.
    pub fn search_entries(
        &self,
        account: Option<AccountId>,
        date_from: Option<&str>,
        date_to: Option<&str>,
        text: Option<&str>,
        limit: u64,
        offset: u64,
    ) -> Result<Vec<EntryRow>, StoreError> {
        let mut sql = String::from(
            "SELECT DISTINCT e.id, e.date, e.description, e.reversal_of FROM entries e",
        );
        if account.is_some() {
            sql.push_str(" JOIN lines l ON l.entry_id = e.id");
        }
        let mut conds: Vec<String> = Vec::new();
        let mut args: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
        if let Some(a) = account {
            conds.push("l.account_id = ?".to_string());
            args.push(Box::new(a as i64));
        }
        if let Some(d) = date_from {
            conds.push("e.date >= ?".to_string());
            args.push(Box::new(d.to_string()));
        }
        if let Some(d) = date_to {
            conds.push("e.date <= ?".to_string());
            args.push(Box::new(d.to_string()));
        }
        if let Some(t) = text {
            conds.push("e.description LIKE ?".to_string());
            args.push(Box::new(format!("%{t}%")));
        }
        if !conds.is_empty() {
            sql.push_str(" WHERE ");
            sql.push_str(&conds.join(" AND "));
        }
        sql.push_str(" ORDER BY e.date DESC, e.id DESC LIMIT ? OFFSET ?");
        args.push(Box::new(limit as i64));
        args.push(Box::new(offset as i64));
        let refs: Vec<&dyn rusqlite::ToSql> = args.iter().map(|b| b.as_ref()).collect();
        let mut stmt = self.conn.prepare(&sql)?;
        let rows = stmt.query_map(refs.as_slice(), |r| {
            Ok(EntryRow {
                id: r.get::<_, i64>(0)? as u64,
                date: r.get(1)?,
                description: r.get(2)?,
                reversal_of: r.get::<_, Option<i64>>(3)?.map(|v| v as u64),
            })
        })?;
        Ok(rows.collect::<Result<_, _>>()?)
    }

    /// Tail the event log from `seq` (exclusive).
    pub fn events_since(&self, seq: u64, limit: u64) -> Result<Vec<EventRow>, StoreError> {
        let mut stmt = self.conn.prepare(
            "SELECT seq, type, payload, created_at FROM events WHERE seq > ? ORDER BY seq LIMIT ?",
        )?;
        let rows = stmt.query_map(params![seq as i64, limit as i64], |r| {
            let payload_str: String = r.get(2)?;
            Ok(EventRow {
                seq: r.get::<_, i64>(0)? as u64,
                event_type: r.get(1)?,
                payload: serde_json::from_str(&payload_str).unwrap_or(Value::Null),
                created_at: r.get(3)?,
            })
        })?;
        Ok(rows.collect::<Result<_, _>>()?)
    }

    // ------------------------------------------------------------------
    // Maintenance: verify and rebuild
    // ------------------------------------------------------------------

    /// Verify everything: the journal hash chain, a fresh replay's reports
    /// against the live ledger, and the SQLite projection against the replay.
    /// Loud on any divergence.
    pub fn verify(&self) -> Result<(), StoreError> {
        let lines = read_and_verify_journal(self.journal.path())?;
        let mut replayed = Ledger::new(self.base_currency);
        for line in &lines {
            replay_line_into_ledger(&mut replayed, line, self.base_currency)?;
        }
        let a = reports::snapshot(&replayed).map_err(StoreError::Ledger)?;
        let b = reports::snapshot(&self.ledger).map_err(StoreError::Ledger)?;
        if a != b {
            return Err(StoreError::VerifyMismatch(
                "live ledger diverged from journal replay".into(),
            ));
        }
        let c = self.load_ledger_from_sqlite()?;
        let c_snap = reports::snapshot(&c).map_err(StoreError::Ledger)?;
        if a != c_snap {
            return Err(StoreError::VerifyMismatch(
                "sqlite projection diverged from journal".into(),
            ));
        }
        // The accounting snapshot ignores the asset register, so compare it
        // explicitly: journal replay, live ledger, and projection must agree.
        let assets_of = |l: &Ledger| l.registered_assets().cloned().collect::<Vec<_>>();
        let (ra, rb, rc) = (assets_of(&replayed), assets_of(&self.ledger), assets_of(&c));
        if ra != rb || ra != rc {
            return Err(StoreError::VerifyMismatch(
                "asset register diverged between journal, live ledger, and sqlite projection"
                    .into(),
            ));
        }
        Ok(())
    }

    /// Verify every blob's bytes against its SHA-256. Returns the count
    /// checked. Reads all bytes; run on a schedule, not per request.
    pub fn verify_blobs(&self) -> Result<u64, StoreError> {
        let mut stmt = self.conn.prepare("SELECT sha256 FROM blobs")?;
        let shas: Vec<String> = stmt
            .query_map([], |r| r.get(0))?
            .collect::<Result<_, _>>()?;
        let mut n = 0u64;
        for sha in shas {
            self.blobs.get(&sha)?;
            n += 1;
        }
        Ok(n)
    }

    /// Drop the SQLite projection and rebuild it from the journal, then
    /// re-hydrate the live ledger. The answer to "what if SQLite corrupts":
    /// don't investigate — rebuild.
    pub fn rebuild(&mut self) -> Result<(), StoreError> {
        let old = std::mem::replace(&mut self.conn, Connection::open_in_memory()?);
        old.close().map_err(|(_, e)| StoreError::Rusqlite(e))?;
        std::fs::remove_file(&self.db_path)?;
        self.conn = Self::open_conn(&self.db_path)?;
        Self::migrate(&self.conn)?;
        Self::write_meta(&self.conn, "base_currency", self.base_currency.code())?;
        Self::write_meta(&self.conn, "last_applied_seq", "0")?;
        let lines = read_and_verify_journal(self.journal.path())?;
        for line in &lines {
            apply_line_to_sqlite(&self.conn, line)?;
        }
        self.ledger = Ledger::new(self.base_currency);
        for line in &lines {
            replay_line_into_ledger(&mut self.ledger, line, self.base_currency)?;
        }
        Ok(())
    }

    /// Rebuild a ledger purely from the SQLite projection (used by verify).
    fn load_ledger_from_sqlite(&self) -> Result<Ledger, StoreError> {
        let mut ledger = Ledger::new(self.base_currency);
        let mut stmt = self.conn.prepare(
            "SELECT id, code, name, type, parent, placeholder, currency FROM accounts ORDER BY id",
        )?;
        let accounts = stmt.query_map([], |r| {
            Ok((
                r.get::<_, i64>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, String>(3)?,
                r.get::<_, Option<i64>>(4)?,
                r.get::<_, i64>(5)?,
                r.get::<_, String>(6)?,
            ))
        })?;
        for a in accounts {
            let (id, code, name, typ, parent, placeholder, currency) = a?;
            let account_type: AccountType = serde_json::from_value(Value::String(typ))?;
            let cur = Currency::new(&currency)
                .map_err(|_| StoreError::JournalCorrupt(format!("bad currency {currency}")))?;
            if cur != self.base_currency {
                return Err(StoreError::JournalCorrupt(format!(
                    "account {code} currency {currency} != base"
                )));
            }
            let got = ledger.add_account(
                &code,
                &name,
                account_type,
                parent.map(|p| p as u64),
                placeholder != 0,
            )?;
            if got != id as u64 {
                return Err(StoreError::ReplayMismatch(format!(
                    "account id: sqlite says {id}, replay got {got}"
                )));
            }
        }
        let mut estmt = self
            .conn
            .prepare("SELECT id, date, description, reversal_of FROM entries ORDER BY id")?;
        let entries = estmt.query_map([], |r| {
            Ok((
                r.get::<_, i64>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, Option<i64>>(3)?,
            ))
        })?;
        for e in entries {
            let (id, date, description, reversal_of) = e?;
            let date = chrono::NaiveDate::parse_from_str(&date, "%Y-%m-%d")
                .map_err(|_| StoreError::JournalCorrupt(format!("bad date {date}")))?;
            match reversal_of {
                Some(orig) => {
                    // Reversals allocate the next id; replayed in id order the
                    // allocator is in sync, so the id must match the row.
                    let got = ledger.reverse_entry(orig as u64, date, &description)?;
                    if got != id as u64 {
                        return Err(StoreError::ReplayMismatch(format!(
                            "reversal id: sqlite says {id}, replay got {got}"
                        )));
                    }
                }
                None => {
                    let got = ledger.create_entry(date, &description);
                    if got != id as u64 {
                        return Err(StoreError::ReplayMismatch(format!(
                            "entry id: sqlite says {id}, replay got {got}"
                        )));
                    }
                    let mut lstmt = self.conn.prepare(
                        "SELECT account_id, side, amount, currency FROM lines WHERE entry_id = ? ORDER BY id",
                    )?;
                    let lrows = lstmt.query_map(params![id], |r| {
                        Ok((
                            r.get::<_, i64>(0)?,
                            r.get::<_, String>(1)?,
                            r.get::<_, String>(2)?,
                            r.get::<_, String>(3)?,
                        ))
                    })?;
                    for lr in lrows {
                        let (account, side, amount, currency) = lr?;
                        ledger.add_line(
                            got,
                            account as u64,
                            parse_side(&side)?,
                            Money::new(
                                amount.parse().map_err(|_| {
                                    StoreError::JournalCorrupt(format!("bad amount {amount}"))
                                })?,
                                Currency::new(&currency).map_err(|_| {
                                    StoreError::JournalCorrupt(format!("bad currency {currency}"))
                                })?,
                            ),
                        )?;
                    }
                    ledger.post_entry(got)?;
                }
            }
        }
        // Assets: registration order is id order; disposal is a per-row flag.
        let mut astmt = self.conn.prepare(
            "SELECT id, description, purchase_date, cost_amount, cost_currency, salvage_amount, salvage_currency, useful_life_months, method, cost_account, accum_depr_account, expense_account, status, disposed_date, proceeds_amount FROM assets ORDER BY id",
        )?;
        let assets = astmt.query_map([], |r| {
            Ok((
                r.get::<_, i64>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, String>(3)?,
                r.get::<_, String>(4)?,
                r.get::<_, String>(5)?,
                r.get::<_, String>(6)?,
                r.get::<_, i64>(7)?,
                r.get::<_, String>(8)?,
                r.get::<_, i64>(9)?,
                r.get::<_, i64>(10)?,
                r.get::<_, i64>(11)?,
                r.get::<_, String>(12)?,
                r.get::<_, Option<String>>(13)?,
                r.get::<_, Option<String>>(14)?,
            ))
        })?;
        for a in assets {
            let (
                id,
                description,
                purchase_date,
                cost_amount,
                cost_currency,
                salvage_amount,
                salvage_currency,
                life_months,
                method,
                cost_account,
                accum_depr_account,
                expense_account,
                status,
                disposed_date,
                proceeds_amount,
            ) = a?;
            if method != "straight_line" {
                return Err(StoreError::JournalCorrupt(format!(
                    "unknown method {method}"
                )));
            }
            let parse_amt = |s: &str, c: &str| -> Result<Money, StoreError> {
                Ok(Money::new(
                    s.parse()
                        .map_err(|_| StoreError::JournalCorrupt(format!("bad amount {s}")))?,
                    Currency::new(c)
                        .map_err(|_| StoreError::JournalCorrupt(format!("bad currency {c}")))?,
                ))
            };
            let got = ledger.register_asset(NewAsset {
                description,
                purchase_date: parse_date(&purchase_date)?,
                cost: parse_amt(&cost_amount, &cost_currency)?,
                salvage_value: parse_amt(&salvage_amount, &salvage_currency)?,
                useful_life_months: life_months as u32,
                cost_account: cost_account as u64,
                accum_depr_account: accum_depr_account as u64,
                expense_account: expense_account as u64,
            })?;
            if got != id as u64 {
                return Err(StoreError::ReplayMismatch(format!(
                    "asset id: sqlite says {id}, replay got {got}"
                )));
            }
            if status == "disposed" {
                let date = disposed_date
                    .as_deref()
                    .ok_or_else(|| StoreError::JournalCorrupt("disposed without date".into()))?;
                let proceeds = proceeds_amount.as_deref().ok_or_else(|| {
                    StoreError::JournalCorrupt("disposed without proceeds".into())
                })?;
                ledger.dispose_asset(
                    got,
                    parse_date(date)?,
                    parse_amt(proceeds, &cost_currency)?,
                )?;
            }
        }
        Ok(ledger)
    }
}

// ---------------------------------------------------------------------------
// Replay and apply helpers
// ---------------------------------------------------------------------------

fn journal_line_json(l: &ledger_core::JournalLine) -> Value {
    let side = match l.side {
        Side::Debit => "debit",
        Side::Credit => "credit",
    };
    json!({
        "account": l.account,
        "side": side,
        "amount": l.amount.amount.to_string(),
        "currency": l.amount.currency.code(),
    })
}

fn flip_line_json(l: &ledger_core::JournalLine) -> Value {
    let side = match l.side {
        Side::Debit => "credit",
        Side::Credit => "debit",
    };
    json!({
        "account": l.account,
        "side": side,
        "amount": l.amount.amount.to_string(),
        "currency": l.amount.currency.code(),
    })
}

fn parse_side(s: &str) -> Result<Side, StoreError> {
    match s {
        "debit" => Ok(Side::Debit),
        "credit" => Ok(Side::Credit),
        _ => Err(StoreError::JournalCorrupt(format!("bad side {s}"))),
    }
}

fn parse_date(s: &str) -> Result<chrono::NaiveDate, StoreError> {
    chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d")
        .map_err(|_| StoreError::JournalCorrupt(format!("bad date {s}")))
}

/// Parse a nested {"amount","currency"} money object.
fn parse_money(p: &Value, key: &str) -> Result<Money, StoreError> {
    let o = p
        .get(key)
        .ok_or_else(|| StoreError::JournalCorrupt(format!("missing '{key}'")))?;
    Ok(Money::new(
        get_str(o, "amount")?
            .parse()
            .map_err(|_| StoreError::JournalCorrupt("bad amount".into()))?,
        Currency::new(get_str(o, "currency")?)
            .map_err(|_| StoreError::JournalCorrupt("bad currency".into()))?,
    ))
}

/// Parse a calendar-month period "YYYY-MM".
fn parse_period(period: &str) -> Result<(i32, u32), StoreError> {
    let err = || StoreError::JournalCorrupt(format!("bad period '{period}': want YYYY-MM"));
    let mut parts = period.split('-');
    let year: i32 = parts.next().ok_or_else(err)?.parse().map_err(|_| err())?;
    let month: u32 = parts.next().ok_or_else(err)?.parse().map_err(|_| err())?;
    if parts.next().is_some() || year < 1 || !(1..=12).contains(&month) {
        return Err(err());
    }
    Ok((year, month))
}

/// Last calendar day of a month.
fn last_day_of_month(year: i32, month: u32) -> NaiveDate {
    let first = NaiveDate::from_ymd_opt(year, month, 1).expect("valid period");
    first
        .checked_add_months(Months::new(1))
        .expect("month arithmetic")
        .pred_opt()
        .expect("month arithmetic")
}

fn get_u64(p: &Value, key: &str) -> Result<u64, StoreError> {
    p.get(key)
        .and_then(|v| v.as_u64())
        .ok_or_else(|| StoreError::JournalCorrupt(format!("missing/invalid '{key}'")))
}

fn get_str<'a>(p: &'a Value, key: &str) -> Result<&'a str, StoreError> {
    p.get(key)
        .and_then(|v| v.as_str())
        .ok_or_else(|| StoreError::JournalCorrupt(format!("missing/invalid '{key}'")))
}

/// Replay one journal line into a ledger, asserting recorded ids. Used by
/// hydration, `verify`, and `rebuild`: the journal is the script, the engine
/// is the actor.
fn replay_line_into_ledger(
    ledger: &mut Ledger,
    line: &JournalLine,
    base_currency: Currency,
) -> Result<(), StoreError> {
    let p = &line.payload;
    match line.line_type.as_str() {
        "store.created" => {
            let code = get_str(p, "base_currency")?;
            if Currency::new(code)
                .map_err(|_| StoreError::JournalCorrupt(format!("bad currency {code}")))?
                != base_currency
            {
                return Err(StoreError::JournalCorrupt(
                    "base currency changed mid-journal".into(),
                ));
            }
        }
        "account.added" => {
            let want = get_u64(p, "id")?;
            let account_type: AccountType = serde_json::from_value(p["account_type"].clone())?;
            let parent = p.get("parent").and_then(|v| v.as_u64());
            let got = ledger.add_account(
                get_str(p, "code")?,
                get_str(p, "name")?,
                account_type,
                parent,
                p.get("placeholder")
                    .and_then(|v| v.as_bool())
                    .unwrap_or(false),
            )?;
            if got != want {
                return Err(StoreError::ReplayMismatch(format!(
                    "account id: journal says {want}, replay got {got}"
                )));
            }
        }
        "entry.created" => {
            let want = get_u64(p, "id")?;
            let got =
                ledger.create_entry(parse_date(get_str(p, "date")?)?, get_str(p, "description")?);
            if got != want {
                return Err(StoreError::ReplayMismatch(format!(
                    "entry id: journal says {want}, replay got {got}"
                )));
            }
        }
        "entry.line_added" => {
            let id = get_u64(p, "id")?;
            ledger.add_line(
                id,
                get_u64(p, "account")?,
                parse_side(get_str(p, "side")?)?,
                Money::new(
                    get_str(p, "amount")?
                        .parse()
                        .map_err(|_| StoreError::JournalCorrupt("bad amount".into()))?,
                    Currency::new(get_str(p, "currency")?)?,
                ),
            )?;
        }
        "entry.posted" => {
            let id = get_u64(p, "id")?;
            // New journals build the draft via entry.line_added events; old
            // journals (pre-durable-drafts) only have the lines in the
            // posted payload. If the draft already has lines, just post.
            let has_lines = ledger
                .get_entry(id)
                .map(|e| !e.lines().is_empty())
                .unwrap_or(false);
            if !has_lines {
                for l in p.get("lines").and_then(|v| v.as_array()).ok_or_else(|| {
                    StoreError::JournalCorrupt("entry.posted without lines".into())
                })? {
                    ledger.add_line(
                        id,
                        get_u64(l, "account")?,
                        parse_side(get_str(l, "side")?)?,
                        Money::new(
                            get_str(l, "amount")?
                                .parse()
                                .map_err(|_| StoreError::JournalCorrupt("bad amount".into()))?,
                            Currency::new(get_str(l, "currency")?)?,
                        ),
                    )?;
                }
            }
            ledger.post_entry(id)?;
        }
        "entry.reversed" => {
            let want = get_u64(p, "id")?;
            let orig = get_u64(p, "reverses")?;
            let got = ledger.reverse_entry(
                orig,
                parse_date(get_str(p, "date")?)?,
                get_str(p, "description")?,
            )?;
            if got != want {
                return Err(StoreError::ReplayMismatch(format!(
                    "reversal id: journal says {want}, replay got {got}"
                )));
            }
        }
        "receipt.attached" | "receipt.linked" | "document.filed" => {
            // Not accounting state; the projection records them.
        }
        "asset.registered" => {
            let want = get_u64(p, "id")?;
            let got = ledger.register_asset(NewAsset {
                description: get_str(p, "description")?.to_string(),
                purchase_date: parse_date(get_str(p, "purchase_date")?)?,
                cost: parse_money(p, "cost")?,
                salvage_value: parse_money(p, "salvage_value")?,
                useful_life_months: get_u64(p, "useful_life_months")? as u32,
                cost_account: get_u64(p, "cost_account")?,
                accum_depr_account: get_u64(p, "accum_depr_account")?,
                expense_account: get_u64(p, "expense_account")?,
            })?;
            if got != want {
                return Err(StoreError::ReplayMismatch(format!(
                    "asset id: journal says {want}, replay got {got}"
                )));
            }
        }
        "asset.depreciation_posted" => {
            // The entry itself was journaled as entry.created/entry.posted;
            // this line is idempotency bookkeeping for the projection.
        }
        "asset.disposed" => {
            let id = get_u64(p, "asset_id")?;
            ledger.dispose_asset(
                id,
                parse_date(get_str(p, "date")?)?,
                parse_money(p, "proceeds")?,
            )?;
        }
        other => {
            return Err(StoreError::JournalCorrupt(format!(
                "unknown line type: {other}"
            )));
        }
    }
    Ok(())
}

/// Apply one journal line to the SQLite projection: INSERTs plus the events
/// mirror, then advance `last_applied_seq`. The applier only ever INSERTs
/// (and one UPDATE for receipt linking).
fn apply_line_to_sqlite(conn: &Connection, line: &JournalLine) -> Result<(), StoreError> {
    let p = &line.payload;
    match line.line_type.as_str() {
        "store.created" => {}
        "account.added" => {
            conn.execute(
                "INSERT INTO accounts(id, code, name, type, parent, placeholder, currency) VALUES (?,?,?,?,?,?,?)",
                params![
                    get_u64(p, "id")? as i64,
                    get_str(p, "code")?,
                    get_str(p, "name")?,
                    p["account_type"].as_str().unwrap_or(""),
                    p.get("parent").and_then(|v| v.as_u64()).map(|v| v as i64),
                    p.get("placeholder").and_then(|v| v.as_bool()).unwrap_or(false) as i64,
                    get_str(p, "currency")?,
                ],
            )?;
        }
        "entry.created" => {}
        // Draft lines are durable in the journal but never touch the
        // projection until posted.
        "entry.line_added" => {}
        "entry.posted" => {
            conn.execute(
                "INSERT INTO entries(id, date, description, status, reversal_of) VALUES (?,?,?,'posted',?)",
                params![
                    get_u64(p, "id")? as i64,
                    get_str(p, "date")?,
                    get_str(p, "description")?,
                    p.get("reversal_of").and_then(|v| v.as_u64()).map(|v| v as i64),
                ],
            )?;
            for l in p
                .get("lines")
                .and_then(|v| v.as_array())
                .ok_or_else(|| StoreError::JournalCorrupt("entry.posted without lines".into()))?
            {
                conn.execute(
                    "INSERT INTO lines(entry_id, account_id, side, amount, currency) VALUES (?,?,?,?,?)",
                    params![
                        get_u64(p, "id")? as i64,
                        get_u64(l, "account")? as i64,
                        get_str(l, "side")?,
                        get_str(l, "amount")?,
                        get_str(l, "currency")?,
                    ],
                )?;
            }
        }
        "entry.reversed" => {
            conn.execute(
                "INSERT INTO entries(id, date, description, status, reversal_of) VALUES (?,?,?,'posted',?)",
                params![
                    get_u64(p, "id")? as i64,
                    get_str(p, "date")?,
                    get_str(p, "description")?,
                    get_u64(p, "reverses")? as i64,
                ],
            )?;
            for l in p
                .get("lines")
                .and_then(|v| v.as_array())
                .ok_or_else(|| StoreError::JournalCorrupt("entry.reversed without lines".into()))?
            {
                conn.execute(
                    "INSERT INTO lines(entry_id, account_id, side, amount, currency) VALUES (?,?,?,?,?)",
                    params![
                        get_u64(p, "id")? as i64,
                        get_u64(l, "account")? as i64,
                        get_str(l, "side")?,
                        get_str(l, "amount")?,
                        get_str(l, "currency")?,
                    ],
                )?;
            }
        }
        "receipt.attached" | "document.filed" => {
            conn.execute(
                "INSERT OR IGNORE INTO blobs(sha256, size_bytes, mime_type, original_name, created_at) VALUES (?,?,?,?,?)",
                params![
                    get_str(p, "blob_sha")?,
                    get_u64(p, "size_bytes")? as i64,
                    get_str(p, "mime_type")?,
                    get_str(p, "file_name")?,
                    line.ts,
                ],
            )?;
            conn.execute(
                "INSERT INTO attachments(id, blob_sha, entry_id, category, description, created_at) VALUES (?,?,?,?,?,?)",
                params![
                    get_u64(p, "attachment_id")? as i64,
                    get_str(p, "blob_sha")?,
                    p.get("entry_id").and_then(|v| v.as_u64()).map(|v| v as i64),
                    get_str(p, "category")?,
                    p.get("description").and_then(|v| v.as_str()),
                    line.ts,
                ],
            )?;
        }
        "receipt.linked" => {
            conn.execute(
                "UPDATE attachments SET entry_id = ? WHERE id = ?",
                params![
                    get_u64(p, "entry_id")? as i64,
                    get_u64(p, "attachment_id")? as i64,
                ],
            )?;
        }
        "asset.registered" => {
            let cost = &p["cost"];
            let salvage = &p["salvage_value"];
            conn.execute(
                "INSERT INTO assets(id, description, purchase_date, cost_amount, cost_currency, salvage_amount, salvage_currency, useful_life_months, method, cost_account, accum_depr_account, expense_account, status) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
                params![
                    get_u64(p, "id")? as i64,
                    get_str(p, "description")?,
                    get_str(p, "purchase_date")?,
                    get_str(cost, "amount")?,
                    get_str(cost, "currency")?,
                    get_str(salvage, "amount")?,
                    get_str(salvage, "currency")?,
                    get_u64(p, "useful_life_months")? as i64,
                    get_str(p, "method")?,
                    get_u64(p, "cost_account")? as i64,
                    get_u64(p, "accum_depr_account")? as i64,
                    get_u64(p, "expense_account")? as i64,
                    "active",
                ],
            )?;
        }
        "asset.depreciation_posted" => {
            conn.execute(
                "INSERT OR IGNORE INTO asset_depreciation(asset_id, period, entry_id, amount) VALUES (?,?,?,?)",
                params![
                    get_u64(p, "asset_id")? as i64,
                    get_str(p, "period")?,
                    get_u64(p, "entry_id")? as i64,
                    get_str(p, "amount")?,
                ],
            )?;
        }
        "asset.disposed" => {
            let proceeds = &p["proceeds"];
            conn.execute(
                "UPDATE assets SET status = 'disposed', disposed_date = ?, proceeds_amount = ? WHERE id = ?",
                params![
                    get_str(p, "date")?,
                    get_str(proceeds, "amount")?,
                    get_u64(p, "asset_id")? as i64,
                ],
            )?;
        }
        other => {
            return Err(StoreError::JournalCorrupt(format!(
                "unknown line type: {other}"
            )));
        }
    }
    conn.execute(
        "INSERT INTO events(type, payload, created_at) VALUES (?,?,?)",
        params![line.line_type, p.to_string(), line.ts],
    )?;
    conn.execute(
        "INSERT INTO meta(key, value) VALUES ('last_applied_seq', ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
        params![line.seq.to_string()],
    )?;
    Ok(())
}

fn row_to_attachment(r: &rusqlite::Row) -> rusqlite::Result<Attachment> {
    Ok(Attachment {
        id: r.get::<_, i64>(0)? as u64,
        blob_sha: r.get(1)?,
        entry_id: r.get::<_, Option<i64>>(2)?.map(|v| v as u64),
        category: r.get(3)?,
        file_name: r.get(4)?,
        mime_type: r.get(5)?,
        size_bytes: r.get::<_, i64>(6)? as u64,
        description: r.get(7)?,
        created_at: r.get(8)?,
    })
}