use super::{
build_in_clause, map_db_error, params_refs, parse_datetime, parse_datetime_opt,
parse_datetime_row, parse_decimal_opt, parse_enum, parse_uuid, parse_uuid_row, uuid_params,
};
use chrono::Utc;
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
use stateset_core::{
AddShipmentEvent, BatchResult, CommerceError, CreateShipment, CreateShipmentItem, OrderId,
ProductId, Result, Shipment, ShipmentEvent, ShipmentFilter, ShipmentId, ShipmentItem,
ShipmentRepository, ShipmentStatus, ShippingCarrier, UpdateShipment, validate_batch_size,
};
use uuid::Uuid;
#[derive(Debug)]
pub struct SqliteShipmentRepository {
pool: Pool<SqliteConnectionManager>,
}
impl SqliteShipmentRepository {
#[must_use]
pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
Self { pool }
}
fn load_items(&self, shipment_id: ShipmentId) -> Result<Vec<ShipmentItem>> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut stmt = conn
.prepare(
"SELECT id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at
FROM shipment_items WHERE shipment_id = ?",
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let rows = stmt
.query_map([shipment_id.to_string()], |row| {
Ok(ShipmentItem {
id: parse_uuid_row(&row.get::<_, String>(0)?, "shipment_item", "id")?,
shipment_id: ShipmentId::from(parse_uuid_row(
&row.get::<_, String>(1)?,
"shipment_item",
"shipment_id",
)?),
order_item_id: row
.get::<_, Option<String>>(2)?
.map(|s| parse_uuid_row(&s, "shipment_item", "order_item_id"))
.transpose()?,
product_id: row
.get::<_, Option<String>>(3)?
.map(|s| parse_uuid_row(&s, "shipment_item", "product_id"))
.transpose()?
.map(ProductId::from),
sku: row.get(4)?,
name: row.get(5)?,
quantity: row.get(6)?,
created_at: parse_datetime_row(
&row.get::<_, String>(7)?,
"shipment_item",
"created_at",
)?,
updated_at: parse_datetime_row(
&row.get::<_, String>(8)?,
"shipment_item",
"updated_at",
)?,
})
})
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut items = Vec::new();
for row in rows {
items.push(row.map_err(|e| CommerceError::DatabaseError(e.to_string()))?);
}
Ok(items)
}
fn load_events(&self, shipment_id: ShipmentId) -> Result<Vec<ShipmentEvent>> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut stmt = conn
.prepare(
"SELECT id, shipment_id, event_type, location, description, event_time, created_at
FROM shipment_events WHERE shipment_id = ? ORDER BY event_time DESC",
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let rows = stmt
.query_map([shipment_id.to_string()], |row| {
Ok(ShipmentEvent {
id: parse_uuid_row(&row.get::<_, String>(0)?, "shipment_event", "id")?,
shipment_id: ShipmentId::from(parse_uuid_row(
&row.get::<_, String>(1)?,
"shipment_event",
"shipment_id",
)?),
event_type: row.get(2)?,
location: row.get(3)?,
description: row.get(4)?,
event_time: parse_datetime_row(
&row.get::<_, String>(5)?,
"shipment_event",
"event_time",
)?,
created_at: parse_datetime_row(
&row.get::<_, String>(6)?,
"shipment_event",
"created_at",
)?,
})
})
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut events = Vec::new();
for row in rows {
events.push(row.map_err(|e| CommerceError::DatabaseError(e.to_string()))?);
}
Ok(events)
}
fn update_status(&self, id: ShipmentId, status: ShipmentStatus) -> Result<Shipment> {
let now = Utc::now();
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE shipments SET status = ?, updated_at = ? WHERE id = ?",
rusqlite::params![status.to_string(), now.to_rfc3339(), id.to_string()],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get(id)?.ok_or(CommerceError::NotFound)
}
}
impl ShipmentRepository for SqliteShipmentRepository {
fn create(&self, input: CreateShipment) -> Result<Shipment> {
let id = Uuid::new_v4();
let shipment_number = Shipment::generate_shipment_number();
let now = Utc::now();
let carrier = input.carrier.unwrap_or_default();
let method = input.shipping_method.unwrap_or_default();
let tracking_url = input.tracking_number.as_ref().and_then(|tn| carrier.tracking_url(tn));
let mut items = 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(|e| CommerceError::DatabaseError(e.to_string()))?;
tx.execute(
"INSERT INTO shipments (id, shipment_number, order_id, status, carrier, shipping_method,
tracking_number, tracking_url, recipient_name, recipient_email, recipient_phone,
shipping_address, weight_kg, dimensions, shipping_cost, insurance_amount,
signature_required, estimated_delivery, notes, created_at, updated_at)
VALUES (?, ?, ?, 'pending', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
id.to_string(),
shipment_number,
input.order_id.to_string(),
carrier.to_string(),
method.to_string(),
input.tracking_number,
tracking_url,
input.recipient_name,
input.recipient_email,
input.recipient_phone,
input.shipping_address,
input.weight_kg.map(|w| w.to_string()),
input.dimensions,
input.shipping_cost.map(|c| c.to_string()),
input.insurance_amount.map(|a| a.to_string()),
i32::from(input.signature_required.unwrap_or(false)),
input.estimated_delivery.map(|dt| dt.to_rfc3339()),
input.notes,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
if let Some(item_inputs) = &input.items {
for item_input in item_inputs {
let item_id = Uuid::new_v4();
tx.execute(
"INSERT INTO shipment_items (id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
item_id.to_string(),
id.to_string(),
item_input.order_item_id.map(|u| u.to_string()),
item_input.product_id.map(|u| u.to_string()),
item_input.sku,
item_input.name,
item_input.quantity,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
items.push(ShipmentItem {
id: item_id,
shipment_id: ShipmentId::from(id),
order_item_id: item_input.order_item_id,
product_id: item_input.product_id,
sku: item_input.sku.clone(),
name: item_input.name.clone(),
quantity: item_input.quantity,
created_at: now,
updated_at: now,
});
}
}
tx.commit().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
Ok(Shipment {
id: ShipmentId::from(id),
shipment_number,
order_id: input.order_id,
status: ShipmentStatus::Pending,
carrier,
shipping_method: method,
tracking_number: input.tracking_number,
tracking_url,
recipient_name: input.recipient_name,
recipient_email: input.recipient_email,
recipient_phone: input.recipient_phone,
shipping_address: input.shipping_address,
weight_kg: input.weight_kg,
dimensions: input.dimensions,
shipping_cost: input.shipping_cost,
insurance_amount: input.insurance_amount,
signature_required: input.signature_required.unwrap_or(false),
shipped_at: None,
estimated_delivery: input.estimated_delivery,
delivered_at: None,
notes: input.notes,
items,
events: vec![],
version: 1,
created_at: now,
updated_at: now,
})
}
fn get(&self, id: ShipmentId) -> Result<Option<Shipment>> {
let shipment_data = {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let result = conn.query_row(
"SELECT id, shipment_number, order_id, status, carrier, shipping_method,
tracking_number, tracking_url, recipient_name, recipient_email, recipient_phone,
shipping_address, weight_kg, dimensions, shipping_cost, insurance_amount,
signature_required, shipped_at, estimated_delivery, delivered_at, notes,
created_at, updated_at
FROM shipments WHERE id = ?",
[id.to_string()],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, String>(5)?,
row.get::<_, Option<String>>(6)?,
row.get::<_, Option<String>>(7)?,
row.get::<_, String>(8)?,
row.get::<_, Option<String>>(9)?,
row.get::<_, Option<String>>(10)?,
row.get::<_, String>(11)?,
row.get::<_, Option<String>>(12)?,
row.get::<_, Option<String>>(13)?,
row.get::<_, Option<String>>(14)?,
row.get::<_, Option<String>>(15)?,
row.get::<_, i32>(16)?,
row.get::<_, Option<String>>(17)?,
row.get::<_, Option<String>>(18)?,
row.get::<_, Option<String>>(19)?,
row.get::<_, Option<String>>(20)?,
row.get::<_, String>(21)?,
row.get::<_, String>(22)?,
))
},
);
match result {
Ok(data) => Some(data),
Err(rusqlite::Error::QueryReturnedNoRows) => None,
Err(e) => return Err(CommerceError::DatabaseError(e.to_string())),
}
};
match shipment_data {
Some((
id_str,
shipment_number,
order_id,
status,
carrier,
shipping_method,
tracking_number,
tracking_url,
recipient_name,
recipient_email,
recipient_phone,
shipping_address,
weight_kg,
dimensions,
shipping_cost,
insurance_amount,
signature_required,
shipped_at,
estimated_delivery,
delivered_at,
notes,
created_at,
updated_at,
)) => {
let shipment_id = ShipmentId::from(parse_uuid(&id_str, "shipment", "id")?);
let items = self.load_items(shipment_id)?;
let events = self.load_events(shipment_id)?;
Ok(Some(Shipment {
id: shipment_id,
shipment_number,
order_id: OrderId::from(parse_uuid(&order_id, "shipment", "order_id")?),
status: parse_enum(&status, "shipment", "status")?,
carrier: parse_enum(&carrier, "shipment", "carrier")?,
shipping_method: parse_enum(&shipping_method, "shipment", "shipping_method")?,
tracking_number,
tracking_url,
recipient_name,
recipient_email,
recipient_phone,
shipping_address,
weight_kg: parse_decimal_opt(weight_kg, "shipment", "weight_kg")?,
dimensions,
shipping_cost: parse_decimal_opt(shipping_cost, "shipment", "shipping_cost")?,
insurance_amount: parse_decimal_opt(
insurance_amount,
"shipment",
"insurance_amount",
)?,
signature_required: signature_required != 0,
shipped_at: parse_datetime_opt(shipped_at, "shipment", "shipped_at")?,
estimated_delivery: parse_datetime_opt(
estimated_delivery,
"shipment",
"estimated_delivery",
)?,
delivered_at: parse_datetime_opt(delivered_at, "shipment", "delivered_at")?,
notes,
items,
events,
version: 1, created_at: parse_datetime(&created_at, "shipment", "created_at")?,
updated_at: parse_datetime(&updated_at, "shipment", "updated_at")?,
}))
}
None => Ok(None),
}
}
fn get_by_number(&self, shipment_number: &str) -> Result<Option<Shipment>> {
let id_result = {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let result = conn.query_row(
"SELECT id FROM shipments WHERE shipment_number = ?",
[shipment_number],
|row| row.get::<_, String>(0),
);
match result {
Ok(id_str) => Some(ShipmentId::from(parse_uuid(&id_str, "shipment", "id")?)),
Err(rusqlite::Error::QueryReturnedNoRows) => None,
Err(e) => return Err(CommerceError::DatabaseError(e.to_string())),
}
};
match id_result {
Some(id) => self.get(id),
None => Ok(None),
}
}
fn get_by_tracking(&self, tracking_number: &str) -> Result<Option<Shipment>> {
let id_result = {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let result = conn.query_row(
"SELECT id FROM shipments WHERE tracking_number = ?",
[tracking_number],
|row| row.get::<_, String>(0),
);
match result {
Ok(id_str) => Some(ShipmentId::from(parse_uuid(&id_str, "shipment", "id")?)),
Err(rusqlite::Error::QueryReturnedNoRows) => None,
Err(e) => return Err(CommerceError::DatabaseError(e.to_string())),
}
};
match id_result {
Some(id) => self.get(id),
None => Ok(None),
}
}
fn update(&self, id: ShipmentId, input: UpdateShipment) -> Result<Shipment> {
let existing = self.get(id)?.ok_or(CommerceError::NotFound)?;
let now = Utc::now();
let new_status = input.status.unwrap_or(existing.status);
let new_carrier = input.carrier.unwrap_or(existing.carrier);
let new_tracking = input.tracking_number.or(existing.tracking_number);
let new_tracking_url = new_tracking.as_ref().and_then(|tn| new_carrier.tracking_url(tn));
let new_recipient_name = input.recipient_name.unwrap_or(existing.recipient_name);
let new_recipient_email = input.recipient_email.or(existing.recipient_email);
let new_recipient_phone = input.recipient_phone.or(existing.recipient_phone);
let new_shipping_address = input.shipping_address.unwrap_or(existing.shipping_address);
let new_weight = input.weight_kg.or(existing.weight_kg);
let new_dimensions = input.dimensions.or(existing.dimensions);
let new_shipping_cost = input.shipping_cost.or(existing.shipping_cost);
let new_estimated_delivery = input.estimated_delivery.or(existing.estimated_delivery);
let new_notes = input.notes.or(existing.notes);
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE shipments SET status = ?, carrier = ?, tracking_number = ?, tracking_url = ?,
recipient_name = ?, recipient_email = ?, recipient_phone = ?, shipping_address = ?,
weight_kg = ?, dimensions = ?, shipping_cost = ?, estimated_delivery = ?, notes = ?,
updated_at = ? WHERE id = ?",
rusqlite::params![
new_status.to_string(),
new_carrier.to_string(),
new_tracking,
new_tracking_url,
new_recipient_name,
new_recipient_email,
new_recipient_phone,
new_shipping_address,
new_weight.map(|w| w.to_string()),
new_dimensions,
new_shipping_cost.map(|c| c.to_string()),
new_estimated_delivery.map(|dt| dt.to_rfc3339()),
new_notes,
now.to_rfc3339(),
id.to_string(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get(id)?.ok_or(CommerceError::NotFound)
}
fn list(&self, filter: ShipmentFilter) -> Result<Vec<Shipment>> {
let ids = {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let limit = i64::from(filter.limit.unwrap_or(100));
let offset = i64::from(filter.offset.unwrap_or(0));
let mut sql = "SELECT id FROM shipments WHERE 1=1".to_string();
let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(order_id) = filter.order_id {
sql.push_str(" AND order_id = ?");
params.push(Box::new(order_id.to_string()));
}
if let Some(status) = filter.status {
sql.push_str(" AND status = ?");
params.push(Box::new(status.to_string()));
}
if let Some(carrier) = filter.carrier {
sql.push_str(" AND carrier = ?");
params.push(Box::new(carrier.to_string()));
}
if let Some(tracking_number) = filter.tracking_number {
sql.push_str(" AND tracking_number = ?");
params.push(Box::new(tracking_number));
}
sql.push_str(" ORDER BY created_at DESC LIMIT ? OFFSET ?");
params.push(Box::new(limit));
params.push(Box::new(offset));
let mut stmt =
conn.prepare(&sql).map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let param_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(std::convert::AsRef::as_ref).collect();
let rows = stmt
.query_map(param_refs.as_slice(), |row| row.get::<_, String>(0))
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut id_list = Vec::new();
for row in rows {
let id_str = row.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
id_list.push(ShipmentId::from(parse_uuid(&id_str, "shipment", "id")?));
}
id_list
};
let mut shipments = Vec::new();
for id in ids {
if let Some(shipment) = self.get(id)? {
shipments.push(shipment);
}
}
Ok(shipments)
}
fn for_order(&self, order_id: OrderId) -> Result<Vec<Shipment>> {
self.list(ShipmentFilter { order_id: Some(order_id), ..Default::default() })
}
fn delete(&self, id: ShipmentId) -> Result<()> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE shipments SET status = 'cancelled', updated_at = ? WHERE id = ?",
rusqlite::params![Utc::now().to_rfc3339(), id.to_string()],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(())
}
fn mark_processing(&self, id: ShipmentId) -> Result<Shipment> {
self.update_status(id, ShipmentStatus::Processing)
}
fn mark_ready(&self, id: ShipmentId) -> Result<Shipment> {
self.update_status(id, ShipmentStatus::ReadyToShip)
}
fn ship(&self, id: ShipmentId, tracking_number: Option<String>) -> Result<Shipment> {
let existing = self.get(id)?.ok_or(CommerceError::NotFound)?;
let now = Utc::now();
let tracking_url =
tracking_number.as_ref().and_then(|tn| existing.carrier.tracking_url(tn));
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE shipments SET status = 'shipped', tracking_number = COALESCE(?, tracking_number),
tracking_url = COALESCE(?, tracking_url), shipped_at = ?, updated_at = ? WHERE id = ?",
rusqlite::params![
tracking_number,
tracking_url,
now.to_rfc3339(),
now.to_rfc3339(),
id.to_string(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get(id)?.ok_or(CommerceError::NotFound)
}
fn mark_in_transit(&self, id: ShipmentId) -> Result<Shipment> {
self.update_status(id, ShipmentStatus::InTransit)
}
fn mark_out_for_delivery(&self, id: ShipmentId) -> Result<Shipment> {
self.update_status(id, ShipmentStatus::OutForDelivery)
}
fn mark_delivered(&self, id: ShipmentId) -> Result<Shipment> {
let now = Utc::now();
{
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute(
"UPDATE shipments SET status = 'delivered', delivered_at = ?, updated_at = ? WHERE id = ?",
rusqlite::params![now.to_rfc3339(), now.to_rfc3339(), id.to_string()],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
}
self.get(id)?.ok_or(CommerceError::NotFound)
}
fn mark_failed(&self, id: ShipmentId) -> Result<Shipment> {
self.update_status(id, ShipmentStatus::Failed)
}
fn hold(&self, id: ShipmentId) -> Result<Shipment> {
self.update_status(id, ShipmentStatus::OnHold)
}
fn cancel(&self, id: ShipmentId) -> Result<Shipment> {
self.update_status(id, ShipmentStatus::Cancelled)
}
fn add_item(&self, shipment_id: ShipmentId, item: CreateShipmentItem) -> Result<ShipmentItem> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let id = Uuid::new_v4();
let now = Utc::now();
conn.execute(
"INSERT INTO shipment_items (id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
id.to_string(),
shipment_id.to_string(),
item.order_item_id.map(|u| u.to_string()),
item.product_id.map(|u| u.to_string()),
item.sku,
item.name,
item.quantity,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(ShipmentItem {
id,
shipment_id,
order_item_id: item.order_item_id,
product_id: item.product_id,
sku: item.sku,
name: item.name,
quantity: item.quantity,
created_at: now,
updated_at: now,
})
}
fn remove_item(&self, item_id: Uuid) -> Result<()> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
conn.execute("DELETE FROM shipment_items WHERE id = ?", [item_id.to_string()])
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(())
}
fn get_items(&self, shipment_id: ShipmentId) -> Result<Vec<ShipmentItem>> {
self.load_items(shipment_id)
}
fn add_event(&self, shipment_id: ShipmentId, event: AddShipmentEvent) -> Result<ShipmentEvent> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let id = Uuid::new_v4();
let now = Utc::now();
let event_time = event.event_time.unwrap_or(now);
conn.execute(
"INSERT INTO shipment_events (id, shipment_id, event_type, location, description, event_time, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
id.to_string(),
shipment_id.to_string(),
event.event_type,
event.location,
event.description,
event_time.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(ShipmentEvent {
id,
shipment_id,
event_type: event.event_type,
location: event.location,
description: event.description,
event_time,
created_at: now,
})
}
fn get_events(&self, shipment_id: ShipmentId) -> Result<Vec<ShipmentEvent>> {
self.load_events(shipment_id)
}
fn count(&self, filter: ShipmentFilter) -> Result<u64> {
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let mut sql = "SELECT COUNT(*) FROM shipments WHERE 1=1".to_string();
let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(order_id) = filter.order_id {
sql.push_str(" AND order_id = ?");
params.push(Box::new(order_id.to_string()));
}
if let Some(status) = filter.status {
sql.push_str(" AND status = ?");
params.push(Box::new(status.to_string()));
}
if let Some(carrier) = filter.carrier {
sql.push_str(" AND carrier = ?");
params.push(Box::new(carrier.to_string()));
}
let param_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(std::convert::AsRef::as_ref).collect();
let count: i64 = conn
.query_row(&sql, param_refs.as_slice(), |row| row.get(0))
.map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
Ok(count as u64)
}
fn create_batch(&self, inputs: Vec<CreateShipment>) -> Result<BatchResult<Shipment>> {
validate_batch_size(&inputs)?;
let mut result = BatchResult::with_capacity(inputs.len());
for (index, input) in inputs.into_iter().enumerate() {
match self.create(input) {
Ok(shipment) => result.record_success(shipment),
Err(e) => result.record_failure(index, None, &e),
}
}
Ok(result)
}
fn create_batch_atomic(&self, inputs: Vec<CreateShipment>) -> Result<Vec<Shipment>> {
validate_batch_size(&inputs)?;
if inputs.is_empty() {
return Ok(vec![]);
}
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 mut results = Vec::with_capacity(inputs.len());
for input in inputs {
let id = Uuid::new_v4();
let shipment_number = Shipment::generate_shipment_number();
let now = Utc::now();
let carrier = input.carrier.unwrap_or_default();
let method = input.shipping_method.unwrap_or_default();
let tracking_url =
input.tracking_number.as_ref().and_then(|tn| carrier.tracking_url(tn));
tx.execute(
"INSERT INTO shipments (id, shipment_number, order_id, status, carrier, shipping_method,
tracking_number, tracking_url, recipient_name, recipient_email, recipient_phone,
shipping_address, weight_kg, dimensions, shipping_cost, insurance_amount,
signature_required, estimated_delivery, notes, created_at, updated_at)
VALUES (?, ?, ?, 'pending', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
id.to_string(),
shipment_number,
input.order_id.to_string(),
carrier.to_string(),
method.to_string(),
input.tracking_number,
tracking_url,
input.recipient_name,
input.recipient_email,
input.recipient_phone,
input.shipping_address,
input.weight_kg.map(|w| w.to_string()),
input.dimensions,
input.shipping_cost.map(|c| c.to_string()),
input.insurance_amount.map(|a| a.to_string()),
i32::from(input.signature_required.unwrap_or(false)),
input.estimated_delivery.map(|dt| dt.to_rfc3339()),
input.notes,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(map_db_error)?;
let mut items = Vec::new();
if let Some(item_inputs) = &input.items {
for item_input in item_inputs {
let item_id = Uuid::new_v4();
tx.execute(
"INSERT INTO shipment_items (id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
rusqlite::params![
item_id.to_string(),
id.to_string(),
item_input.order_item_id.map(|u| u.to_string()),
item_input.product_id.map(|u| u.to_string()),
item_input.sku,
item_input.name,
item_input.quantity,
now.to_rfc3339(),
now.to_rfc3339(),
],
)
.map_err(map_db_error)?;
items.push(ShipmentItem {
id: item_id,
shipment_id: ShipmentId::from(id),
order_item_id: item_input.order_item_id,
product_id: item_input.product_id,
sku: item_input.sku.clone(),
name: item_input.name.clone(),
quantity: item_input.quantity,
created_at: now,
updated_at: now,
});
}
}
results.push(Shipment {
id: ShipmentId::from(id),
shipment_number,
order_id: input.order_id,
status: ShipmentStatus::Pending,
carrier,
shipping_method: method,
tracking_number: input.tracking_number,
tracking_url,
recipient_name: input.recipient_name,
recipient_email: input.recipient_email,
recipient_phone: input.recipient_phone,
shipping_address: input.shipping_address,
weight_kg: input.weight_kg,
dimensions: input.dimensions,
shipping_cost: input.shipping_cost,
insurance_amount: input.insurance_amount,
signature_required: input.signature_required.unwrap_or(false),
shipped_at: None,
estimated_delivery: input.estimated_delivery,
delivered_at: None,
notes: input.notes,
items,
events: vec![],
version: 1,
created_at: now,
updated_at: now,
});
}
tx.commit().map_err(map_db_error)?;
Ok(results)
}
fn update_batch(
&self,
updates: Vec<(ShipmentId, UpdateShipment)>,
) -> Result<BatchResult<Shipment>> {
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(shipment) => result.record_success(shipment),
Err(e) => result.record_failure(index, Some(id.to_string()), &e),
}
}
Ok(result)
}
fn update_batch_atomic(
&self,
updates: Vec<(ShipmentId, UpdateShipment)>,
) -> Result<Vec<Shipment>> {
validate_batch_size(&updates)?;
if updates.is_empty() {
return Ok(vec![]);
}
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 mut updated_ids = Vec::with_capacity(updates.len());
for (id, input) in updates {
let now = Utc::now();
type ShipmentExistingRow = (
String,
String,
Option<String>,
String,
Option<String>,
Option<String>,
String,
Option<String>,
Option<String>,
);
let existing_data: ShipmentExistingRow = tx
.query_row(
"SELECT carrier, shipping_method, tracking_number, recipient_name, recipient_email, recipient_phone, shipping_address, weight_kg, dimensions FROM shipments WHERE id = ?",
[id.to_string()],
|row| {
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
row.get(5)?,
row.get(6)?,
row.get(7)?,
row.get(8)?,
))
},
)
.map_err(|e| match e {
rusqlite::Error::QueryReturnedNoRows => CommerceError::NotFound,
e => map_db_error(e),
})?;
let existing_carrier: ShippingCarrier =
parse_enum(&existing_data.0, "shipment", "carrier")?;
let new_status = input.status.map(|s| s.to_string());
let new_carrier = input.carrier.unwrap_or(existing_carrier);
let new_tracking = input.tracking_number.or(existing_data.2);
let new_tracking_url =
new_tracking.as_ref().and_then(|tn| new_carrier.tracking_url(tn));
let new_recipient_name = input.recipient_name.unwrap_or(existing_data.3);
let new_recipient_email = input.recipient_email.or(existing_data.4);
let new_recipient_phone = input.recipient_phone.or(existing_data.5);
let new_shipping_address = input.shipping_address.unwrap_or(existing_data.6);
let new_weight = input.weight_kg.map(|w| w.to_string()).or(existing_data.7);
let new_dimensions = input.dimensions.or(existing_data.8);
let new_shipping_cost = input.shipping_cost.map(|c| c.to_string());
let new_estimated_delivery = input.estimated_delivery.map(|dt| dt.to_rfc3339());
let new_notes = input.notes;
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) = new_status {
update_parts.push("status = ?");
params.push(Box::new(status));
}
update_parts.push("carrier = ?");
params.push(Box::new(new_carrier.to_string()));
update_parts.push("tracking_number = ?");
params.push(Box::new(new_tracking));
update_parts.push("tracking_url = ?");
params.push(Box::new(new_tracking_url));
update_parts.push("recipient_name = ?");
params.push(Box::new(new_recipient_name));
update_parts.push("recipient_email = ?");
params.push(Box::new(new_recipient_email));
update_parts.push("recipient_phone = ?");
params.push(Box::new(new_recipient_phone));
update_parts.push("shipping_address = ?");
params.push(Box::new(new_shipping_address));
update_parts.push("weight_kg = ?");
params.push(Box::new(new_weight));
update_parts.push("dimensions = ?");
params.push(Box::new(new_dimensions));
if let Some(cost) = new_shipping_cost {
update_parts.push("shipping_cost = ?");
params.push(Box::new(cost));
}
if let Some(delivery) = new_estimated_delivery {
update_parts.push("estimated_delivery = ?");
params.push(Box::new(delivery));
}
if let Some(notes) = new_notes {
update_parts.push("notes = ?");
params.push(Box::new(notes));
}
params.push(Box::new(id.to_string()));
let sql = format!("UPDATE shipments SET {} WHERE id = ?", update_parts.join(", "));
let params_refs: Vec<&dyn rusqlite::ToSql> =
params.iter().map(std::convert::AsRef::as_ref).collect();
tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
updated_ids.push(id);
}
tx.commit().map_err(map_db_error)?;
let mut results = Vec::with_capacity(updated_ids.len());
for id in updated_ids {
if let Some(shipment) = self.get(id)? {
results.push(shipment);
}
}
Ok(results)
}
fn delete_batch(&self, ids: Vec<ShipmentId>) -> Result<BatchResult<Uuid>> {
validate_batch_size(&ids)?;
let mut result = BatchResult::with_capacity(ids.len());
for (index, id) in ids.into_iter().enumerate() {
let raw_id: Uuid = id.into();
match self.delete(id) {
Ok(()) => result.record_success(raw_id),
Err(e) => result.record_failure(index, Some(id.to_string()), &e),
}
}
Ok(result)
}
fn delete_batch_atomic(&self, ids: Vec<ShipmentId>) -> 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 raw_ids: Vec<Uuid> = ids.iter().map(|id| (*id).into()).collect();
let placeholders = build_in_clause(ids.len());
let params = uuid_params(&raw_ids);
let params_refs = params_refs(¶ms);
let sql = format!("DELETE FROM shipment_events WHERE shipment_id IN ({placeholders})");
tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
let sql = format!("DELETE FROM shipment_items WHERE shipment_id IN ({placeholders})");
tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
let now = Utc::now().to_rfc3339();
for id in &ids {
tx.execute(
"UPDATE shipments SET status = 'cancelled', updated_at = ? WHERE id = ?",
rusqlite::params![now, id.to_string()],
)
.map_err(map_db_error)?;
}
tx.commit().map_err(map_db_error)?;
Ok(())
}
fn get_batch(&self, ids: Vec<ShipmentId>) -> Result<Vec<Shipment>> {
validate_batch_size(&ids)?;
if ids.is_empty() {
return Ok(vec![]);
}
let conn = self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
let placeholders = build_in_clause(ids.len());
let sql = format!(
"SELECT id, shipment_number, order_id, status, carrier, shipping_method,
tracking_number, tracking_url, recipient_name, recipient_email, recipient_phone,
shipping_address, weight_kg, dimensions, shipping_cost, insurance_amount,
signature_required, shipped_at, estimated_delivery, delivered_at, notes,
created_at, updated_at
FROM shipments WHERE id IN ({placeholders})"
);
let raw_ids: Vec<Uuid> = ids.iter().map(|id| (*id).into()).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 rows = stmt
.query_map(params_refs.as_slice(), |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, String>(5)?,
row.get::<_, Option<String>>(6)?,
row.get::<_, Option<String>>(7)?,
row.get::<_, String>(8)?,
row.get::<_, Option<String>>(9)?,
row.get::<_, Option<String>>(10)?,
row.get::<_, String>(11)?,
row.get::<_, Option<String>>(12)?,
row.get::<_, Option<String>>(13)?,
row.get::<_, Option<String>>(14)?,
row.get::<_, Option<String>>(15)?,
row.get::<_, i32>(16)?,
row.get::<_, Option<String>>(17)?,
row.get::<_, Option<String>>(18)?,
row.get::<_, Option<String>>(19)?,
row.get::<_, Option<String>>(20)?,
row.get::<_, String>(21)?,
row.get::<_, String>(22)?,
))
})
.map_err(map_db_error)?;
let mut shipments = Vec::new();
for row in rows {
let (
id_str,
shipment_number,
order_id,
status,
carrier,
shipping_method,
tracking_number,
tracking_url,
recipient_name,
recipient_email,
recipient_phone,
shipping_address,
weight_kg,
dimensions,
shipping_cost,
insurance_amount,
signature_required,
shipped_at,
estimated_delivery,
delivered_at,
notes,
created_at,
updated_at,
) = row.map_err(map_db_error)?;
let shipment_id = ShipmentId::from(parse_uuid(&id_str, "shipment", "id")?);
let items = self.load_items(shipment_id)?;
let events = self.load_events(shipment_id)?;
shipments.push(Shipment {
id: shipment_id,
shipment_number,
order_id: OrderId::from(parse_uuid(&order_id, "shipment", "order_id")?),
status: parse_enum(&status, "shipment", "status")?,
carrier: parse_enum(&carrier, "shipment", "carrier")?,
shipping_method: parse_enum(&shipping_method, "shipment", "shipping_method")?,
tracking_number,
tracking_url,
recipient_name,
recipient_email,
recipient_phone,
shipping_address,
weight_kg: parse_decimal_opt(weight_kg, "shipment", "weight_kg")?,
dimensions,
shipping_cost: parse_decimal_opt(shipping_cost, "shipment", "shipping_cost")?,
insurance_amount: parse_decimal_opt(
insurance_amount,
"shipment",
"insurance_amount",
)?,
signature_required: signature_required != 0,
shipped_at: parse_datetime_opt(shipped_at, "shipment", "shipped_at")?,
estimated_delivery: parse_datetime_opt(
estimated_delivery,
"shipment",
"estimated_delivery",
)?,
delivered_at: parse_datetime_opt(delivered_at, "shipment", "delivered_at")?,
notes,
items,
events,
version: 1,
created_at: parse_datetime(&created_at, "shipment", "created_at")?,
updated_at: parse_datetime(&updated_at, "shipment", "updated_at")?,
});
}
Ok(shipments)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::SqliteDatabase;
use rust_decimal_macros::dec;
use stateset_core::{
CreateShipment, OrderId, ShipmentFilter, ShipmentRepository, ShipmentStatus,
ShippingCarrier, ShippingMethod,
};
fn fresh_repo() -> SqliteShipmentRepository {
SqliteDatabase::in_memory().expect("in-memory").shipments()
}
fn make_shipment(repo: &SqliteShipmentRepository, tracking: Option<&str>) -> Shipment {
repo.create(CreateShipment {
order_id: OrderId::new(),
carrier: Some(ShippingCarrier::Ups),
shipping_method: Some(ShippingMethod::Ground),
tracking_number: tracking.map(String::from),
recipient_name: "Ada Lovelace".into(),
recipient_email: Some("ada@example.com".into()),
recipient_phone: None,
shipping_address: "1 Babbage Way, London".into(),
weight_kg: Some(dec!(2.5)),
dimensions: Some("30x20x10cm".into()),
shipping_cost: Some(dec!(8.99)),
insurance_amount: None,
signature_required: Some(false),
estimated_delivery: None,
notes: None,
items: None,
})
.expect("create shipment")
}
#[test]
fn create_shipment_round_trips() {
let repo = fresh_repo();
let s = make_shipment(&repo, Some("1Z9999"));
assert_eq!(s.recipient_name, "Ada Lovelace");
assert_eq!(s.tracking_number.as_deref(), Some("1Z9999"));
assert_eq!(s.carrier, ShippingCarrier::Ups);
assert!(!s.shipment_number.is_empty());
let by_id = repo.get(s.id).expect("ok").expect("found");
assert_eq!(by_id.id, s.id);
let by_num = repo.get_by_number(&s.shipment_number).expect("ok").expect("found");
assert_eq!(by_num.id, s.id);
assert!(repo.get_by_number("missing").expect("ok").is_none());
}
#[test]
fn get_by_tracking_finds_shipment() {
let repo = fresh_repo();
let s = make_shipment(&repo, Some("TRACK-XYZ"));
let by_track = repo.get_by_tracking("TRACK-XYZ").expect("ok").expect("found");
assert_eq!(by_track.id, s.id);
assert!(repo.get_by_tracking("missing").expect("ok").is_none());
}
#[test]
fn list_filters_by_status() {
let repo = fresh_repo();
let pending = make_shipment(&repo, Some("P1"));
let to_cancel = make_shipment(&repo, Some("P2"));
repo.cancel(to_cancel.id).expect("cancel");
let pendings = repo
.list(ShipmentFilter { status: Some(ShipmentStatus::Pending), ..Default::default() })
.expect("pending");
let cancelleds = repo
.list(ShipmentFilter { status: Some(ShipmentStatus::Cancelled), ..Default::default() })
.expect("cancelled");
assert!(pendings.iter().any(|s| s.id == pending.id));
assert!(cancelleds.iter().any(|s| s.id == to_cancel.id));
}
#[test]
fn list_filters_by_carrier() {
let repo = fresh_repo();
make_shipment(&repo, Some("UPS-1"));
make_shipment(&repo, Some("UPS-2"));
repo.create(CreateShipment {
order_id: OrderId::new(),
carrier: Some(ShippingCarrier::FedEx),
shipping_method: Some(ShippingMethod::Express),
tracking_number: Some("FEDEX-1".into()),
recipient_name: "Test".into(),
recipient_email: None,
recipient_phone: None,
shipping_address: "123 Test St".into(),
weight_kg: None,
dimensions: None,
shipping_cost: None,
insurance_amount: None,
signature_required: None,
estimated_delivery: None,
notes: None,
items: None,
})
.expect("fedex");
let ups = repo
.list(ShipmentFilter { carrier: Some(ShippingCarrier::Ups), ..Default::default() })
.expect("ups");
assert!(ups.iter().all(|s| s.carrier == ShippingCarrier::Ups));
assert!(ups.len() >= 2);
}
#[test]
fn cancel_transitions_to_cancelled() {
let repo = fresh_repo();
let s = make_shipment(&repo, Some("CANCEL-1"));
let cancelled = repo.cancel(s.id).expect("cancel");
assert_eq!(cancelled.status, ShipmentStatus::Cancelled);
}
#[test]
fn get_items_returns_empty_for_shipment_without_items() {
let repo = fresh_repo();
let s = make_shipment(&repo, Some("NO-ITEMS"));
let items = repo.get_items(s.id).expect("items");
assert!(items.is_empty());
}
#[test]
fn get_events_returns_at_most_one_initial_event() {
let repo = fresh_repo();
let s = make_shipment(&repo, Some("NO-EVENTS"));
let events = repo.get_events(s.id).expect("events");
assert!(events.len() <= 1);
}
#[test]
fn create_batch_returns_per_input_results() {
let repo = fresh_repo();
let result = repo
.create_batch(vec![
CreateShipment {
order_id: OrderId::new(),
carrier: Some(ShippingCarrier::Ups),
shipping_method: Some(ShippingMethod::Ground),
tracking_number: Some("B1".into()),
recipient_name: "X".into(),
recipient_email: None,
recipient_phone: None,
shipping_address: "addr".into(),
weight_kg: None,
dimensions: None,
shipping_cost: None,
insurance_amount: None,
signature_required: None,
estimated_delivery: None,
notes: None,
items: None,
},
CreateShipment {
order_id: OrderId::new(),
carrier: Some(ShippingCarrier::FedEx),
shipping_method: Some(ShippingMethod::Express),
tracking_number: Some("B2".into()),
recipient_name: "Y".into(),
recipient_email: None,
recipient_phone: None,
shipping_address: "addr".into(),
weight_kg: None,
dimensions: None,
shipping_cost: None,
insurance_amount: None,
signature_required: None,
estimated_delivery: None,
notes: None,
items: None,
},
])
.expect("batch");
assert_eq!(result.success_count, 2);
assert_eq!(result.failure_count, 0);
}
#[test]
fn get_unknown_id_returns_none() {
let repo = fresh_repo();
assert!(repo.get(stateset_core::ShipmentId::new()).expect("ok").is_none());
}
}