use super::map_db_error;
use chrono::{DateTime, Utc};
use rust_decimal::Decimal;
use sqlx::FromRow;
use sqlx::postgres::PgPool;
use stateset_core::{
AddShipmentEvent, BatchResult, CommerceError, CreateShipment, CreateShipmentItem, OrderId,
ProductId, Result, Shipment, ShipmentEvent, ShipmentFilter, ShipmentId, ShipmentItem,
ShipmentRepository, ShipmentStatus, ShippingCarrier, ShippingMethod, UpdateShipment,
validate_batch_size,
};
use uuid::Uuid;
#[derive(Debug, Clone)]
pub struct PgShipmentRepository {
pool: PgPool,
}
#[derive(FromRow)]
struct ShipmentRow {
id: Uuid,
shipment_number: String,
order_id: Uuid,
status: String,
carrier: String,
shipping_method: String,
tracking_number: Option<String>,
tracking_url: Option<String>,
recipient_name: String,
recipient_email: Option<String>,
recipient_phone: Option<String>,
shipping_address: String,
weight_kg: Option<Decimal>,
dimensions: Option<String>,
shipping_cost: Option<Decimal>,
insurance_amount: Option<Decimal>,
signature_required: bool,
shipped_at: Option<DateTime<Utc>>,
estimated_delivery: Option<DateTime<Utc>>,
delivered_at: Option<DateTime<Utc>>,
notes: Option<String>,
version: i32,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
}
#[derive(FromRow)]
struct ShipmentItemRow {
id: Uuid,
shipment_id: Uuid,
order_item_id: Option<Uuid>,
product_id: Option<Uuid>,
sku: String,
name: String,
quantity: i32,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
}
#[derive(FromRow)]
struct ShipmentEventRow {
id: Uuid,
shipment_id: Uuid,
event_type: String,
location: Option<String>,
description: Option<String>,
event_time: DateTime<Utc>,
created_at: DateTime<Utc>,
}
impl PgShipmentRepository {
pub const fn new(pool: PgPool) -> Self {
Self { pool }
}
fn row_to_shipment(
row: ShipmentRow,
items: Vec<ShipmentItem>,
events: Vec<ShipmentEvent>,
) -> Result<Shipment> {
let ShipmentRow {
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,
version,
created_at,
updated_at,
} = row;
let status: ShipmentStatus = status.parse().map_err(|e| {
CommerceError::DatabaseError(format!(
"Invalid shipment.status '{}': {}",
status.as_str(),
e
))
})?;
let carrier: ShippingCarrier = carrier.parse().map_err(|e| {
CommerceError::DatabaseError(format!(
"Invalid shipment.carrier '{}': {}",
carrier.as_str(),
e
))
})?;
let shipping_method: ShippingMethod = shipping_method.parse().map_err(|e| {
CommerceError::DatabaseError(format!(
"Invalid shipment.shipping_method '{}': {}",
shipping_method.as_str(),
e
))
})?;
Ok(Shipment {
id: ShipmentId::from(id),
shipment_number,
order_id: OrderId::from(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,
items,
events,
version,
created_at,
updated_at,
})
}
fn row_to_item(row: ShipmentItemRow) -> ShipmentItem {
ShipmentItem {
id: row.id,
shipment_id: ShipmentId::from(row.shipment_id),
order_item_id: row.order_item_id,
product_id: row.product_id.map(ProductId::from),
sku: row.sku,
name: row.name,
quantity: row.quantity,
created_at: row.created_at,
updated_at: row.updated_at,
}
}
fn row_to_event(row: ShipmentEventRow) -> ShipmentEvent {
ShipmentEvent {
id: row.id,
shipment_id: ShipmentId::from(row.shipment_id),
event_type: row.event_type,
location: row.location,
description: row.description,
event_time: row.event_time,
created_at: row.created_at,
}
}
async fn load_items_async(&self, shipment_id: Uuid) -> Result<Vec<ShipmentItem>> {
let rows = sqlx::query_as::<_, ShipmentItemRow>(
"SELECT id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at
FROM shipment_items WHERE shipment_id = $1"
)
.bind(shipment_id)
.fetch_all(&self.pool)
.await
.map_err(map_db_error)?;
Ok(rows.into_iter().map(Self::row_to_item).collect())
}
async fn load_events_async(&self, shipment_id: Uuid) -> Result<Vec<ShipmentEvent>> {
let rows = sqlx::query_as::<_, ShipmentEventRow>(
"SELECT id, shipment_id, event_type, location, description, event_time, created_at
FROM shipment_events WHERE shipment_id = $1 ORDER BY event_time DESC",
)
.bind(shipment_id)
.fetch_all(&self.pool)
.await
.map_err(map_db_error)?;
Ok(rows.into_iter().map(Self::row_to_event).collect())
}
async fn load_items_batch_async(
&self,
ids: &[Uuid],
) -> Result<std::collections::HashMap<Uuid, Vec<ShipmentItem>>> {
let mut map: std::collections::HashMap<Uuid, Vec<ShipmentItem>> =
std::collections::HashMap::with_capacity(ids.len());
if ids.is_empty() {
return Ok(map);
}
let rows = sqlx::query_as::<_, ShipmentItemRow>(
"SELECT id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at
FROM shipment_items WHERE shipment_id = ANY($1)"
)
.bind(ids.to_vec())
.fetch_all(&self.pool)
.await
.map_err(map_db_error)?;
for row in rows {
let parent = row.shipment_id;
map.entry(parent).or_default().push(Self::row_to_item(row));
}
Ok(map)
}
async fn load_events_batch_async(
&self,
ids: &[Uuid],
) -> Result<std::collections::HashMap<Uuid, Vec<ShipmentEvent>>> {
let mut map: std::collections::HashMap<Uuid, Vec<ShipmentEvent>> =
std::collections::HashMap::with_capacity(ids.len());
if ids.is_empty() {
return Ok(map);
}
let rows = sqlx::query_as::<_, ShipmentEventRow>(
"SELECT id, shipment_id, event_type, location, description, event_time, created_at
FROM shipment_events WHERE shipment_id = ANY($1) ORDER BY event_time DESC",
)
.bind(ids.to_vec())
.fetch_all(&self.pool)
.await
.map_err(map_db_error)?;
for row in rows {
let parent = row.shipment_id;
map.entry(parent).or_default().push(Self::row_to_event(row));
}
Ok(map)
}
async fn update_status_async(&self, id: Uuid, status: ShipmentStatus) -> Result<Shipment> {
let now = Utc::now();
sqlx::query("UPDATE shipments SET status = $1, updated_at = $2 WHERE id = $3")
.bind(status.to_string())
.bind(now)
.bind(id)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
self.get_async(id).await?.ok_or(CommerceError::NotFound)
}
pub async fn create_async(&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 tx = self.pool.begin().await.map_err(map_db_error)?;
sqlx::query(
"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 ($1, $2, $3, 'pending', $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)"
)
.bind(id)
.bind(&shipment_number)
.bind(input.order_id.into_uuid())
.bind(carrier.to_string())
.bind(method.to_string())
.bind(&input.tracking_number)
.bind(&tracking_url)
.bind(&input.recipient_name)
.bind(&input.recipient_email)
.bind(&input.recipient_phone)
.bind(&input.shipping_address)
.bind(input.weight_kg)
.bind(&input.dimensions)
.bind(input.shipping_cost)
.bind(input.insurance_amount)
.bind(input.signature_required.unwrap_or(false))
.bind(input.estimated_delivery)
.bind(&input.notes)
.bind(now)
.bind(now)
.execute(tx.as_mut())
.await
.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();
sqlx::query(
"INSERT INTO shipment_items (id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)"
)
.bind(item_id)
.bind(id)
.bind(item_input.order_item_id)
.bind(item_input.product_id.map(|pid| pid.into_uuid()))
.bind(&item_input.sku)
.bind(&item_input.name)
.bind(item_input.quantity)
.bind(now)
.bind(now)
.execute(tx.as_mut())
.await
.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,
});
}
}
tx.commit().await.map_err(map_db_error)?;
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,
})
}
pub async fn get_async(&self, id: Uuid) -> Result<Option<Shipment>> {
let row = sqlx::query_as::<_, ShipmentRow>(
"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,
version, created_at, updated_at
FROM shipments WHERE id = $1",
)
.bind(id)
.fetch_optional(&self.pool)
.await
.map_err(map_db_error)?;
match row {
Some(row) => {
let items = self.load_items_async(row.id).await?;
let events = self.load_events_async(row.id).await?;
Ok(Some(Self::row_to_shipment(row, items, events)?))
}
None => Ok(None),
}
}
pub async fn get_by_number_async(&self, shipment_number: &str) -> Result<Option<Shipment>> {
let id: Option<(Uuid,)> =
sqlx::query_as("SELECT id FROM shipments WHERE shipment_number = $1")
.bind(shipment_number)
.fetch_optional(&self.pool)
.await
.map_err(map_db_error)?;
match id {
Some((id,)) => self.get_async(id).await,
None => Ok(None),
}
}
pub async fn get_by_tracking_async(&self, tracking_number: &str) -> Result<Option<Shipment>> {
let id: Option<(Uuid,)> =
sqlx::query_as("SELECT id FROM shipments WHERE tracking_number = $1")
.bind(tracking_number)
.fetch_optional(&self.pool)
.await
.map_err(map_db_error)?;
match id {
Some((id,)) => self.get_async(id).await,
None => Ok(None),
}
}
pub async fn update_async(&self, id: Uuid, input: UpdateShipment) -> Result<Shipment> {
let existing = self.get_async(id).await?.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);
sqlx::query(
"UPDATE shipments SET status = $1, carrier = $2, tracking_number = $3, tracking_url = $4,
recipient_name = $5, recipient_email = $6, recipient_phone = $7, shipping_address = $8,
weight_kg = $9, dimensions = $10, shipping_cost = $11, estimated_delivery = $12, notes = $13,
updated_at = $14 WHERE id = $15"
)
.bind(new_status.to_string())
.bind(new_carrier.to_string())
.bind(&new_tracking)
.bind(&new_tracking_url)
.bind(&new_recipient_name)
.bind(&new_recipient_email)
.bind(&new_recipient_phone)
.bind(&new_shipping_address)
.bind(new_weight)
.bind(&new_dimensions)
.bind(new_shipping_cost)
.bind(new_estimated_delivery)
.bind(&new_notes)
.bind(now)
.bind(id)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
self.get_async(id).await?.ok_or(CommerceError::NotFound)
}
pub async fn list_async(&self, filter: ShipmentFilter) -> Result<Vec<Shipment>> {
let limit = super::effective_limit(filter.limit);
let offset = filter.offset.unwrap_or(0) as i64;
let mut query = String::from("SELECT id FROM shipments WHERE 1=1");
let mut param_idx = 1;
if filter.order_id.is_some() {
query.push_str(&format!(" AND order_id = ${}", param_idx));
param_idx += 1;
}
if filter.status.is_some() {
query.push_str(&format!(" AND status = ${}", param_idx));
param_idx += 1;
}
if filter.carrier.is_some() {
query.push_str(&format!(" AND carrier = ${}", param_idx));
param_idx += 1;
}
if filter.tracking_number.is_some() {
query.push_str(&format!(" AND tracking_number = ${}", param_idx));
param_idx += 1;
}
query.push_str(&format!(
" ORDER BY created_at DESC LIMIT ${} OFFSET ${}",
param_idx,
param_idx + 1
));
let mut q = sqlx::query_as::<_, (Uuid,)>(&query);
if let Some(order_id) = filter.order_id {
q = q.bind(order_id.into_uuid());
}
if let Some(status) = filter.status {
q = q.bind(status.to_string());
}
if let Some(carrier) = filter.carrier {
q = q.bind(carrier.to_string());
}
if let Some(tracking_number) = &filter.tracking_number {
q = q.bind(tracking_number);
}
q = q.bind(limit).bind(offset);
let ids = q.fetch_all(&self.pool).await.map_err(map_db_error)?;
let id_list: Vec<Uuid> = ids.into_iter().map(|(id,)| id).collect();
if id_list.is_empty() {
return Ok(Vec::new());
}
let rows = sqlx::query_as::<_, ShipmentRow>(
"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,
version, created_at, updated_at
FROM shipments WHERE id = ANY($1)",
)
.bind(&id_list)
.fetch_all(&self.pool)
.await
.map_err(map_db_error)?;
let mut rows_by_id: std::collections::HashMap<Uuid, ShipmentRow> =
rows.into_iter().map(|r| (r.id, r)).collect();
let mut items_by_id = self.load_items_batch_async(&id_list).await?;
let mut events_by_id = self.load_events_batch_async(&id_list).await?;
let mut shipments = Vec::with_capacity(id_list.len());
for id in id_list {
if let Some(row) = rows_by_id.remove(&id) {
let items = items_by_id.remove(&id).unwrap_or_default();
let events = events_by_id.remove(&id).unwrap_or_default();
shipments.push(Self::row_to_shipment(row, items, events)?);
}
}
Ok(shipments)
}
pub async fn for_order_async(&self, order_id: Uuid) -> Result<Vec<Shipment>> {
self.list_async(ShipmentFilter {
order_id: Some(OrderId::from(order_id)),
..Default::default()
})
.await
}
pub async fn delete_async(&self, id: Uuid) -> Result<()> {
sqlx::query("UPDATE shipments SET status = 'cancelled', updated_at = $1 WHERE id = $2")
.bind(Utc::now())
.bind(id)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
Ok(())
}
pub async fn mark_processing_async(&self, id: Uuid) -> Result<Shipment> {
self.update_status_async(id, ShipmentStatus::Processing).await
}
pub async fn mark_ready_async(&self, id: Uuid) -> Result<Shipment> {
self.update_status_async(id, ShipmentStatus::ReadyToShip).await
}
pub async fn ship_async(&self, id: Uuid, tracking_number: Option<String>) -> Result<Shipment> {
let existing = self.get_async(id).await?.ok_or(CommerceError::NotFound)?;
let now = Utc::now();
let tracking_url =
tracking_number.as_ref().and_then(|tn| existing.carrier.tracking_url(tn));
sqlx::query(
"UPDATE shipments SET status = 'shipped', tracking_number = COALESCE($1, tracking_number),
tracking_url = COALESCE($2, tracking_url), shipped_at = $3, updated_at = $4 WHERE id = $5"
)
.bind(&tracking_number)
.bind(&tracking_url)
.bind(now)
.bind(now)
.bind(id)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
self.get_async(id).await?.ok_or(CommerceError::NotFound)
}
pub async fn mark_in_transit_async(&self, id: Uuid) -> Result<Shipment> {
self.update_status_async(id, ShipmentStatus::InTransit).await
}
pub async fn mark_out_for_delivery_async(&self, id: Uuid) -> Result<Shipment> {
self.update_status_async(id, ShipmentStatus::OutForDelivery).await
}
pub async fn mark_delivered_async(&self, id: Uuid) -> Result<Shipment> {
let now = Utc::now();
sqlx::query("UPDATE shipments SET status = 'delivered', delivered_at = $1, updated_at = $2 WHERE id = $3")
.bind(now)
.bind(now)
.bind(id)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
self.get_async(id).await?.ok_or(CommerceError::NotFound)
}
pub async fn mark_failed_async(&self, id: Uuid) -> Result<Shipment> {
self.update_status_async(id, ShipmentStatus::Failed).await
}
pub async fn hold_async(&self, id: Uuid) -> Result<Shipment> {
self.update_status_async(id, ShipmentStatus::OnHold).await
}
pub async fn cancel_async(&self, id: Uuid) -> Result<Shipment> {
self.update_status_async(id, ShipmentStatus::Cancelled).await
}
pub async fn add_item_async(
&self,
shipment_id: Uuid,
item: CreateShipmentItem,
) -> Result<ShipmentItem> {
let id = Uuid::new_v4();
let now = Utc::now();
sqlx::query(
"INSERT INTO shipment_items (id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)"
)
.bind(id)
.bind(shipment_id)
.bind(item.order_item_id)
.bind(item.product_id.map(|pid| pid.into_uuid()))
.bind(&item.sku)
.bind(&item.name)
.bind(item.quantity)
.bind(now)
.bind(now)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
Ok(ShipmentItem {
id,
shipment_id: ShipmentId::from(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,
})
}
pub async fn remove_item_async(&self, item_id: Uuid) -> Result<()> {
sqlx::query("DELETE FROM shipment_items WHERE id = $1")
.bind(item_id)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
Ok(())
}
pub async fn get_items_async(&self, shipment_id: Uuid) -> Result<Vec<ShipmentItem>> {
self.load_items_async(shipment_id).await
}
pub async fn add_event_async(
&self,
shipment_id: Uuid,
event: AddShipmentEvent,
) -> Result<ShipmentEvent> {
let id = Uuid::new_v4();
let now = Utc::now();
let event_time = event.event_time.unwrap_or(now);
sqlx::query(
"INSERT INTO shipment_events (id, shipment_id, event_type, location, description, event_time, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)"
)
.bind(id)
.bind(shipment_id)
.bind(&event.event_type)
.bind(&event.location)
.bind(&event.description)
.bind(event_time)
.bind(now)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
Ok(ShipmentEvent {
id,
shipment_id: ShipmentId::from(shipment_id),
event_type: event.event_type,
location: event.location,
description: event.description,
event_time,
created_at: now,
})
}
pub async fn get_events_async(&self, shipment_id: Uuid) -> Result<Vec<ShipmentEvent>> {
self.load_events_async(shipment_id).await
}
pub async fn count_async(&self, filter: ShipmentFilter) -> Result<u64> {
let mut query = String::from("SELECT COUNT(*) FROM shipments WHERE 1=1");
let mut param_idx = 1;
if filter.order_id.is_some() {
query.push_str(&format!(" AND order_id = ${}", param_idx));
param_idx += 1;
}
if filter.status.is_some() {
query.push_str(&format!(" AND status = ${}", param_idx));
param_idx += 1;
}
if filter.carrier.is_some() {
query.push_str(&format!(" AND carrier = ${}", param_idx));
}
let mut q = sqlx::query_as::<_, (i64,)>(&query);
if let Some(order_id) = filter.order_id {
q = q.bind(order_id.into_uuid());
}
if let Some(status) = filter.status {
q = q.bind(status.to_string());
}
if let Some(carrier) = filter.carrier {
q = q.bind(carrier.to_string());
}
let (count,) = q.fetch_one(&self.pool).await.map_err(map_db_error)?;
Ok(count as u64)
}
pub async fn create_batch_async(
&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_async(input).await {
Ok(shipment) => result.record_success(shipment),
Err(e) => result.record_failure(index, None, &e),
}
}
Ok(result)
}
pub async fn create_batch_atomic_async(
&self,
inputs: Vec<CreateShipment>,
) -> Result<Vec<Shipment>> {
validate_batch_size(&inputs)?;
let mut tx = self.pool.begin().await.map_err(map_db_error)?;
let mut shipments = 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));
sqlx::query(
"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 ($1, $2, $3, 'pending', $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)"
)
.bind(id)
.bind(&shipment_number)
.bind(input.order_id.into_uuid())
.bind(carrier.to_string())
.bind(method.to_string())
.bind(&input.tracking_number)
.bind(&tracking_url)
.bind(&input.recipient_name)
.bind(&input.recipient_email)
.bind(&input.recipient_phone)
.bind(&input.shipping_address)
.bind(input.weight_kg)
.bind(&input.dimensions)
.bind(input.shipping_cost)
.bind(input.insurance_amount)
.bind(input.signature_required.unwrap_or(false))
.bind(input.estimated_delivery)
.bind(&input.notes)
.bind(now)
.bind(now)
.execute(tx.as_mut())
.await
.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();
sqlx::query(
"INSERT INTO shipment_items (id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)"
)
.bind(item_id)
.bind(id)
.bind(item_input.order_item_id)
.bind(item_input.product_id.map(|pid| pid.into_uuid()))
.bind(&item_input.sku)
.bind(&item_input.name)
.bind(item_input.quantity)
.bind(now)
.bind(now)
.execute(tx.as_mut())
.await
.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,
});
}
}
shipments.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().await.map_err(map_db_error)?;
Ok(shipments)
}
pub async fn update_batch_async(
&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() {
let raw_id = id.into_uuid();
match self.update_async(raw_id, input).await {
Ok(shipment) => result.record_success(shipment),
Err(e) => result.record_failure(index, Some(raw_id.to_string()), &e),
}
}
Ok(result)
}
pub async fn update_batch_atomic_async(
&self,
updates: Vec<(ShipmentId, UpdateShipment)>,
) -> Result<Vec<Shipment>> {
validate_batch_size(&updates)?;
let mut tx = self.pool.begin().await.map_err(map_db_error)?;
let mut shipments = Vec::with_capacity(updates.len());
let now = Utc::now();
for (id, input) in updates {
let raw_id = id.into_uuid();
let existing_row = sqlx::query_as::<_, ShipmentRow>(
"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,
version, created_at, updated_at
FROM shipments WHERE id = $1 FOR UPDATE"
)
.bind(raw_id)
.fetch_optional(tx.as_mut())
.await
.map_err(map_db_error)?
.ok_or(CommerceError::NotFound)?;
let existing_status: ShipmentStatus = existing_row.status.parse().map_err(|e| {
CommerceError::DatabaseError(format!(
"Invalid shipment.status '{}': {}",
existing_row.status.as_str(),
e
))
})?;
let existing_carrier: ShippingCarrier = existing_row.carrier.parse().map_err(|e| {
CommerceError::DatabaseError(format!(
"Invalid shipment.carrier '{}': {}",
existing_row.carrier.as_str(),
e
))
})?;
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_row.tracking_number.clone());
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_row.recipient_name.clone());
let new_recipient_email =
input.recipient_email.or(existing_row.recipient_email.clone());
let new_recipient_phone =
input.recipient_phone.or(existing_row.recipient_phone.clone());
let new_shipping_address =
input.shipping_address.unwrap_or(existing_row.shipping_address.clone());
let new_weight = input.weight_kg.or(existing_row.weight_kg);
let new_dimensions = input.dimensions.or(existing_row.dimensions.clone());
let new_shipping_cost = input.shipping_cost.or(existing_row.shipping_cost);
let new_estimated_delivery =
input.estimated_delivery.or(existing_row.estimated_delivery);
let new_notes = input.notes.or(existing_row.notes.clone());
sqlx::query(
"UPDATE shipments SET status = $1, carrier = $2, tracking_number = $3, tracking_url = $4,
recipient_name = $5, recipient_email = $6, recipient_phone = $7, shipping_address = $8,
weight_kg = $9, dimensions = $10, shipping_cost = $11, estimated_delivery = $12, notes = $13,
updated_at = $14 WHERE id = $15"
)
.bind(new_status.to_string())
.bind(new_carrier.to_string())
.bind(&new_tracking)
.bind(&new_tracking_url)
.bind(&new_recipient_name)
.bind(&new_recipient_email)
.bind(&new_recipient_phone)
.bind(&new_shipping_address)
.bind(new_weight)
.bind(&new_dimensions)
.bind(new_shipping_cost)
.bind(new_estimated_delivery)
.bind(&new_notes)
.bind(now)
.bind(raw_id)
.execute(tx.as_mut())
.await
.map_err(map_db_error)?;
let updated_row = sqlx::query_as::<_, ShipmentRow>(
"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,
version, created_at, updated_at
FROM shipments WHERE id = $1"
)
.bind(raw_id)
.fetch_one(tx.as_mut())
.await
.map_err(map_db_error)?;
let item_rows = sqlx::query_as::<_, ShipmentItemRow>(
"SELECT id, shipment_id, order_item_id, product_id, sku, name, quantity, created_at, updated_at
FROM shipment_items WHERE shipment_id = $1"
)
.bind(raw_id)
.fetch_all(tx.as_mut())
.await
.map_err(map_db_error)?;
let event_rows = sqlx::query_as::<_, ShipmentEventRow>(
"SELECT id, shipment_id, event_type, location, description, event_time, created_at
FROM shipment_events WHERE shipment_id = $1 ORDER BY event_time DESC",
)
.bind(raw_id)
.fetch_all(tx.as_mut())
.await
.map_err(map_db_error)?;
let items: Vec<ShipmentItem> = item_rows.into_iter().map(Self::row_to_item).collect();
let events: Vec<ShipmentEvent> =
event_rows.into_iter().map(Self::row_to_event).collect();
shipments.push(Self::row_to_shipment(updated_row, items, events)?);
}
tx.commit().await.map_err(map_db_error)?;
Ok(shipments)
}
pub async fn delete_batch_async(&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 = id.into_uuid();
match self.delete_async(raw_id).await {
Ok(()) => result.record_success(raw_id),
Err(e) => result.record_failure(index, Some(raw_id.to_string()), &e),
}
}
Ok(result)
}
pub async fn delete_batch_atomic_async(&self, ids: Vec<ShipmentId>) -> Result<()> {
validate_batch_size(&ids)?;
let raw_ids: Vec<Uuid> = ids.into_iter().map(|id| id.into_uuid()).collect();
let mut tx = self.pool.begin().await.map_err(map_db_error)?;
sqlx::query("DELETE FROM shipment_events WHERE shipment_id = ANY($1)")
.bind(&raw_ids)
.execute(tx.as_mut())
.await
.map_err(map_db_error)?;
sqlx::query("DELETE FROM shipment_items WHERE shipment_id = ANY($1)")
.bind(&raw_ids)
.execute(tx.as_mut())
.await
.map_err(map_db_error)?;
sqlx::query(
"UPDATE shipments SET status = 'cancelled', updated_at = $1 WHERE id = ANY($2)",
)
.bind(Utc::now())
.bind(&raw_ids)
.execute(tx.as_mut())
.await
.map_err(map_db_error)?;
tx.commit().await.map_err(map_db_error)?;
Ok(())
}
pub async fn get_batch_async(&self, ids: Vec<ShipmentId>) -> Result<Vec<Shipment>> {
validate_batch_size(&ids)?;
let raw_ids: Vec<Uuid> = ids.into_iter().map(|id| id.into_uuid()).collect();
let rows = sqlx::query_as::<_, ShipmentRow>(
"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,
version, created_at, updated_at
FROM shipments WHERE id = ANY($1)",
)
.bind(&raw_ids)
.fetch_all(&self.pool)
.await
.map_err(map_db_error)?;
let mut items_by_id = self.load_items_batch_async(&raw_ids).await?;
let mut events_by_id = self.load_events_batch_async(&raw_ids).await?;
let mut shipments = Vec::with_capacity(rows.len());
for row in rows {
let items = items_by_id.remove(&row.id).unwrap_or_default();
let events = events_by_id.remove(&row.id).unwrap_or_default();
shipments.push(Self::row_to_shipment(row, items, events)?);
}
Ok(shipments)
}
pub fn create_batch_sync(&self, inputs: Vec<CreateShipment>) -> Result<BatchResult<Shipment>> {
super::block_on(self.create_batch_async(inputs))
}
pub fn create_batch_atomic_sync(&self, inputs: Vec<CreateShipment>) -> Result<Vec<Shipment>> {
super::block_on(self.create_batch_atomic_async(inputs))
}
pub fn update_batch_sync(
&self,
updates: Vec<(ShipmentId, UpdateShipment)>,
) -> Result<BatchResult<Shipment>> {
super::block_on(self.update_batch_async(updates))
}
pub fn update_batch_atomic_sync(
&self,
updates: Vec<(ShipmentId, UpdateShipment)>,
) -> Result<Vec<Shipment>> {
super::block_on(self.update_batch_atomic_async(updates))
}
pub fn delete_batch_sync(&self, ids: Vec<ShipmentId>) -> Result<BatchResult<Uuid>> {
super::block_on(self.delete_batch_async(ids))
}
pub fn delete_batch_atomic_sync(&self, ids: Vec<ShipmentId>) -> Result<()> {
super::block_on(self.delete_batch_atomic_async(ids))
}
pub fn get_batch_sync(&self, ids: Vec<ShipmentId>) -> Result<Vec<Shipment>> {
super::block_on(self.get_batch_async(ids))
}
}
impl ShipmentRepository for PgShipmentRepository {
fn create(&self, input: CreateShipment) -> Result<Shipment> {
super::block_on(self.create_async(input))
}
fn get(&self, id: ShipmentId) -> Result<Option<Shipment>> {
super::block_on(self.get_async(id.into_uuid()))
}
fn get_by_number(&self, shipment_number: &str) -> Result<Option<Shipment>> {
super::block_on(self.get_by_number_async(shipment_number))
}
fn get_by_tracking(&self, tracking_number: &str) -> Result<Option<Shipment>> {
super::block_on(self.get_by_tracking_async(tracking_number))
}
fn update(&self, id: ShipmentId, input: UpdateShipment) -> Result<Shipment> {
super::block_on(self.update_async(id.into_uuid(), input))
}
fn list(&self, filter: ShipmentFilter) -> Result<Vec<Shipment>> {
super::block_on(self.list_async(filter))
}
fn for_order(&self, order_id: OrderId) -> Result<Vec<Shipment>> {
super::block_on(self.for_order_async(order_id.into_uuid()))
}
fn delete(&self, id: ShipmentId) -> Result<()> {
super::block_on(self.delete_async(id.into_uuid()))
}
fn mark_processing(&self, id: ShipmentId) -> Result<Shipment> {
super::block_on(self.mark_processing_async(id.into_uuid()))
}
fn mark_ready(&self, id: ShipmentId) -> Result<Shipment> {
super::block_on(self.mark_ready_async(id.into_uuid()))
}
fn ship(&self, id: ShipmentId, tracking_number: Option<String>) -> Result<Shipment> {
super::block_on(self.ship_async(id.into_uuid(), tracking_number))
}
fn mark_in_transit(&self, id: ShipmentId) -> Result<Shipment> {
super::block_on(self.mark_in_transit_async(id.into_uuid()))
}
fn mark_out_for_delivery(&self, id: ShipmentId) -> Result<Shipment> {
super::block_on(self.mark_out_for_delivery_async(id.into_uuid()))
}
fn mark_delivered(&self, id: ShipmentId) -> Result<Shipment> {
super::block_on(self.mark_delivered_async(id.into_uuid()))
}
fn mark_failed(&self, id: ShipmentId) -> Result<Shipment> {
super::block_on(self.mark_failed_async(id.into_uuid()))
}
fn hold(&self, id: ShipmentId) -> Result<Shipment> {
super::block_on(self.hold_async(id.into_uuid()))
}
fn cancel(&self, id: ShipmentId) -> Result<Shipment> {
super::block_on(self.cancel_async(id.into_uuid()))
}
fn add_item(&self, shipment_id: ShipmentId, item: CreateShipmentItem) -> Result<ShipmentItem> {
super::block_on(self.add_item_async(shipment_id.into_uuid(), item))
}
fn remove_item(&self, item_id: Uuid) -> Result<()> {
super::block_on(self.remove_item_async(item_id))
}
fn get_items(&self, shipment_id: ShipmentId) -> Result<Vec<ShipmentItem>> {
super::block_on(self.get_items_async(shipment_id.into_uuid()))
}
fn add_event(&self, shipment_id: ShipmentId, event: AddShipmentEvent) -> Result<ShipmentEvent> {
super::block_on(self.add_event_async(shipment_id.into_uuid(), event))
}
fn get_events(&self, shipment_id: ShipmentId) -> Result<Vec<ShipmentEvent>> {
super::block_on(self.get_events_async(shipment_id.into_uuid()))
}
fn count(&self, filter: ShipmentFilter) -> Result<u64> {
super::block_on(self.count_async(filter))
}
fn create_batch(&self, inputs: Vec<CreateShipment>) -> Result<BatchResult<Shipment>> {
self.create_batch_sync(inputs)
}
fn create_batch_atomic(&self, inputs: Vec<CreateShipment>) -> Result<Vec<Shipment>> {
self.create_batch_atomic_sync(inputs)
}
fn update_batch(
&self,
updates: Vec<(ShipmentId, UpdateShipment)>,
) -> Result<BatchResult<Shipment>> {
self.update_batch_sync(updates)
}
fn update_batch_atomic(
&self,
updates: Vec<(ShipmentId, UpdateShipment)>,
) -> Result<Vec<Shipment>> {
self.update_batch_atomic_sync(updates)
}
fn delete_batch(&self, ids: Vec<ShipmentId>) -> Result<BatchResult<Uuid>> {
self.delete_batch_sync(ids)
}
fn delete_batch_atomic(&self, ids: Vec<ShipmentId>) -> Result<()> {
self.delete_batch_atomic_sync(ids)
}
fn get_batch(&self, ids: Vec<ShipmentId>) -> Result<Vec<Shipment>> {
self.get_batch_sync(ids)
}
}