use super::{
backorder::{cancel_backorders_for_order_in_tx, create_backorder_in_tx},
build_in_clause,
inventory::{ReservationConfirmOutcome, SqliteInventoryRepository},
map_db_error, params_refs, parse_datetime_row, parse_decimal_row, parse_enum, parse_enum_row,
parse_json_opt_row, parse_uuid_row, sum_decimal_query, uuid_params, with_immediate_transaction,
};
use chrono::Utc;
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
use rust_decimal::Decimal;
use stateset_core::{
Address, BatchResult, CommerceError, CreateBackorder, CreateOrder, CreateOrderItem, CustomerId,
FulfillmentStatus, Order, OrderFilter, OrderId, OrderItem, OrderItemId, OrderRepository,
OrderStatus, PaymentStatus, ProductId, ReserveInventory, Result, UpdateOrder,
validate_batch_size, validate_currency_code, validate_postal_code, validate_price,
validate_required_text, validate_required_uuid, validate_sku,
};
use uuid::Uuid;
#[derive(Debug)]
pub struct SqliteOrderRepository {
pool: Pool<SqliteConnectionManager>,
}
impl SqliteOrderRepository {
#[must_use]
pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
Self { pool }
}
fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
}
fn generate_order_number() -> String {
use std::sync::atomic::{AtomicU64, Ordering};
static COUNTER: AtomicU64 = AtomicU64::new(0);
let now = Utc::now();
let timestamp = now.timestamp();
let nanos = now.timestamp_subsec_nanos();
let seq = COUNTER.fetch_add(1, Ordering::Relaxed);
format!("ORD-{}-{:06}-{:08X}", timestamp, nanos / 1000, seq as u32)
}
fn row_to_order(row: &rusqlite::Row<'_>) -> rusqlite::Result<Order> {
let shipping_addr: Option<Address> = parse_json_opt_row(
row.get::<_, Option<String>>("shipping_address")?,
"order",
"shipping_address",
)?;
let billing_addr: Option<Address> = parse_json_opt_row(
row.get::<_, Option<String>>("billing_address")?,
"order",
"billing_address",
)?;
Ok(Order {
id: OrderId::from(parse_uuid_row(&row.get::<_, String>("id")?, "order", "id")?),
order_number: row.get("order_number")?,
customer_id: CustomerId::from(parse_uuid_row(
&row.get::<_, String>("customer_id")?,
"order",
"customer_id",
)?),
status: parse_enum_row(&row.get::<_, String>("status")?, "order", "status")?,
order_date: parse_datetime_row(
&row.get::<_, String>("order_date")?,
"order",
"order_date",
)?,
total_amount: parse_decimal_row(
&row.get::<_, String>("total_amount")?,
"order",
"total_amount",
)?,
currency: row.get("currency")?,
payment_status: parse_enum_row(
&row.get::<_, String>("payment_status")?,
"order",
"payment_status",
)?,
fulfillment_status: parse_enum_row(
&row.get::<_, String>("fulfillment_status")?,
"order",
"fulfillment_status",
)?,
payment_method: row.get("payment_method")?,
shipping_method: row.get("shipping_method")?,
tracking_number: row.get("tracking_number")?,
notes: row.get("notes")?,
shipping_address: shipping_addr,
billing_address: billing_addr,
items: vec![], version: row.get::<_, Option<i32>>("version")?.unwrap_or(1),
created_at: parse_datetime_row(
&row.get::<_, String>("created_at")?,
"order",
"created_at",
)?,
updated_at: parse_datetime_row(
&row.get::<_, String>("updated_at")?,
"order",
"updated_at",
)?,
})
}
fn validate_order_item_input(item: &CreateOrderItem) -> Result<()> {
validate_required_uuid("order_item.product_id", item.product_id.into_uuid())?;
if let Some(variant_id) = item.variant_id {
validate_required_uuid("order_item.variant_id", variant_id)?;
}
validate_sku(&item.sku)?;
validate_required_text("order_item.name", &item.name, 255)?;
if item.quantity <= 0 {
return Err(CommerceError::InvalidInput {
field: "order_item.quantity".to_string(),
message: "must be greater than zero".into(),
});
}
validate_price(item.unit_price)?;
if let Some(discount) = item.discount {
validate_price(discount)?;
}
if let Some(tax) = item.tax_amount {
validate_price(tax)?;
}
let subtotal = item.unit_price * Decimal::from(item.quantity);
let discount = item.discount.unwrap_or_default();
let tax = item.tax_amount.unwrap_or_default();
if discount > subtotal {
return Err(CommerceError::ValidationError(
"Order item discount cannot exceed subtotal".into(),
));
}
let total = subtotal - discount + tax;
if total < Decimal::ZERO {
return Err(CommerceError::ValidationError(
"Order item total cannot be negative".into(),
));
}
Ok(())
}
fn validate_address_input(address: &Address, field_prefix: &str) -> Result<()> {
validate_required_text(&format!("{field_prefix}.line1"), &address.line1, 255)?;
validate_required_text(&format!("{field_prefix}.city"), &address.city, 255)?;
validate_postal_code(&address.postal_code)?;
validate_required_text(&format!("{field_prefix}.country"), &address.country, 64)?;
if let Some(line2) = &address.line2 {
validate_required_text(&format!("{field_prefix}.line2"), line2, 255)?;
}
if let Some(state) = &address.state {
validate_required_text(&format!("{field_prefix}.state"), state, 64)?;
}
Ok(())
}
fn validate_order_input(input: &CreateOrder) -> Result<()> {
validate_required_uuid("order.customer_id", input.customer_id.into_uuid())?;
if let Some(ref currency) = input.currency {
validate_currency_code(currency.as_str())?;
}
if input.items.is_empty() {
return Err(CommerceError::ValidationError("Order must have at least one item".into()));
}
for item in &input.items {
Self::validate_order_item_input(item)?;
}
if let Some(address) = &input.shipping_address {
Self::validate_address_input(address, "order.shipping_address")?;
}
if let Some(address) = &input.billing_address {
Self::validate_address_input(address, "order.billing_address")?;
}
Ok(())
}
fn load_order_items_with_conn(
conn: &rusqlite::Connection,
order_id: OrderId,
) -> Result<Vec<OrderItem>> {
let mut stmt = conn
.prepare(
"SELECT id, order_id, product_id, variant_id, sku, name, quantity,
unit_price, discount, tax_amount, total
FROM order_items WHERE order_id = ?",
)
.map_err(map_db_error)?;
let items = stmt
.query_map([order_id.to_string()], Self::row_to_order_item)
.map_err(map_db_error)?
.collect::<rusqlite::Result<Vec<_>>>()
.map_err(map_db_error)?;
Ok(items)
}
fn row_to_order_item(row: &rusqlite::Row<'_>) -> rusqlite::Result<OrderItem> {
Ok(OrderItem {
id: OrderItemId::from(parse_uuid_row(
&row.get::<_, String>("id")?,
"order_item",
"id",
)?),
order_id: OrderId::from(parse_uuid_row(
&row.get::<_, String>("order_id")?,
"order_item",
"order_id",
)?),
product_id: ProductId::from(parse_uuid_row(
&row.get::<_, String>("product_id")?,
"order_item",
"product_id",
)?),
variant_id: row.get::<_, Option<String>>("variant_id")?.and_then(|s| s.parse().ok()),
sku: row.get("sku")?,
name: row.get("name")?,
quantity: row.get("quantity")?,
unit_price: parse_decimal_row(
&row.get::<_, String>("unit_price")?,
"order_item",
"unit_price",
)?,
discount: parse_decimal_row(
&row.get::<_, String>("discount")?,
"order_item",
"discount",
)?,
tax_amount: parse_decimal_row(
&row.get::<_, String>("tax_amount")?,
"order_item",
"tax_amount",
)?,
total: parse_decimal_row(&row.get::<_, String>("total")?, "order_item", "total")?,
})
}
fn load_order_items_batch(
conn: &rusqlite::Connection,
ids: &[OrderId],
) -> Result<std::collections::HashMap<OrderId, Vec<OrderItem>>> {
let mut map: std::collections::HashMap<OrderId, Vec<OrderItem>> =
std::collections::HashMap::with_capacity(ids.len());
for chunk in ids.chunks(500) {
let placeholders = build_in_clause(chunk.len());
let sql = format!(
"SELECT id, order_id, product_id, variant_id, sku, name, quantity,
unit_price, discount, tax_amount, total
FROM order_items WHERE order_id IN ({placeholders})"
);
let id_strs: Vec<String> = chunk.iter().map(ToString::to_string).collect();
let param_refs: Vec<&dyn rusqlite::ToSql> =
id_strs.iter().map(|s| s as &dyn rusqlite::ToSql).collect();
let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
let rows = stmt
.query_map(param_refs.as_slice(), Self::row_to_order_item)
.map_err(map_db_error)?;
for row in rows {
let item = row.map_err(map_db_error)?;
map.entry(item.order_id).or_default().push(item);
}
}
Ok(map)
}
fn get_by_cart_id_in_conn(
conn: &rusqlite::Connection,
cart_id: Uuid,
) -> std::result::Result<Option<Order>, rusqlite::Error> {
let result = conn.query_row(
"SELECT * FROM orders WHERE cart_id = ?",
[cart_id.to_string()],
Self::row_to_order,
);
match result {
Ok(mut order) => {
order.items = Self::load_order_items_with_conn(conn, order.id)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
Ok(Some(order))
}
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e),
}
}
pub fn get_by_cart_id(&self, cart_id: Uuid) -> Result<Option<Order>> {
let conn = self.conn()?;
Self::get_by_cart_id_in_conn(&conn, cart_id).map_err(map_db_error)
}
fn create_internal_in_tx(
tx: &rusqlite::Transaction<'_>,
cart_id: Option<Uuid>,
idempotent_by_cart_id: bool,
input: &CreateOrder,
) -> std::result::Result<Order, rusqlite::Error> {
let id = OrderId::new();
let order_number = Self::generate_order_number();
let now = Utc::now();
let currency = input.currency.unwrap_or_default();
let id_str = id.to_string();
let customer_id_str = input.customer_id.to_string();
let now_str = now.to_rfc3339();
let total: Decimal = input
.items
.iter()
.map(|item| {
OrderItem::calculate_total(
item.quantity,
item.unit_price,
item.discount.unwrap_or_default(),
item.tax_amount.unwrap_or_default(),
)
})
.sum();
let total_str = total.to_string();
let shipping_address_json = input
.shipping_address
.as_ref()
.map(|address| {
serde_json::to_string(address).map_err(|error| {
rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::DatabaseError(
format!("Failed to serialize order.shipping_address: {error}"),
)))
})
})
.transpose()?;
let billing_address_json = input
.billing_address
.as_ref()
.map(|address| {
serde_json::to_string(address).map_err(|error| {
rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::DatabaseError(
format!("Failed to serialize order.billing_address: {error}"),
)))
})
})
.transpose()?;
let cart_id_str = if idempotent_by_cart_id {
Some(
cart_id
.ok_or_else(|| {
rusqlite::Error::ToSqlConversionFailure(Box::new(
CommerceError::ValidationError(
"cart_id is required for cart checkout".into(),
),
))
})?
.to_string(),
)
} else {
None
};
let inserted = if idempotent_by_cart_id {
let cart_id_str = cart_id_str.as_deref().ok_or_else(|| {
rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::DatabaseError(
"cart_id was required but missing (internal error)".into(),
)))
})?;
let rows_affected = tx.execute(
"INSERT OR IGNORE INTO orders (id, order_number, customer_id, status, order_date, total_amount,
currency, payment_status, fulfillment_status, payment_method,
shipping_method, notes, shipping_address, billing_address,
cart_id, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
&id_str,
&order_number,
&customer_id_str,
"pending",
&now_str,
&total_str,
¤cy,
"pending",
"unfulfilled",
&input.payment_method,
&input.shipping_method,
&input.notes,
&shipping_address_json,
&billing_address_json,
cart_id_str,
&now_str,
&now_str,
],
)?;
rows_affected > 0
} else {
tx.prepare_cached(
"INSERT INTO orders (id, order_number, customer_id, status, order_date, total_amount,
currency, payment_status, fulfillment_status, payment_method,
shipping_method, notes, shipping_address, billing_address,
created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
)?.execute(
rusqlite::params![
&id_str,
&order_number,
&customer_id_str,
"pending",
&now_str,
&total_str,
¤cy,
"pending",
"unfulfilled",
&input.payment_method,
&input.shipping_method,
&input.notes,
&shipping_address_json,
&billing_address_json,
&now_str,
&now_str,
],
)?;
true
};
if !inserted {
let cart_id = cart_id.ok_or_else(|| {
rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::DatabaseError(
"cart_id was required but missing (internal error)".into(),
)))
})?;
let existing = Self::get_by_cart_id_in_conn(tx, cart_id)?;
return existing.ok_or_else(|| {
rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::DatabaseError(
"Order exists for cart_id but could not be loaded".into(),
)))
});
}
let mut items = Vec::with_capacity(input.items.len());
{
let mut stmt = tx.prepare_cached(
"INSERT INTO order_items (id, order_id, product_id, variant_id, sku, name,
quantity, unit_price, discount, tax_amount, total)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
)?;
for item in &input.items {
let item_id = OrderItemId::new();
let item_total = OrderItem::calculate_total(
item.quantity,
item.unit_price,
item.discount.unwrap_or_default(),
item.tax_amount.unwrap_or_default(),
);
stmt.execute(rusqlite::params![
item_id.to_string(),
&id_str,
item.product_id.to_string(),
item.variant_id.map(|variant_id| variant_id.to_string()),
&item.sku,
&item.name,
item.quantity,
item.unit_price.to_string(),
item.discount.unwrap_or_default().to_string(),
item.tax_amount.unwrap_or_default().to_string(),
item_total.to_string(),
])?;
items.push(OrderItem {
id: item_id,
order_id: id,
product_id: item.product_id,
variant_id: item.variant_id,
sku: item.sku.clone(),
name: item.name.clone(),
quantity: item.quantity,
unit_price: item.unit_price,
discount: item.discount.unwrap_or_default(),
tax_amount: item.tax_amount.unwrap_or_default(),
total: item_total,
});
}
}
let reference_id = &id_str;
let mut inv_lookup = tx.prepare_cached("SELECT id FROM inventory_items WHERE sku = ?")?;
for item in &items {
if item.quantity <= 0 {
continue;
}
let item_row = inv_lookup.query_row([&item.sku], |row| row.get::<_, i64>(0));
let item_id = match item_row {
Ok(item_id) => item_id,
Err(rusqlite::Error::QueryReturnedNoRows) => continue,
Err(e) => return Err(e),
};
let available = sum_decimal_query(
tx,
"SELECT quantity_available FROM inventory_balances WHERE item_id = ?",
&[&item_id],
"inventory_balance",
"quantity_available",
)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
let requested = Decimal::from(item.quantity);
let reserve_qty =
if available > Decimal::ZERO { requested.min(available) } else { Decimal::ZERO };
let mut reserved = Decimal::ZERO;
if reserve_qty > Decimal::ZERO {
let reserve_input = ReserveInventory {
sku: item.sku.clone(),
location_id: None,
quantity: reserve_qty,
reference_type: "order".to_string(),
reference_id: reference_id.clone(),
expires_in_seconds: None,
};
match SqliteInventoryRepository::reserve_in_tx(tx, &reserve_input) {
Ok(_reservation) => {
reserved = reserve_qty;
}
Err(err) => {
let commerce_err = map_db_error(err);
if matches!(commerce_err, CommerceError::InsufficientStock { .. }) {
reserved = Decimal::ZERO;
} else {
return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
commerce_err,
)));
}
}
}
}
let remaining = requested - reserved;
if remaining > Decimal::ZERO {
let backorder_input = CreateBackorder {
order_id: id.into_uuid(),
order_line_id: Some(item.id.into_uuid()),
customer_id: input.customer_id.into_uuid(),
sku: item.sku.clone(),
quantity: remaining,
priority: None,
expected_date: None,
promised_date: None,
source_location_id: None,
notes: Some("Auto backorder: insufficient stock".to_string()),
};
create_backorder_in_tx(tx, &backorder_input)?;
}
}
Ok(Order {
id,
order_number,
customer_id: input.customer_id,
status: OrderStatus::Pending,
order_date: now,
total_amount: total,
currency,
payment_status: PaymentStatus::Pending,
fulfillment_status: FulfillmentStatus::Unfulfilled,
payment_method: input.payment_method.clone(),
shipping_method: input.shipping_method.clone(),
tracking_number: None,
notes: input.notes.clone(),
shipping_address: input.shipping_address.clone(),
billing_address: input.billing_address.clone(),
items,
version: 1,
created_at: now,
updated_at: now,
})
}
pub(crate) fn create_from_cart_in_tx(
tx: &rusqlite::Transaction<'_>,
cart_id: Uuid,
input: &CreateOrder,
) -> std::result::Result<Order, rusqlite::Error> {
Self::validate_order_input(input)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
if let Some(existing) = Self::get_by_cart_id_in_conn(tx, cart_id)? {
return Ok(existing);
}
Self::create_internal_in_tx(tx, Some(cart_id), true, input)
}
pub fn create_from_cart(&self, cart_id: Uuid, input: CreateOrder) -> Result<Order> {
if let Some(existing) = self.get_by_cart_id(cart_id)? {
return Ok(existing);
}
self.create_internal(Some(cart_id), true, input)
}
fn create_internal(
&self,
cart_id: Option<Uuid>,
idempotent_by_cart_id: bool,
input: CreateOrder,
) -> Result<Order> {
Self::validate_order_input(&input)?;
with_immediate_transaction(&self.pool, |tx| {
Self::create_internal_in_tx(tx, cart_id, idempotent_by_cart_id, &input)
})
}
}
impl OrderRepository for SqliteOrderRepository {
fn create(&self, input: CreateOrder) -> Result<Order> {
self.create_internal(None, false, input)
}
fn get(&self, id: OrderId) -> Result<Option<Order>> {
let conn = self.conn()?;
let result = conn.query_row(
"SELECT * FROM orders WHERE id = ?",
[id.to_string()],
Self::row_to_order,
);
match result {
Ok(mut order) => {
order.items = Self::load_order_items_with_conn(&conn, id)?;
Ok(Some(order))
}
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(map_db_error(e)),
}
}
fn get_by_number(&self, order_number: &str) -> Result<Option<Order>> {
let conn = self.conn()?;
let result = conn.query_row(
"SELECT * FROM orders WHERE order_number = ?",
[order_number],
Self::row_to_order,
);
match result {
Ok(mut order) => {
order.items = Self::load_order_items_with_conn(&conn, order.id)?;
Ok(Some(order))
}
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(map_db_error(e)),
}
}
fn update(&self, id: OrderId, input: UpdateOrder) -> Result<Order> {
if let Some(address) = &input.shipping_address {
Self::validate_address_input(address, "order.shipping_address")?;
}
if let Some(address) = &input.billing_address {
Self::validate_address_input(address, "order.billing_address")?;
}
let shipping_address_json = input
.shipping_address
.as_ref()
.map(|a| {
serde_json::to_string(a).map_err(|e| {
CommerceError::DatabaseError(format!(
"Failed to serialize order.shipping_address: {e}"
))
})
})
.transpose()?;
let billing_address_json = input
.billing_address
.as_ref()
.map(|a| {
serde_json::to_string(a).map_err(|e| {
CommerceError::DatabaseError(format!(
"Failed to serialize order.billing_address: {e}"
))
})
})
.transpose()?;
struct UpdateOutcome {
order: Order,
post_commit_error: Option<CommerceError>,
}
let outcome = with_immediate_transaction(&self.pool, |tx| {
let now = Utc::now();
let (current_version, current_status_raw, current_payment_status_raw): (
i32,
String,
String,
) = tx
.query_row(
"SELECT version, status, payment_status FROM orders WHERE id = ?",
[id.to_string()],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.map_err(|e| match e {
rusqlite::Error::QueryReturnedNoRows => {
rusqlite::Error::ToSqlConversionFailure(Box::new(
CommerceError::OrderNotFound(id.into_uuid()),
))
}
e => e,
})?;
let current_status: OrderStatus = parse_enum(¤t_status_raw, "order", "status")
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
let current_payment_status: PaymentStatus =
parse_enum(¤t_payment_status_raw, "order", "payment_status")
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
let mut reservation_expired: Option<Uuid> = None;
if let Some(status) = input.status {
if !current_status.can_transition_to(status) {
if status == OrderStatus::Cancelled {
return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
CommerceError::OrderCannotBeCancelled(current_status.to_string()),
)));
}
return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
CommerceError::InvalidOrderStatusTransition {
from: current_status.to_string(),
to: status.to_string(),
},
)));
}
if status == OrderStatus::Refunded {
let effective_payment_status =
input.payment_status.unwrap_or(current_payment_status);
if !matches!(
effective_payment_status,
PaymentStatus::Paid
| PaymentStatus::PartiallyPaid
| PaymentStatus::Refunded
| PaymentStatus::PartiallyRefunded
) {
return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
CommerceError::OrderCannotBeRefunded(
effective_payment_status.to_string(),
),
)));
}
}
if status == OrderStatus::Shipped {
let reservation_ids =
SqliteInventoryRepository::list_reservation_ids_by_reference_in_tx(
tx,
"order",
&id.to_string(),
)?;
for reservation_id in &reservation_ids {
if SqliteInventoryRepository::expire_reservation_if_needed_in_tx(
tx,
*reservation_id,
now,
)? && reservation_expired.is_none()
{
reservation_expired = Some(*reservation_id);
}
}
if reservation_expired.is_none() {
for reservation_id in reservation_ids {
match SqliteInventoryRepository::confirm_reservation_in_tx_with_now(
tx,
reservation_id,
now,
)? {
ReservationConfirmOutcome::Confirmed => {}
ReservationConfirmOutcome::Expired => {
if reservation_expired.is_none() {
reservation_expired = Some(reservation_id);
}
break;
}
}
}
}
}
}
if let Some(expired_id) = reservation_expired {
let result = tx.query_row(
"SELECT * FROM orders WHERE id = ?",
[id.to_string()],
Self::row_to_order,
);
let mut order = match result {
Ok(order) => order,
Err(rusqlite::Error::QueryReturnedNoRows) => {
return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
CommerceError::OrderNotFound(id.into_uuid()),
)));
}
Err(e) => return Err(e),
};
order.items = Self::load_order_items_with_conn(tx, id)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
return Ok(UpdateOutcome {
order,
post_commit_error: Some(CommerceError::ReservationExpired(expired_id)),
});
}
let mut updates = vec!["updated_at = ?"];
let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(now.to_rfc3339())];
if let Some(status) = &input.status {
updates.push("status = ?");
params.push(Box::new(status.to_string()));
}
if let Some(payment_status) = &input.payment_status {
updates.push("payment_status = ?");
params.push(Box::new(payment_status.to_string()));
}
if let Some(fulfillment_status) = &input.fulfillment_status {
updates.push("fulfillment_status = ?");
params.push(Box::new(fulfillment_status.to_string()));
}
if let Some(tracking) = &input.tracking_number {
updates.push("tracking_number = ?");
params.push(Box::new(tracking.clone()));
}
if let Some(notes) = &input.notes {
updates.push("notes = ?");
params.push(Box::new(notes.clone()));
}
if let Some(addr_json) = &shipping_address_json {
updates.push("shipping_address = ?");
params.push(Box::new(addr_json.clone()));
}
if let Some(addr_json) = &billing_address_json {
updates.push("billing_address = ?");
params.push(Box::new(addr_json.clone()));
}
updates.push("version = version + 1");
params.push(Box::new(id.to_string()));
params.push(Box::new(current_version));
let sql =
format!("UPDATE orders SET {} WHERE id = ? AND version = ?", updates.join(", "));
let params_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(std::convert::AsRef::as_ref).collect();
let rows_affected = tx.execute(&sql, params_refs.as_slice())?;
if rows_affected == 0 {
return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
CommerceError::VersionConflict {
entity: "order".to_string(),
id: id.to_string(),
expected_version: current_version,
},
)));
}
if matches!(input.status, Some(OrderStatus::Cancelled)) {
let reservation_ids =
SqliteInventoryRepository::list_reservation_ids_by_reference_in_tx(
tx,
"order",
&id.to_string(),
)?;
for reservation_id in reservation_ids {
SqliteInventoryRepository::release_reservation_in_tx(tx, reservation_id)?;
}
cancel_backorders_for_order_in_tx(tx, id.into_uuid())?;
}
let result = tx.query_row(
"SELECT * FROM orders WHERE id = ?",
[id.to_string()],
Self::row_to_order,
);
let mut order = match result {
Ok(order) => order,
Err(rusqlite::Error::QueryReturnedNoRows) => {
return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
CommerceError::OrderNotFound(id.into_uuid()),
)));
}
Err(e) => return Err(e),
};
order.items = Self::load_order_items_with_conn(tx, id)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
Ok(UpdateOutcome { order, post_commit_error: None })
})?;
if let Some(err) = outcome.post_commit_error {
return Err(err);
}
Ok(outcome.order)
}
fn list(&self, filter: OrderFilter) -> Result<Vec<Order>> {
let conn = self.conn()?;
let mut sql = "SELECT * FROM orders WHERE 1=1".to_string();
let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![];
if let Some(customer_id) = &filter.customer_id {
sql.push_str(" AND customer_id = ?");
params.push(Box::new(customer_id.to_string()));
}
if let Some(status) = &filter.status {
sql.push_str(" AND status = ?");
params.push(Box::new(status.to_string()));
}
if let Some(payment_status) = &filter.payment_status {
sql.push_str(" AND payment_status = ?");
params.push(Box::new(payment_status.to_string()));
}
if let Some(fulfillment_status) = &filter.fulfillment_status {
sql.push_str(" AND fulfillment_status = ?");
params.push(Box::new(fulfillment_status.to_string()));
}
if let Some(from) = &filter.from_date {
sql.push_str(" AND order_date >= ?");
params.push(Box::new(from.to_rfc3339()));
}
if let Some(to) = &filter.to_date {
sql.push_str(" AND order_date <= ?");
params.push(Box::new(to.to_rfc3339()));
}
if let Some((cursor_date, cursor_id)) = &filter.after_cursor {
sql.push_str(" AND (order_date < ? OR (order_date = ? AND id < ?))");
params.push(Box::new(cursor_date.clone()));
params.push(Box::new(cursor_date.clone()));
params.push(Box::new(cursor_id.clone()));
}
sql.push_str(" ORDER BY order_date DESC, id DESC");
let offset = if filter.after_cursor.is_none() { filter.offset } else { None };
crate::sqlite::append_limit_offset(&mut sql, filter.limit, offset);
let params_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(std::convert::AsRef::as_ref).collect();
let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
let orders = stmt
.query_map(params_refs.as_slice(), Self::row_to_order)
.map_err(map_db_error)?
.collect::<rusqlite::Result<Vec<_>>>()
.map_err(map_db_error)?;
let ids: Vec<OrderId> = orders.iter().map(|o| o.id).collect();
let mut items_by_id = Self::load_order_items_batch(&conn, &ids)?;
let mut result = vec![];
for mut order in orders {
order.items = items_by_id.remove(&order.id).unwrap_or_default();
result.push(order);
}
Ok(result)
}
fn delete(&self, id: OrderId) -> Result<()> {
with_immediate_transaction(&self.pool, |tx| {
tx.execute("DELETE FROM order_items WHERE order_id = ?", [id.to_string()])?;
tx.execute("DELETE FROM orders WHERE id = ?", [id.to_string()])?;
Ok(())
})
}
fn add_item(&self, order_id: OrderId, item: CreateOrderItem) -> Result<OrderItem> {
validate_required_uuid("order.id", order_id.into_uuid())?;
Self::validate_order_item_input(&item)?;
let item_id = OrderItemId::new();
let item_total = OrderItem::calculate_total(
item.quantity,
item.unit_price,
item.discount.unwrap_or_default(),
item.tax_amount.unwrap_or_default(),
);
with_immediate_transaction(&self.pool, |tx| {
tx.execute(
"INSERT INTO order_items (id, order_id, product_id, variant_id, sku, name,
quantity, unit_price, discount, tax_amount, total)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
item_id.to_string(),
order_id.to_string(),
item.product_id.to_string(),
item.variant_id.map(|v| v.to_string()),
item.sku,
item.name,
item.quantity,
item.unit_price.to_string(),
item.discount.unwrap_or_default().to_string(),
item.tax_amount.unwrap_or_default().to_string(),
item_total.to_string(),
],
)?;
self.update_order_total(tx, order_id)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
Ok(())
})?;
Ok(OrderItem {
id: item_id,
order_id,
product_id: item.product_id,
variant_id: item.variant_id,
sku: item.sku,
name: item.name,
quantity: item.quantity,
unit_price: item.unit_price,
discount: item.discount.unwrap_or_default(),
tax_amount: item.tax_amount.unwrap_or_default(),
total: item_total,
})
}
fn remove_item(&self, order_id: OrderId, item_id: OrderItemId) -> Result<()> {
with_immediate_transaction(&self.pool, |tx| {
tx.execute(
"DELETE FROM order_items WHERE id = ? AND order_id = ?",
[item_id.to_string(), order_id.to_string()],
)?;
self.update_order_total(tx, order_id)
.map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?;
Ok(())
})
}
fn count(&self, filter: OrderFilter) -> Result<u64> {
let conn = self.conn()?;
let mut sql = "SELECT COUNT(*) FROM orders WHERE 1=1".to_string();
let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![];
if let Some(customer_id) = &filter.customer_id {
sql.push_str(" AND customer_id = ?");
params.push(Box::new(customer_id.to_string()));
}
if let Some(status) = &filter.status {
sql.push_str(" AND status = ?");
params.push(Box::new(status.to_string()));
}
if let Some(payment_status) = &filter.payment_status {
sql.push_str(" AND payment_status = ?");
params.push(Box::new(payment_status.to_string()));
}
if let Some(fulfillment_status) = &filter.fulfillment_status {
sql.push_str(" AND fulfillment_status = ?");
params.push(Box::new(fulfillment_status.to_string()));
}
if let Some(from) = &filter.from_date {
sql.push_str(" AND order_date >= ?");
params.push(Box::new(from.to_rfc3339()));
}
if let Some(to) = &filter.to_date {
sql.push_str(" AND order_date <= ?");
params.push(Box::new(to.to_rfc3339()));
}
let params_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(std::convert::AsRef::as_ref).collect();
let count: i64 =
conn.query_row(&sql, params_refs.as_slice(), |row| row.get(0)).map_err(map_db_error)?;
Ok(count as u64)
}
fn create_batch(&self, inputs: Vec<CreateOrder>) -> Result<BatchResult<Order>> {
validate_batch_size(&inputs)?;
let mut result = BatchResult::with_capacity(inputs.len());
let mut conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
for (index, input) in inputs.into_iter().enumerate() {
Self::validate_order_input(&input)?;
let tx_result =
conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate);
match tx_result {
Ok(tx) => {
match Self::create_internal_in_tx(&tx, None, false, &input) {
Ok(order) => {
if let Err(e) = tx.commit() {
result.record_failure(index, None, &map_db_error(e));
} else {
result.record_success(order);
}
}
Err(e) => {
result.record_failure(index, None, &map_db_error(e));
}
}
}
Err(e) => result.record_failure(index, None, &map_db_error(e)),
}
}
Ok(result)
}
fn create_batch_atomic(&self, inputs: Vec<CreateOrder>) -> Result<Vec<Order>> {
validate_batch_size(&inputs)?;
if inputs.is_empty() {
return Ok(vec![]);
}
let mut conn = self.conn()?;
let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
let mut results = Vec::with_capacity(inputs.len());
for input in inputs {
Self::validate_order_input(&input)?;
let id = OrderId::new();
let order_number = Self::generate_order_number();
let now = Utc::now();
let currency = input.currency.unwrap_or_default();
let total: Decimal = input
.items
.iter()
.map(|item| {
OrderItem::calculate_total(
item.quantity,
item.unit_price,
item.discount.unwrap_or_default(),
item.tax_amount.unwrap_or_default(),
)
})
.sum();
let shipping_address_json = input
.shipping_address
.as_ref()
.map(|a| {
serde_json::to_string(a).map_err(|e| {
CommerceError::DatabaseError(format!(
"Failed to serialize order.shipping_address: {e}"
))
})
})
.transpose()?;
let billing_address_json = input
.billing_address
.as_ref()
.map(|a| {
serde_json::to_string(a).map_err(|e| {
CommerceError::DatabaseError(format!(
"Failed to serialize order.billing_address: {e}"
))
})
})
.transpose()?;
tx.execute(
"INSERT INTO orders (id, order_number, customer_id, status, order_date, total_amount,
currency, payment_status, fulfillment_status, payment_method,
shipping_method, notes, shipping_address, billing_address,
created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
id.to_string(),
&order_number,
input.customer_id.to_string(),
"pending",
now.to_rfc3339(),
total.to_string(),
¤cy,
"pending",
"unfulfilled",
&input.payment_method,
&input.shipping_method,
&input.notes,
&shipping_address_json,
&billing_address_json,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(map_db_error)?;
let mut items = Vec::with_capacity(input.items.len());
for item in &input.items {
let item_id = OrderItemId::new();
let item_total = OrderItem::calculate_total(
item.quantity,
item.unit_price,
item.discount.unwrap_or_default(),
item.tax_amount.unwrap_or_default(),
);
tx.execute(
"INSERT INTO order_items (id, order_id, product_id, variant_id, sku, name,
quantity, unit_price, discount, tax_amount, total)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
item_id.to_string(),
id.to_string(),
item.product_id.to_string(),
item.variant_id.map(|v| v.to_string()),
&item.sku,
&item.name,
item.quantity,
item.unit_price.to_string(),
item.discount.unwrap_or_default().to_string(),
item.tax_amount.unwrap_or_default().to_string(),
item_total.to_string(),
],
)
.map_err(map_db_error)?;
items.push(OrderItem {
id: item_id,
order_id: id,
product_id: item.product_id,
variant_id: item.variant_id,
sku: item.sku.clone(),
name: item.name.clone(),
quantity: item.quantity,
unit_price: item.unit_price,
discount: item.discount.unwrap_or_default(),
tax_amount: item.tax_amount.unwrap_or_default(),
total: item_total,
});
}
results.push(Order {
id,
order_number,
customer_id: input.customer_id,
status: OrderStatus::Pending,
order_date: now,
total_amount: total,
currency,
payment_status: PaymentStatus::Pending,
fulfillment_status: FulfillmentStatus::Unfulfilled,
payment_method: input.payment_method,
shipping_method: input.shipping_method,
tracking_number: None,
notes: input.notes,
shipping_address: input.shipping_address,
billing_address: input.billing_address,
items,
version: 1,
created_at: now,
updated_at: now,
});
}
tx.commit().map_err(map_db_error)?;
Ok(results)
}
fn update_batch(&self, updates: Vec<(OrderId, UpdateOrder)>) -> Result<BatchResult<Order>> {
validate_batch_size(&updates)?;
let mut result = BatchResult::with_capacity(updates.len());
for (index, (id, input)) in updates.into_iter().enumerate() {
match self.update(id, input) {
Ok(order) => result.record_success(order),
Err(e) => result.record_failure(index, Some(id.to_string()), &e),
}
}
Ok(result)
}
fn update_batch_atomic(&self, updates: Vec<(OrderId, UpdateOrder)>) -> Result<Vec<Order>> {
validate_batch_size(&updates)?;
if updates.is_empty() {
return Ok(vec![]);
}
let mut conn = self.conn()?;
let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
let mut results = Vec::with_capacity(updates.len());
for (id, input) in updates {
let now = Utc::now();
let (current_version, current_status_raw, current_payment_status_raw): (
i32,
String,
String,
) = tx
.query_row(
"SELECT version, status, payment_status FROM orders WHERE id = ?",
[id.to_string()],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.map_err(|e| match e {
rusqlite::Error::QueryReturnedNoRows => {
CommerceError::OrderNotFound(id.into_uuid())
}
e => map_db_error(e),
})?;
let current_status: OrderStatus = parse_enum(¤t_status_raw, "order", "status")?;
let current_payment_status: PaymentStatus =
parse_enum(¤t_payment_status_raw, "order", "payment_status")?;
let mut update_parts = vec!["updated_at = ?"];
let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(now.to_rfc3339())];
if let Some(status) = &input.status {
let next_status = *status;
if !current_status.can_transition_to(next_status) {
if next_status == OrderStatus::Cancelled {
return Err(CommerceError::OrderCannotBeCancelled(
current_status.to_string(),
));
}
return Err(CommerceError::InvalidOrderStatusTransition {
from: current_status.to_string(),
to: next_status.to_string(),
});
}
if next_status == OrderStatus::Refunded {
let effective_payment_status =
input.payment_status.unwrap_or(current_payment_status);
if !matches!(
effective_payment_status,
PaymentStatus::Paid
| PaymentStatus::PartiallyPaid
| PaymentStatus::Refunded
| PaymentStatus::PartiallyRefunded
) {
return Err(CommerceError::OrderCannotBeRefunded(
effective_payment_status.to_string(),
));
}
}
update_parts.push("status = ?");
params.push(Box::new(status.to_string()));
}
if let Some(payment_status) = &input.payment_status {
update_parts.push("payment_status = ?");
params.push(Box::new(payment_status.to_string()));
}
if let Some(fulfillment_status) = &input.fulfillment_status {
update_parts.push("fulfillment_status = ?");
params.push(Box::new(fulfillment_status.to_string()));
}
if let Some(tracking) = &input.tracking_number {
update_parts.push("tracking_number = ?");
params.push(Box::new(tracking.clone()));
}
if let Some(notes) = &input.notes {
update_parts.push("notes = ?");
params.push(Box::new(notes.clone()));
}
if let Some(addr) = &input.shipping_address {
Self::validate_address_input(addr, "order.shipping_address")?;
update_parts.push("shipping_address = ?");
let address_json = serde_json::to_string(addr).map_err(|e| {
CommerceError::DatabaseError(format!(
"Failed to serialize order.shipping_address: {e}"
))
})?;
params.push(Box::new(address_json));
}
if let Some(addr) = &input.billing_address {
Self::validate_address_input(addr, "order.billing_address")?;
update_parts.push("billing_address = ?");
let address_json = serde_json::to_string(addr).map_err(|e| {
CommerceError::DatabaseError(format!(
"Failed to serialize order.billing_address: {e}"
))
})?;
params.push(Box::new(address_json));
}
update_parts.push("version = version + 1");
params.push(Box::new(id.to_string()));
params.push(Box::new(current_version));
let sql = format!(
"UPDATE orders SET {} WHERE id = ? AND version = ?",
update_parts.join(", ")
);
let params_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(std::convert::AsRef::as_ref).collect();
let rows_affected = tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
if rows_affected == 0 {
return Err(CommerceError::VersionConflict {
entity: "order".to_string(),
id: id.to_string(),
expected_version: current_version,
});
}
let order = tx
.query_row(
"SELECT * FROM orders WHERE id = ?",
[id.to_string()],
Self::row_to_order,
)
.map_err(map_db_error)?;
results.push(order);
}
tx.commit().map_err(map_db_error)?;
let conn = self.conn()?;
let ids: Vec<OrderId> = results.iter().map(|o| o.id).collect();
let mut items_by_id = Self::load_order_items_batch(&conn, &ids)?;
for order in &mut results {
order.items = items_by_id.remove(&order.id).unwrap_or_default();
}
Ok(results)
}
fn delete_batch(&self, ids: Vec<OrderId>) -> Result<BatchResult<OrderId>> {
validate_batch_size(&ids)?;
let mut result = BatchResult::with_capacity(ids.len());
for (index, id) in ids.into_iter().enumerate() {
match self.delete(id) {
Ok(()) => result.record_success(id),
Err(e) => result.record_failure(index, Some(id.to_string()), &e),
}
}
Ok(result)
}
fn delete_batch_atomic(&self, ids: Vec<OrderId>) -> Result<()> {
validate_batch_size(&ids)?;
if ids.is_empty() {
return Ok(());
}
let mut conn = self.conn()?;
let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
let placeholders = build_in_clause(ids.len());
let raw_ids: Vec<Uuid> = ids.iter().map(|id| id.into_uuid()).collect();
let params = uuid_params(&raw_ids);
let params_refs = params_refs(¶ms);
let sql = format!("DELETE FROM order_items WHERE order_id IN ({placeholders})");
tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
let sql = format!("DELETE FROM orders WHERE id IN ({placeholders})");
tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
tx.commit().map_err(map_db_error)?;
Ok(())
}
fn get_batch(&self, ids: Vec<OrderId>) -> Result<Vec<Order>> {
validate_batch_size(&ids)?;
if ids.is_empty() {
return Ok(vec![]);
}
let conn = self.conn()?;
let placeholders = build_in_clause(ids.len());
let sql = format!("SELECT * FROM orders WHERE id IN ({placeholders})");
let raw_ids: Vec<Uuid> = ids.iter().map(|id| id.into_uuid()).collect();
let params = uuid_params(&raw_ids);
let params_refs = params_refs(¶ms);
let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
let orders = stmt
.query_map(params_refs.as_slice(), Self::row_to_order)
.map_err(map_db_error)?
.collect::<rusqlite::Result<Vec<_>>>()
.map_err(map_db_error)?;
let order_ids: Vec<OrderId> = orders.iter().map(|o| o.id).collect();
let mut items_by_id = Self::load_order_items_batch(&conn, &order_ids)?;
let mut result = vec![];
for mut order in orders {
order.items = items_by_id.remove(&order.id).unwrap_or_default();
result.push(order);
}
Ok(result)
}
}
impl SqliteOrderRepository {
fn update_order_total(&self, conn: &rusqlite::Connection, order_id: OrderId) -> Result<()> {
let current_version: i32 = conn
.query_row("SELECT version FROM orders WHERE id = ?", [order_id.to_string()], |row| {
row.get(0)
})
.map_err(|e| match e {
rusqlite::Error::QueryReturnedNoRows => {
CommerceError::OrderNotFound(order_id.into_uuid())
}
e => map_db_error(e),
})?;
let order_id_param = order_id.to_string();
let order_params: [&dyn rusqlite::ToSql; 1] = [&order_id_param];
let total = sum_decimal_query(
conn,
"SELECT total FROM order_items WHERE order_id = ?",
&order_params,
"order_item",
"total",
)?;
let total = total.to_string();
let rows_affected = conn
.execute(
"UPDATE orders SET total_amount = ?, updated_at = ?, version = version + 1 WHERE id = ? AND version = ?",
rusqlite::params![
total,
Utc::now().to_rfc3339(),
order_id.to_string(),
current_version
],
)
.map_err(map_db_error)?;
if rows_affected == 0 {
return Err(CommerceError::VersionConflict {
entity: "order".to_string(),
id: order_id.to_string(),
expected_version: current_version,
});
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::str::FromStr;
#[test]
fn parses_payment_status_snake_case_and_legacy() {
assert_eq!(
PaymentStatus::from_str("partially_paid").unwrap(),
PaymentStatus::PartiallyPaid
);
assert_eq!(PaymentStatus::from_str("partiallypaid").unwrap(), PaymentStatus::PartiallyPaid);
assert_eq!(
PaymentStatus::from_str("partially_refunded").unwrap(),
PaymentStatus::PartiallyRefunded
);
assert_eq!(
PaymentStatus::from_str("partiallyrefunded").unwrap(),
PaymentStatus::PartiallyRefunded
);
}
#[test]
fn parses_fulfillment_status_snake_case_and_legacy() {
assert_eq!(
FulfillmentStatus::from_str("partially_fulfilled").unwrap(),
FulfillmentStatus::PartiallyFulfilled
);
assert_eq!(
FulfillmentStatus::from_str("partiallyfulfilled").unwrap(),
FulfillmentStatus::PartiallyFulfilled
);
}
}