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};
const RATE_SCALE: u32 = 10;
#[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()))?;
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) => {
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(), 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());
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()))?;
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)?;
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)?;
}
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> {
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()
}
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());
let rate = input
.rate
.round_dp_with_strategy(RATE_SCALE, RoundingStrategy::MidpointAwayFromZero);
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)?;
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)?;
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(¶ms);
let affected = tx.execute(&query, params_ref.as_slice()).map_err(map_db_error)?;
if affected != ids.len() {
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);
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);
}
}