stateset_db/sqlite/

currency.rs

//! SQLite implementation of currency repository

use chrono::Utc;
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
use rusqlite::params;
use rust_decimal::{Decimal, RoundingStrategy};
use stateset_core::{
    CommerceError, ConversionResult, ConvertCurrency, Currency, ExchangeRate, ExchangeRateFilter,
    Result, SetExchangeRate, StoreCurrencySettings,
};
use uuid::Uuid;

use super::{
    build_in_clause, map_db_error, params_refs, parse_datetime_row, parse_decimal_row,
    parse_enum_row, parse_json_row, parse_uuid_row, uuid_params,
};
use stateset_core::{BatchResult, validate_batch_size};

/// Fractional digits of precision for stored exchange rates.
///
/// Postgres stores `exchange_rates.rate` as `DECIMAL(20, 10)`, which rounds any
/// higher-precision rate to 10 dp (half away from zero). SQLite stores the rate
/// as TEXT with no such constraint, so we round explicitly to keep conversions
/// identical across backends.
const RATE_SCALE: u32 = 10;

/// SQLite currency repository
#[derive(Debug)]
pub struct SqliteCurrencyRepository {
    pool: Pool<SqliteConnectionManager>,
}

impl SqliteCurrencyRepository {
    #[must_use]
    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
        Self { pool }
    }

    fn row_to_exchange_rate(row: &rusqlite::Row<'_>) -> rusqlite::Result<ExchangeRate> {
        Ok(ExchangeRate {
            id: parse_uuid_row(&row.get::<_, String>("id")?, "exchange_rate", "id")?,
            base_currency: parse_enum_row(
                &row.get::<_, String>("base_currency")?,
                "exchange_rate",
                "base_currency",
            )?,
            quote_currency: parse_enum_row(
                &row.get::<_, String>("quote_currency")?,
                "exchange_rate",
                "quote_currency",
            )?,
            rate: parse_decimal_row(&row.get::<_, String>("rate")?, "exchange_rate", "rate")?,
            source: row.get("source")?,
            rate_at: parse_datetime_row(
                &row.get::<_, String>("rate_at")?,
                "exchange_rate",
                "rate_at",
            )?,
            created_at: parse_datetime_row(
                &row.get::<_, String>("created_at")?,
                "exchange_rate",
                "created_at",
            )?,
            updated_at: parse_datetime_row(
                &row.get::<_, String>("updated_at")?,
                "exchange_rate",
                "updated_at",
            )?,
        })
    }
}

impl stateset_core::CurrencyRepository for SqliteCurrencyRepository {
    fn get_rate(&self, from: Currency, to: Currency) -> Result<Option<ExchangeRate>> {
        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;

        // Same currency = rate of 1
        if from == to {
            return Ok(Some(ExchangeRate {
                id: Uuid::nil(),
                base_currency: from,
                quote_currency: to,
                rate: Decimal::ONE,
                source: "identity".into(),
                rate_at: Utc::now(),
                created_at: Utc::now(),
                updated_at: Utc::now(),
            }));
        }

        let result = conn.query_row(
            "SELECT id, base_currency, quote_currency, rate, source, rate_at, created_at, updated_at
             FROM exchange_rates
             WHERE base_currency = ? AND quote_currency = ?",
            params![from.code(), to.code()],
            Self::row_to_exchange_rate,
        );

        match result {
            Ok(rate) => Ok(Some(rate)),
            Err(rusqlite::Error::QueryReturnedNoRows) => {
                // Try to find inverse rate
                let inverse_result = conn.query_row(
                    "SELECT id, base_currency, quote_currency, rate, source, rate_at, created_at, updated_at
                     FROM exchange_rates
                     WHERE base_currency = ? AND quote_currency = ?",
                    params![to.code(), from.code()],
                    |row| {
                        let direct = Self::row_to_exchange_rate(row)?;
                        let inverse_rate = direct.rate;
                        let rate = if inverse_rate.is_zero() {
                            Decimal::ZERO
                        } else {
                            Decimal::ONE / inverse_rate
                        };

                        Ok(ExchangeRate {
                            id: Uuid::new_v4(), // Generated for inverse
                            base_currency: from,
                            quote_currency: to,
                            rate,
                            source: format!("inverse:{}", direct.source),
                            rate_at: direct.rate_at,
                            created_at: Utc::now(),
                            updated_at: Utc::now(),
                        })
                    },
                );

                match inverse_result {
                    Ok(rate) => Ok(Some(rate)),
                    Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
                    Err(e) => Err(map_db_error(e)),
                }
            }
            Err(e) => Err(map_db_error(e)),
        }
    }

    fn get_rates_for(&self, base: Currency) -> Result<Vec<ExchangeRate>> {
        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;

        let mut stmt = conn
            .prepare(
                "SELECT id, base_currency, quote_currency, rate, source, rate_at, created_at, updated_at
                 FROM exchange_rates
                 WHERE base_currency = ?
                 ORDER BY quote_currency",
            )
            .map_err(map_db_error)?;

        let rows = stmt
            .query_map(params![base.code()], Self::row_to_exchange_rate)
            .map_err(map_db_error)?;

        rows.collect::<std::result::Result<Vec<_>, _>>().map_err(map_db_error)
    }

    fn list_rates(&self, filter: ExchangeRateFilter) -> Result<Vec<ExchangeRate>> {
        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;

        let mut query = String::from(
            "SELECT id, base_currency, quote_currency, rate, source, rate_at, created_at, updated_at
             FROM exchange_rates WHERE 1=1",
        );
        let mut params_vec: Vec<String> = Vec::new();

        if let Some(base) = &filter.base_currency {
            query.push_str(" AND base_currency = ?");
            params_vec.push(base.code().to_string());
        }

        if let Some(quote) = &filter.quote_currency {
            query.push_str(" AND quote_currency = ?");
            params_vec.push(quote.code().to_string());
        }

        if let Some(since) = &filter.since {
            query.push_str(" AND rate_at >= ?");
            params_vec.push(since.to_rfc3339());
        }

        query.push_str(" ORDER BY base_currency, quote_currency");
        crate::sqlite::append_limit_offset(&mut query, filter.limit, filter.offset);

        let mut stmt = conn.prepare(&query).map_err(map_db_error)?;
        let params: Vec<&dyn rusqlite::ToSql> =
            params_vec.iter().map(|s| s as &dyn rusqlite::ToSql).collect();

        let rows =
            stmt.query_map(params.as_slice(), Self::row_to_exchange_rate).map_err(map_db_error)?;

        rows.collect::<std::result::Result<Vec<_>, _>>().map_err(map_db_error)
    }

    fn set_rate(&self, input: SetExchangeRate) -> Result<ExchangeRate> {
        let id = Uuid::new_v4();
        let now = Utc::now();
        let source = input.source.unwrap_or_else(|| "manual".into());

        // Round to the same scale Postgres enforces via `DECIMAL(20, 10)` so a
        // high-precision rate converts identically on both backends.
        let rate =
            input.rate.round_dp_with_strategy(RATE_SCALE, RoundingStrategy::MidpointAwayFromZero);

        {
            let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;

            // Upsert the rate
            conn.execute(
                "INSERT INTO exchange_rates (id, base_currency, quote_currency, rate, source, rate_at, created_at, updated_at)
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                 ON CONFLICT (base_currency, quote_currency) DO UPDATE SET
                    rate = excluded.rate,
                    source = excluded.source,
                    rate_at = excluded.rate_at,
                    updated_at = excluded.updated_at",
                params![
                    id.to_string(),
                    input.base_currency.code(),
                    input.quote_currency.code(),
                    rate.to_string(),
                    source,
                    now.to_rfc3339(),
                    now.to_rfc3339(),
                    now.to_rfc3339()
                ],
            )
            .map_err(map_db_error)?;

            // Record in history
            conn.execute(
                "INSERT INTO exchange_rate_history (id, base_currency, quote_currency, rate, source, rate_at)
                 VALUES (?, ?, ?, ?, ?, ?)",
                params![
                    Uuid::new_v4().to_string(),
                    input.base_currency.code(),
                    input.quote_currency.code(),
                    rate.to_string(),
                    source,
                    now.to_rfc3339()
                ],
            )
            .map_err(map_db_error)?;
        }

        // Fetch and return the rate
        self.get_rate(input.base_currency, input.quote_currency)?.ok_or(CommerceError::NotFound)
    }

    fn set_rates(&self, rates: Vec<SetExchangeRate>) -> Result<Vec<ExchangeRate>> {
        rates.into_iter().map(|r| self.set_rate(r)).collect()
    }

    fn delete_rate(&self, id: Uuid) -> Result<()> {
        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;

        let affected = conn
            .execute("DELETE FROM exchange_rates WHERE id = ?", params![id.to_string()])
            .map_err(map_db_error)?;

        if affected == 0 { Err(CommerceError::NotFound) } else { Ok(()) }
    }

    fn convert(&self, input: ConvertCurrency) -> Result<ConversionResult> {
        // Same currency = no conversion needed
        if input.from == input.to {
            return Ok(ConversionResult {
                original_amount: input.amount,
                original_currency: input.from,
                converted_amount: input.amount,
                target_currency: input.to,
                rate: Decimal::ONE,
                inverse_rate: Decimal::ONE,
                rate_at: Utc::now(),
            });
        }

        let rate = self.get_rate(input.from, input.to)?.ok_or(CommerceError::ValidationError(
            format!("No exchange rate found for {} to {}", input.from, input.to),
        ))?;

        let converted_amount = input.amount * rate.rate;
        let inverse_rate =
            if rate.rate.is_zero() { Decimal::ZERO } else { Decimal::ONE / rate.rate };

        Ok(ConversionResult {
            original_amount: input.amount,
            original_currency: input.from,
            converted_amount,
            target_currency: input.to,
            rate: rate.rate,
            inverse_rate,
            rate_at: rate.rate_at,
        })
    }

    fn get_settings(&self) -> Result<StoreCurrencySettings> {
        let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;

        let result = conn.query_row(
            "SELECT base_currency, enabled_currencies, auto_convert, rounding_mode
             FROM store_currency_settings
             WHERE id = 'default'",
            [],
            |row| {
                let base_currency = parse_enum_row(
                    &row.get::<_, String>(0)?,
                    "store_currency_settings",
                    "base_currency",
                )?;
                let enabled_currencies: Vec<Currency> = parse_json_row(
                    &row.get::<_, String>(1)?,
                    "store_currency_settings",
                    "enabled_currencies",
                )?;
                let auto_convert: bool = row.get::<_, i32>(2)? != 0;
                let rounding_mode = parse_enum_row(
                    &row.get::<_, String>(3)?,
                    "store_currency_settings",
                    "rounding_mode",
                )?;

                Ok(StoreCurrencySettings {
                    base_currency,
                    enabled_currencies,
                    auto_convert,
                    rounding_mode,
                })
            },
        );

        match result {
            Ok(settings) => Ok(settings),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(StoreCurrencySettings::default()),
            Err(e) => Err(map_db_error(e)),
        }
    }

    fn update_settings(&self, settings: StoreCurrencySettings) -> Result<StoreCurrencySettings> {
        let enabled_json = serde_json::to_string(&settings.enabled_currencies)
            .map_err(|e| CommerceError::DatabaseError(e.to_string()))?;

        let rounding_str = settings.rounding_mode.to_string();

        {
            let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;

            conn.execute(
                "INSERT INTO store_currency_settings (id, base_currency, enabled_currencies, auto_convert, rounding_mode, updated_at)
                 VALUES ('default', ?, ?, ?, ?, datetime('now'))
                 ON CONFLICT (id) DO UPDATE SET
                    base_currency = excluded.base_currency,
                    enabled_currencies = excluded.enabled_currencies,
                    auto_convert = excluded.auto_convert,
                    rounding_mode = excluded.rounding_mode,
                    updated_at = excluded.updated_at",
                params![
                    settings.base_currency.code(),
                    enabled_json,
                    i32::from(settings.auto_convert),
                    rounding_str
                ],
            )
            .map_err(map_db_error)?;
        }

        self.get_settings()
    }

    // === Batch Operations ===

    fn set_rates_atomic(&self, rates: Vec<SetExchangeRate>) -> Result<Vec<ExchangeRate>> {
        validate_batch_size(&rates)?;

        if rates.is_empty() {
            return Ok(Vec::new());
        }

        let mut conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;

        let now = Utc::now();
        let mut rate_ids: Vec<(Currency, Currency)> = Vec::with_capacity(rates.len());

        for input in &rates {
            let id = Uuid::new_v4();
            let source = input.source.clone().unwrap_or_else(|| "manual".into());

            // Match Postgres `DECIMAL(20, 10)` precision (see `RATE_SCALE`).
            let rate = input
                .rate
                .round_dp_with_strategy(RATE_SCALE, RoundingStrategy::MidpointAwayFromZero);

            // Upsert the rate
            tx.execute(
                "INSERT INTO exchange_rates (id, base_currency, quote_currency, rate, source, rate_at, created_at, updated_at)
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                 ON CONFLICT (base_currency, quote_currency) DO UPDATE SET
                    rate = excluded.rate,
                    source = excluded.source,
                    rate_at = excluded.rate_at,
                    updated_at = excluded.updated_at",
                params![
                    id.to_string(),
                    input.base_currency.code(),
                    input.quote_currency.code(),
                    rate.to_string(),
                    source,
                    now.to_rfc3339(),
                    now.to_rfc3339(),
                    now.to_rfc3339()
                ],
            )
            .map_err(map_db_error)?;

            // Record in history
            tx.execute(
                "INSERT INTO exchange_rate_history (id, base_currency, quote_currency, rate, source, rate_at)
                 VALUES (?, ?, ?, ?, ?, ?)",
                params![
                    Uuid::new_v4().to_string(),
                    input.base_currency.code(),
                    input.quote_currency.code(),
                    rate.to_string(),
                    source,
                    now.to_rfc3339()
                ],
            )
            .map_err(map_db_error)?;

            rate_ids.push((input.base_currency, input.quote_currency));
        }

        tx.commit().map_err(map_db_error)?;

        // Fetch and return all the rates
        let mut results = Vec::with_capacity(rate_ids.len());
        for (from, to) in rate_ids {
            if let Some(rate) = self.get_rate(from, to)? {
                results.push(rate);
            }
        }

        Ok(results)
    }

    fn delete_rates_batch(&self, ids: Vec<Uuid>) -> Result<BatchResult<Uuid>> {
        validate_batch_size(&ids)?;

        let mut result = BatchResult::with_capacity(ids.len());

        for (index, id) in ids.into_iter().enumerate() {
            match self.delete_rate(id) {
                Ok(()) => result.record_success(id),
                Err(e) => result.record_failure(index, Some(id.to_string()), &e),
            }
        }

        Ok(result)
    }

    fn delete_rates_atomic(&self, ids: Vec<Uuid>) -> Result<()> {
        validate_batch_size(&ids)?;

        if ids.is_empty() {
            return Ok(());
        }

        let mut conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;

        let in_clause = build_in_clause(ids.len());
        let query = format!("DELETE FROM exchange_rates WHERE id IN ({in_clause})");

        let params = uuid_params(&ids);
        let params_ref = params_refs(&params);

        let affected = tx.execute(&query, params_ref.as_slice()).map_err(map_db_error)?;

        if affected != ids.len() {
            // Not all rates were found - rollback by not committing
            return Err(CommerceError::NotFound);
        }

        tx.commit().map_err(map_db_error)?;
        Ok(())
    }

    fn get_rates_batch(&self, pairs: Vec<(Currency, Currency)>) -> Result<Vec<ExchangeRate>> {
        validate_batch_size(&pairs)?;

        let mut results = Vec::with_capacity(pairs.len());

        for (from, to) in pairs {
            if let Some(rate) = self.get_rate(from, to)? {
                results.push(rate);
            }
        }

        Ok(results)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::SqliteDatabase;
    use rust_decimal_macros::dec;
    use stateset_core::{Currency, CurrencyRepository, ExchangeRateFilter};

    fn fresh_repo() -> SqliteCurrencyRepository {
        SqliteDatabase::in_memory().expect("in-memory").currency()
    }

    fn set(repo: &SqliteCurrencyRepository, base: Currency, quote: Currency) {
        repo.set_rate(SetExchangeRate {
            base_currency: base,
            quote_currency: quote,
            rate: dec!(1.5),
            source: None,
        })
        .expect("set rate");
    }

    #[test]
    fn list_rates_applies_limit_and_offset() {
        let repo = fresh_repo();
        set(&repo, Currency::USD, Currency::EUR);
        set(&repo, Currency::USD, Currency::GBP);
        set(&repo, Currency::USD, Currency::JPY);

        // The database pre-seeds exchange rates, so assert relative to the
        // unpaginated result rather than an absolute count.
        let all = repo.list_rates(ExchangeRateFilter::default()).expect("list all");
        assert!(all.len() >= 3, "expected at least the three inserted rates");

        let page = repo
            .list_rates(ExchangeRateFilter { limit: Some(2), ..Default::default() })
            .expect("limited");
        assert_eq!(page.len(), 2, "limit must bound the result set");
        assert_eq!(page[0].id, all[0].id);
        assert_eq!(page[1].id, all[1].id);

        let rest = repo
            .list_rates(ExchangeRateFilter { offset: Some(2), ..Default::default() })
            .expect("offset");
        assert_eq!(rest.len(), all.len() - 2, "offset must skip the first rows");
        assert_eq!(rest[0].id, all[2].id);
    }
}