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, BillId, Currency, CustomerId, DocumentLine,
    EntryId, EntryStatus, InvoiceId, Ledger, LedgerError, Money, NewAsset, PeriodId, Side,
    VendorId, arap, 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)
    }

    // ------------------------------------------------------------------
    // Accounting periods
    // ------------------------------------------------------------------

    /// Create an accounting period. Journaled as `period.created`.
    pub fn create_period(
        &mut self,
        name: &str,
        start: NaiveDate,
        end: NaiveDate,
    ) -> Result<PeriodId, StoreError> {
        // Pure pre-validation so the journal write cannot be followed by a
        // failed mutation.
        ledger_core::periods::validate_period_range(start, end)?;
        let id = self.ledger.next_period_id();
        let line = self.append_journal(
            "period.created",
            json!({
                "id": id,
                "name": name,
                "start": start.to_string(),
                "end": end.to_string(),
            }),
        )?;
        let live = self.ledger.create_period(name, start, end)?;
        debug_assert_eq!(live, id, "id allocator diverged from journal");
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(live)
    }

    /// Close an accounting period, auto-posting the closing entry (revenue
    /// and expense accounts zeroed into Retained Earnings). Journaled as
    /// `period.closed` with the closing entry's id and its full detail
    /// (date, description, lines).
    ///
    /// The closing entry is part of the close: it is NOT separately
    /// journaled as entry.created/entry.posted. Its detail rides in the
    /// period.closed payload so the SQLite projection can record it; the
    /// detail only exists after the close, so unlike most mutations this
    /// journals after mutating (the pre-close validation makes a failed
    /// close impossible, so the journal cannot be left behind). Replay
    /// re-runs the deterministic close logic and checks the closing entry
    /// id against the journal.
    pub fn close_period(&mut self, id: PeriodId) -> Result<Option<EntryId>, StoreError> {
        // Pure pre-validation: a doomed close journals nothing, not even
        // the Retained Earnings account creation below.
        self.ledger.preview_close_period(id)?;
        // Retained Earnings must exist before the close so its creation (if
        // needed) is journaled as account.added; the close's own ensure
        // then no-ops and replay stays deterministic.
        self.ensure_control_account("3999", "Retained Earnings", AccountType::Equity)?;
        let closing_entry = self.ledger.close_period(id)?;
        let entry_detail = match closing_entry {
            Some(eid) => {
                let e = self
                    .ledger
                    .get_entry(eid)
                    .expect("close just posted the closing entry");
                Some(json!({
                    "id": eid,
                    "date": e.date().to_string(),
                    "description": e.description(),
                    "lines": e.lines().iter().map(journal_line_json).collect::<Vec<_>>(),
                }))
            }
            None => None,
        };
        let line = self.append_journal(
            "period.closed",
            json!({
                "id": id,
                "closing_entry": closing_entry,
                "entry": entry_detail,
            }),
        )?;
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(closing_entry)
    }

    // ------------------------------------------------------------------
    // Bank reconciliation
    // ------------------------------------------------------------------

    /// Mark a posted entry's line cleared (or uncleared) for bank
    /// reconciliation. Journaled as `line.cleared`.
    pub fn mark_cleared(
        &mut self,
        entry_id: EntryId,
        line_index: usize,
        cleared: bool,
    ) -> Result<(), StoreError> {
        self.ledger.mark_cleared(entry_id, line_index, cleared)?;
        let line = self.append_journal(
            "line.cleared",
            json!({
                "entry_id": entry_id,
                "line_index": line_index as u64,
                "cleared": cleared,
            }),
        )?;
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(())
    }

    // ------------------------------------------------------------------
    // Accounts receivable / payable
    // ------------------------------------------------------------------

    /// Fetch a control account by code, creating it through the journaled
    /// account path when missing. The ledger's internal ensure_account
    /// then no-ops, which keeps replay deterministic: every account
    /// creation is an `account.added` event.
    fn ensure_control_account(
        &mut self,
        code: &str,
        name: &str,
        account_type: AccountType,
    ) -> Result<AccountId, StoreError> {
        if let Some(a) = self.ledger.chart().find_by_code(code) {
            return Ok(a.id);
        }
        self.add_account(code, name, account_type, None, false)
    }

    /// Journal an auto-posted two-line entry (debit one account, credit
    /// another for the same amount) through the standard entry.created /
    /// entry.line_added / entry.posted flow. Returns the predicted entry id
    /// and the journaled lines; the caller mutates the ledger afterwards
    /// and applies the lines to SQLite.
    fn journal_auto_entry(
        &mut self,
        date: NaiveDate,
        description: &str,
        debit_account: AccountId,
        credit_account: AccountId,
        amount: Decimal,
    ) -> Result<(EntryId, Vec<JournalLine>), StoreError> {
        let entry_id = self.ledger.next_entry_id();
        let base = self.base_currency;
        let mut journaled = Vec::with_capacity(4);
        journaled.push(self.append_journal(
            "entry.created",
            json!({
                "id": entry_id,
                "date": date.to_string(),
                "description": description,
            }),
        )?);
        for (account, side) in [(debit_account, Side::Debit), (credit_account, Side::Credit)] {
            journaled.push(self.append_journal(
                "entry.line_added",
                json!({
                    "id": entry_id,
                    "account": account,
                    "side": match side {
                        Side::Debit => "debit",
                        Side::Credit => "credit",
                    },
                    "amount": amount.to_string(),
                    "currency": base.code(),
                }),
            )?);
        }
        journaled.push(self.append_journal(
            "entry.posted",
            json!({
                "id": entry_id,
                "date": date.to_string(),
                "description": description,
                "reversal_of": Option::<u64>::None,
                "lines": [
                    plain_line_json(debit_account, Side::Debit, Money::new(amount, base)),
                    plain_line_json(credit_account, Side::Credit, Money::new(amount, base)),
                ],
            }),
        )?);
        Ok((entry_id, journaled))
    }

    /// Add a customer. Journaled as `customer.added`.
    pub fn add_customer(&mut self, name: &str) -> Result<CustomerId, StoreError> {
        // The engine's only failure mode is an empty name; replicate it so
        // the journal write cannot be followed by a failed mutation.
        if name.trim().is_empty() {
            return Err(LedgerError::EmptyAssetDescription.into());
        }
        let id = self.ledger.next_customer_id();
        let line = self.append_journal("customer.added", json!({ "id": id, "name": name }))?;
        let live = self.ledger.add_customer(name)?;
        debug_assert_eq!(live, id, "id allocator diverged from journal");
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(live)
    }

    /// Add a vendor. Journaled as `vendor.added`.
    pub fn add_vendor(&mut self, name: &str) -> Result<VendorId, StoreError> {
        if name.trim().is_empty() {
            return Err(LedgerError::EmptyAssetDescription.into());
        }
        let id = self.ledger.next_vendor_id();
        let line = self.append_journal("vendor.added", json!({ "id": id, "name": name }))?;
        let live = self.ledger.add_vendor(name)?;
        debug_assert_eq!(live, id, "id allocator diverged from journal");
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(live)
    }

    /// Create a sales invoice. The auto-posted entry (Dr A/R, Cr Revenue)
    /// is journaled first through the standard entry flow, then the
    /// invoice itself as `invoice.created`.
    pub fn create_invoice(
        &mut self,
        customer_id: CustomerId,
        date: NaiveDate,
        due_date: NaiveDate,
        lines: Vec<DocumentLine>,
    ) -> Result<InvoiceId, StoreError> {
        // Pure pre-validation mirroring Ledger::create_invoice, so the
        // journal writes below cannot be followed by a failed mutation.
        // (The auto-posted entry balances by construction and its accounts
        // are ensured below.)
        if self.ledger.get_customer(customer_id).is_none() {
            return Err(LedgerError::UnknownCustomer(customer_id).into());
        }
        arap::validate_lines(&lines)?;
        arap::validate_due_date(date, due_date)?;
        if let Some(pid) = self.ledger.closed_period_containing(date) {
            return Err(LedgerError::ClosedPeriod(pid, date.to_string()).into());
        }
        let ar = self.ensure_control_account("1200", "Accounts Receivable", AccountType::Asset)?;
        let revenue = self.ensure_control_account("4000", "Revenue", AccountType::Revenue)?;
        let invoice_id = self.ledger.next_invoice_id();
        let total: Decimal = lines.iter().map(|l| l.amount).sum();
        let description = format!("Invoice {invoice_id}");
        let (entry_id, mut journaled) =
            self.journal_auto_entry(date, &description, ar, revenue, total)?;
        let doc_lines: Vec<Value> = lines.iter().map(document_line_json).collect();
        journaled.push(self.append_journal(
            "invoice.created",
            json!({
                "id": invoice_id,
                "customer_id": customer_id,
                "date": date.to_string(),
                "due_date": due_date.to_string(),
                "lines": doc_lines,
                "entry_id": entry_id,
            }),
        )?);
        let live = self
            .ledger
            .create_invoice(customer_id, date, due_date, lines)?;
        debug_assert_eq!(live, invoice_id, "id allocator diverged from journal");
        debug_assert_eq!(
            self.ledger
                .get_invoice(live)
                .expect("invoice exists")
                .entry_id,
            entry_id,
            "invoice entry diverged from journal"
        );
        for line in &journaled {
            apply_line_to_sqlite(&self.conn, line)?;
        }
        Ok(live)
    }

    /// Create a vendor bill. The auto-posted entry (Dr General Expenses, Cr
    /// A/P) is journaled first through the standard entry flow, then the
    /// bill itself as `bill.created`.
    pub fn create_bill(
        &mut self,
        vendor_id: VendorId,
        date: NaiveDate,
        due_date: NaiveDate,
        lines: Vec<DocumentLine>,
    ) -> Result<BillId, StoreError> {
        if self.ledger.get_vendor(vendor_id).is_none() {
            return Err(LedgerError::UnknownVendor(vendor_id).into());
        }
        arap::validate_lines(&lines)?;
        arap::validate_due_date(date, due_date)?;
        if let Some(pid) = self.ledger.closed_period_containing(date) {
            return Err(LedgerError::ClosedPeriod(pid, date.to_string()).into());
        }
        let expense =
            self.ensure_control_account("6999", "General Expenses", AccountType::Expense)?;
        let ap = self.ensure_control_account("2000", "Accounts Payable", AccountType::Liability)?;
        let bill_id = self.ledger.next_bill_id();
        let total: Decimal = lines.iter().map(|l| l.amount).sum();
        let description = format!("Bill {bill_id}");
        let (entry_id, mut journaled) =
            self.journal_auto_entry(date, &description, expense, ap, total)?;
        let doc_lines: Vec<Value> = lines.iter().map(document_line_json).collect();
        journaled.push(self.append_journal(
            "bill.created",
            json!({
                "id": bill_id,
                "vendor_id": vendor_id,
                "date": date.to_string(),
                "due_date": due_date.to_string(),
                "lines": doc_lines,
                "entry_id": entry_id,
            }),
        )?);
        let live = self.ledger.create_bill(vendor_id, date, due_date, lines)?;
        debug_assert_eq!(live, bill_id, "id allocator diverged from journal");
        debug_assert_eq!(
            self.ledger.get_bill(live).expect("bill exists").entry_id,
            entry_id,
            "bill entry diverged from journal"
        );
        for line in &journaled {
            apply_line_to_sqlite(&self.conn, line)?;
        }
        Ok(live)
    }

    /// Record a customer payment against an invoice. The payment entry (Dr
    /// cash, Cr A/R) is journaled through the standard entry flow first,
    /// then `invoice.payment_received`.
    pub fn receive_payment(
        &mut self,
        invoice_id: InvoiceId,
        date: NaiveDate,
        amount: Decimal,
        cash_account: AccountId,
    ) -> Result<EntryId, StoreError> {
        // Pure pre-validation mirroring Ledger::receive_payment, plus the
        // account checks post_entry would perform: a failed post after the
        // journal writes would leave the journal ahead of the ledger.
        if amount <= Decimal::ZERO {
            return Err(LedgerError::InvalidPaymentAmount(amount).into());
        }
        let balance = self
            .ledger
            .get_invoice(invoice_id)
            .ok_or(LedgerError::UnknownInvoice(invoice_id))?
            .balance();
        if amount > balance {
            return Err(LedgerError::Overpayment(amount, balance).into());
        }
        if let Some(pid) = self.ledger.closed_period_containing(date) {
            return Err(LedgerError::ClosedPeriod(pid, date.to_string()).into());
        }
        let cash = self
            .ledger
            .chart()
            .get(cash_account)
            .ok_or(LedgerError::UnknownAccount(cash_account))?;
        if cash.placeholder {
            return Err(LedgerError::PlaceholderAccount(cash.code.clone()).into());
        }
        let ar = self.ensure_control_account("1200", "Accounts Receivable", AccountType::Asset)?;
        let description = format!("Payment on invoice {invoice_id}");
        let (entry_id, mut journaled) =
            self.journal_auto_entry(date, &description, cash_account, ar, amount)?;
        journaled.push(self.append_journal(
            "invoice.payment_received",
            json!({
                "invoice_id": invoice_id,
                "date": date.to_string(),
                "amount": amount.to_string(),
                "cash_account": cash_account,
                "entry_id": entry_id,
            }),
        )?);
        let live = self
            .ledger
            .receive_payment(invoice_id, date, amount, cash_account)?;
        debug_assert_eq!(live, entry_id, "id allocator diverged from journal");
        for line in &journaled {
            apply_line_to_sqlite(&self.conn, line)?;
        }
        Ok(live)
    }

    /// Pay a vendor bill. The payment entry (Dr A/P, Cr cash) is journaled
    /// through the standard entry flow first, then `bill.paid`.
    pub fn pay_bill(
        &mut self,
        bill_id: BillId,
        date: NaiveDate,
        amount: Decimal,
        cash_account: AccountId,
    ) -> Result<EntryId, StoreError> {
        if amount <= Decimal::ZERO {
            return Err(LedgerError::InvalidPaymentAmount(amount).into());
        }
        let balance = self
            .ledger
            .get_bill(bill_id)
            .ok_or(LedgerError::UnknownBill(bill_id))?
            .balance();
        if amount > balance {
            return Err(LedgerError::Overpayment(amount, balance).into());
        }
        if let Some(pid) = self.ledger.closed_period_containing(date) {
            return Err(LedgerError::ClosedPeriod(pid, date.to_string()).into());
        }
        let cash = self
            .ledger
            .chart()
            .get(cash_account)
            .ok_or(LedgerError::UnknownAccount(cash_account))?;
        if cash.placeholder {
            return Err(LedgerError::PlaceholderAccount(cash.code.clone()).into());
        }
        let ap = self.ensure_control_account("2000", "Accounts Payable", AccountType::Liability)?;
        let description = format!("Payment on bill {bill_id}");
        let (entry_id, mut journaled) =
            self.journal_auto_entry(date, &description, ap, cash_account, amount)?;
        journaled.push(self.append_journal(
            "bill.paid",
            json!({
                "bill_id": bill_id,
                "date": date.to_string(),
                "amount": amount.to_string(),
                "cash_account": cash_account,
                "entry_id": entry_id,
            }),
        )?);
        let live = self.ledger.pay_bill(bill_id, date, amount, cash_account)?;
        debug_assert_eq!(live, entry_id, "id allocator diverged from journal");
        for line in &journaled {
            apply_line_to_sqlite(&self.conn, line)?;
        }
        Ok(live)
    }

    // ------------------------------------------------------------------
    // Foreign exchange
    // ------------------------------------------------------------------

    /// Record a historical FX rate (foreign -> base) for reference.
    /// Journaled as `fx.rate_set`.
    pub fn set_fx_rate(
        &mut self,
        currency: Currency,
        date: NaiveDate,
        rate: Decimal,
    ) -> Result<(), StoreError> {
        self.ledger.set_fx_rate(currency, date, rate)?;
        let line = self.append_journal(
            "fx.rate_set",
            json!({
                "currency": currency.code(),
                "date": date.to_string(),
                "rate": rate.to_string(),
            }),
        )?;
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(())
    }

    /// Add a foreign-currency line to a draft entry. The base-currency
    /// amount is stored alongside the foreign original and the rate used.
    /// Journaled as `entry.line_added` with the fx fields
    /// (`foreign_currency`, `foreign_amount`, `fx_rate`).
    pub fn add_line_fx(
        &mut self,
        entry_id: EntryId,
        account: AccountId,
        side: Side,
        foreign_currency: Currency,
        foreign_amount: Decimal,
        fx_rate: Decimal,
    ) -> Result<(), StoreError> {
        self.ledger.add_line_fx(
            entry_id,
            account,
            side,
            foreign_currency,
            foreign_amount,
            fx_rate,
        )?;
        // Read back the stored base amount rather than recomputing it: the
        // journal must carry exactly what the ledger holds.
        let stored = self
            .ledger
            .get_entry(entry_id)
            .expect("entry exists after add_line_fx")
            .lines()
            .last()
            .expect("line was just added")
            .amount;
        let line = self.append_journal(
            "entry.line_added",
            json!({
                "id": entry_id,
                "account": account,
                "side": match side {
                    Side::Debit => "debit",
                    Side::Credit => "credit",
                },
                "amount": stored.amount.to_string(),
                "currency": stored.currency.code(),
                "foreign_currency": foreign_currency.code(),
                "foreign_amount": foreign_amount.to_string(),
                "fx_rate": fx_rate.to_string(),
            }),
        )?;
        apply_line_to_sqlite(&self.conn, &line)?;
        Ok(())
    }

    // ------------------------------------------------------------------
    // 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",
    };
    let mut v = json!({
        "account": l.account,
        "side": side,
        "amount": l.amount.amount.to_string(),
        "currency": l.amount.currency.code(),
    });
    // Foreign-currency provenance, when the line carries it. Journals
    // written before foreign-currency lines existed have no such lines;
    // readers treat missing fields as None.
    let obj = v
        .as_object_mut()
        .expect("journal_line_json builds an object");
    if let Some(fc) = l.foreign_currency {
        obj.insert(
            "foreign_currency".to_string(),
            Value::String(fc.code().to_string()),
        );
    }
    if let Some(fa) = l.foreign_amount {
        obj.insert("foreign_amount".to_string(), Value::String(fa.to_string()));
    }
    v
}

/// A plain base-currency line, serialized through the one line-JSON path.
fn plain_line_json(account: AccountId, side: Side, amount: Money) -> Value {
    journal_line_json(&ledger_core::JournalLine {
        account,
        side,
        amount,
        foreign_currency: None,
        foreign_amount: None,
    })
}

/// One invoice/bill document line as journal JSON. Amounts are decimal
/// strings, never floats.
fn document_line_json(l: &DocumentLine) -> Value {
    json!({
        "description": l.description,
        "amount": l.amount.to_string(),
    })
}

/// Parse the `lines` array of an invoice/bill journal payload back into
/// document lines.
fn parse_document_lines(p: &Value) -> Result<Vec<DocumentLine>, StoreError> {
    let arr = p
        .get("lines")
        .and_then(|v| v.as_array())
        .ok_or_else(|| StoreError::JournalCorrupt("missing 'lines'".into()))?;
    arr.iter()
        .map(|l| {
            Ok(DocumentLine {
                description: get_str(l, "description")?.to_string(),
                amount: get_str(l, "amount")?
                    .parse()
                    .map_err(|_| StoreError::JournalCorrupt("bad line amount".into()))?,
            })
        })
        .collect()
}

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")?;
            let account = get_u64(p, "account")?;
            let side = parse_side(get_str(p, "side")?)?;
            let amount: Decimal = get_str(p, "amount")?
                .parse()
                .map_err(|_| StoreError::JournalCorrupt("bad amount".into()))?;
            let currency = Currency::new(get_str(p, "currency")?)
                .map_err(|_| StoreError::JournalCorrupt("bad currency".into()))?;
            // The fx fields are optional: journals written before
            // foreign-currency lines existed carry none, and replay as
            // plain base-currency lines.
            match p.get("foreign_currency").and_then(|v| v.as_str()) {
                Some(fc) => {
                    let foreign_amount: Decimal = get_str(p, "foreign_amount")?
                        .parse()
                        .map_err(|_| StoreError::JournalCorrupt("bad foreign amount".into()))?;
                    let fx_rate: Decimal = get_str(p, "fx_rate")?
                        .parse()
                        .map_err(|_| StoreError::JournalCorrupt("bad fx rate".into()))?;
                    ledger.add_line_fx(
                        id,
                        account,
                        side,
                        Currency::new(fc).map_err(|_| {
                            StoreError::JournalCorrupt("bad foreign currency".into())
                        })?,
                        foreign_amount,
                        fx_rate,
                    )?;
                }
                None => {
                    ledger.add_line(id, account, side, Money::new(amount, 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")?,
            )?;
        }
        "period.created" => {
            let want = get_u64(p, "id")?;
            let got = ledger.create_period(
                get_str(p, "name")?,
                parse_date(get_str(p, "start")?)?,
                parse_date(get_str(p, "end")?)?,
            )?;
            if got != want {
                return Err(StoreError::ReplayMismatch(format!(
                    "period id: journal says {want}, replay got {got}"
                )));
            }
        }
        "period.closed" => {
            let id = get_u64(p, "id")?;
            let want = p.get("closing_entry").and_then(|v| v.as_u64());
            // Deterministic: the close re-derives the closing entry from
            // the replayed state, so it must match the journaled id. The
            // journaled entry detail is for the projection; the ledger
            // re-derives the entry itself.
            let got = ledger.close_period(id)?;
            if got != want {
                return Err(StoreError::ReplayMismatch(format!(
                    "closing entry: journal says {want:?}, replay got {got:?}"
                )));
            }
        }
        "line.cleared" => {
            ledger.mark_cleared(
                get_u64(p, "entry_id")?,
                get_u64(p, "line_index")? as usize,
                p.get("cleared").and_then(|v| v.as_bool()).unwrap_or(false),
            )?;
        }
        "customer.added" => {
            let want = get_u64(p, "id")?;
            let got = ledger.add_customer(get_str(p, "name")?)?;
            if got != want {
                return Err(StoreError::ReplayMismatch(format!(
                    "customer id: journal says {want}, replay got {got}"
                )));
            }
        }
        "vendor.added" => {
            let want = get_u64(p, "id")?;
            let got = ledger.add_vendor(get_str(p, "name")?)?;
            if got != want {
                return Err(StoreError::ReplayMismatch(format!(
                    "vendor id: journal says {want}, replay got {got}"
                )));
            }
        }
        "invoice.created" => {
            // The entry was already rebuilt by the standard entry handlers;
            // this only restores the invoice record, without re-posting.
            ledger.replay_invoice(
                get_u64(p, "id")?,
                get_u64(p, "customer_id")?,
                parse_date(get_str(p, "date")?)?,
                parse_date(get_str(p, "due_date")?)?,
                parse_document_lines(p)?,
                get_u64(p, "entry_id")?,
            )?;
        }
        "bill.created" => {
            ledger.replay_bill(
                get_u64(p, "id")?,
                get_u64(p, "vendor_id")?,
                parse_date(get_str(p, "date")?)?,
                parse_date(get_str(p, "due_date")?)?,
                parse_document_lines(p)?,
                get_u64(p, "entry_id")?,
            )?;
        }
        "invoice.payment_received" => {
            // The payment entry was already rebuilt; this only moves the
            // invoice's paid amount and status.
            ledger.replay_invoice_payment(
                get_u64(p, "invoice_id")?,
                get_str(p, "amount")?
                    .parse()
                    .map_err(|_| StoreError::JournalCorrupt("bad amount".into()))?,
                get_u64(p, "entry_id")?,
            )?;
        }
        "bill.paid" => {
            ledger.replay_bill_payment(
                get_u64(p, "bill_id")?,
                get_str(p, "amount")?
                    .parse()
                    .map_err(|_| StoreError::JournalCorrupt("bad amount".into()))?,
                get_u64(p, "entry_id")?,
            )?;
        }
        "fx.rate_set" => {
            ledger.set_fx_rate(
                Currency::new(get_str(p, "currency")?)
                    .map_err(|_| StoreError::JournalCorrupt("bad currency".into()))?,
                parse_date(get_str(p, "date")?)?,
                get_str(p, "rate")?
                    .parse()
                    .map_err(|_| StoreError::JournalCorrupt("bad rate".into()))?,
            )?;
        }
        other => {
            return Err(StoreError::JournalCorrupt(format!(
                "unknown line type: {other}"
            )));
        }
    }
    Ok(())
}

/// Apply a payment to an invoice or bill row in the SQLite projection:
/// bump `amount_paid` and re-derive the status. Amounts are TEXT in
/// SQLite, so the math happens here in Rust with exact Decimals.
/// `table` is `invoices` or `bills`; an internal constant, never user input.
fn apply_document_payment(
    conn: &Connection,
    table: &str,
    id: u64,
    amount: Decimal,
) -> Result<(), StoreError> {
    let (paid_str, lines_json): (String, String) = conn.query_row(
        &format!("SELECT amount_paid, lines_json FROM {table} WHERE id = ?"),
        params![id as i64],
        |r| Ok((r.get(0)?, r.get(1)?)),
    )?;
    let paid: Decimal = paid_str
        .parse()
        .map_err(|_| StoreError::JournalCorrupt(format!("bad amount_paid {paid_str}")))?;
    let new_paid = paid + amount;
    let lines: Vec<Value> = serde_json::from_str(&lines_json)
        .map_err(|_| StoreError::JournalCorrupt("bad lines_json".into()))?;
    let mut total = Decimal::ZERO;
    for l in &lines {
        total += get_str(l, "amount")?
            .parse::<Decimal>()
            .map_err(|_| StoreError::JournalCorrupt("bad line amount".into()))?;
    }
    let status = if new_paid.is_zero() {
        "open"
    } else if new_paid >= total {
        "paid"
    } else {
        "partial"
    };
    conn.execute(
        &format!("UPDATE {table} SET amount_paid = ?, status = ? WHERE id = ?"),
        params![new_paid.to_string(), status, id as i64],
    )?;
    Ok(())
}

/// Apply one journal line to the SQLite projection: INSERTs plus the events
/// mirror, then advance `last_applied_seq`. The applier is a pure function
/// of the journal line: INSERTs for new facts, targeted UPDATEs/DELETEs
/// for lifecycle changes (receipt linking, disposal, period close,
/// cleared flags, document payments, fx rate updates).
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, foreign_currency, foreign_amount) 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")?,
                        l.get("foreign_currency").and_then(|v| v.as_str()),
                        l.get("foreign_amount").and_then(|v| v.as_str()),
                    ],
                )?;
            }
        }
        "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, foreign_currency, foreign_amount) 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")?,
                        l.get("foreign_currency").and_then(|v| v.as_str()),
                        l.get("foreign_amount").and_then(|v| v.as_str()),
                    ],
                )?;
            }
        }
        "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,
                ],
            )?;
        }
        "period.created" => {
            conn.execute(
                "INSERT INTO periods(id, name, start, end, status, closing_entry) VALUES (?,?,?,?,?,?)",
                params![
                    get_u64(p, "id")? as i64,
                    get_str(p, "name")?,
                    get_str(p, "start")?,
                    get_str(p, "end")?,
                    "open",
                    Option::<i64>::None,
                ],
            )?;
        }
        "period.closed" => {
            // The closing entry rides in the payload (it is not separately
            // journaled as entry.created/entry.posted): record it like any
            // posted entry so the projection stays complete, then mark the
            // period row closed.
            if !p.get("entry").map(|v| v.is_null()).unwrap_or(true) {
                let e = &p["entry"];
                let eid = get_u64(e, "id")? as i64;
                conn.execute(
                    "INSERT INTO entries(id, date, description, status, reversal_of) VALUES (?,?,?,'posted',NULL)",
                    params![eid, get_str(e, "date")?, get_str(e, "description")?],
                )?;
                for l in e.get("lines").and_then(|v| v.as_array()).ok_or_else(|| {
                    StoreError::JournalCorrupt("period.closed entry without lines".into())
                })? {
                    conn.execute(
                        "INSERT INTO lines(entry_id, account_id, side, amount, currency, foreign_currency, foreign_amount) VALUES (?,?,?,?,?,?,?)",
                        params![
                            eid,
                            get_u64(l, "account")? as i64,
                            get_str(l, "side")?,
                            get_str(l, "amount")?,
                            get_str(l, "currency")?,
                            l.get("foreign_currency").and_then(|v| v.as_str()),
                            l.get("foreign_amount").and_then(|v| v.as_str()),
                        ],
                    )?;
                }
            }
            conn.execute(
                "UPDATE periods SET status = 'closed', closing_entry = ? WHERE id = ?",
                params![
                    p.get("closing_entry")
                        .and_then(|v| v.as_u64())
                        .map(|v| v as i64),
                    get_u64(p, "id")? as i64,
                ],
            )?;
        }
        "line.cleared" => {
            let entry_id = get_u64(p, "entry_id")? as i64;
            let line_index = get_u64(p, "line_index")? as i64;
            if p.get("cleared").and_then(|v| v.as_bool()).unwrap_or(false) {
                conn.execute(
                    "INSERT OR IGNORE INTO cleared_lines(entry_id, line_index) VALUES (?, ?)",
                    params![entry_id, line_index],
                )?;
            } else {
                conn.execute(
                    "DELETE FROM cleared_lines WHERE entry_id = ? AND line_index = ?",
                    params![entry_id, line_index],
                )?;
            }
        }
        "customer.added" => {
            conn.execute(
                "INSERT INTO customers(id, name) VALUES (?, ?)",
                params![get_u64(p, "id")? as i64, get_str(p, "name")?],
            )?;
        }
        "vendor.added" => {
            conn.execute(
                "INSERT INTO vendors(id, name) VALUES (?, ?)",
                params![get_u64(p, "id")? as i64, get_str(p, "name")?],
            )?;
        }
        "invoice.created" => {
            let lines = p.get("lines").ok_or_else(|| {
                StoreError::JournalCorrupt("invoice.created without lines".into())
            })?;
            conn.execute(
                "INSERT INTO invoices(id, customer_id, date, due_date, lines_json, amount_paid, status, entry_id) VALUES (?,?,?,?,?,?,?,?)",
                params![
                    get_u64(p, "id")? as i64,
                    get_u64(p, "customer_id")? as i64,
                    get_str(p, "date")?,
                    get_str(p, "due_date")?,
                    lines.to_string(),
                    "0",
                    "open",
                    get_u64(p, "entry_id")? as i64,
                ],
            )?;
        }
        "bill.created" => {
            let lines = p
                .get("lines")
                .ok_or_else(|| StoreError::JournalCorrupt("bill.created without lines".into()))?;
            conn.execute(
                "INSERT INTO bills(id, vendor_id, date, due_date, lines_json, amount_paid, status, entry_id) VALUES (?,?,?,?,?,?,?,?)",
                params![
                    get_u64(p, "id")? as i64,
                    get_u64(p, "vendor_id")? as i64,
                    get_str(p, "date")?,
                    get_str(p, "due_date")?,
                    lines.to_string(),
                    "0",
                    "open",
                    get_u64(p, "entry_id")? as i64,
                ],
            )?;
        }
        "invoice.payment_received" => {
            let amount: Decimal = get_str(p, "amount")?
                .parse()
                .map_err(|_| StoreError::JournalCorrupt("bad amount".into()))?;
            apply_document_payment(conn, "invoices", get_u64(p, "invoice_id")?, amount)?;
        }
        "bill.paid" => {
            let amount: Decimal = get_str(p, "amount")?
                .parse()
                .map_err(|_| StoreError::JournalCorrupt("bad amount".into()))?;
            apply_document_payment(conn, "bills", get_u64(p, "bill_id")?, amount)?;
        }
        "fx.rate_set" => {
            conn.execute(
                "INSERT INTO fx_rates(currency, date, rate) VALUES (?,?,?) \
                 ON CONFLICT(currency, date) DO UPDATE SET rate=excluded.rate",
                params![
                    get_str(p, "currency")?,
                    get_str(p, "date")?,
                    get_str(p, "rate")?,
                ],
            )?;
        }
        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)?,
    })
}