1use super::{
4 map_db_error, parse_datetime_opt_row, parse_datetime_row, parse_decimal_row, parse_enum_row,
5 parse_uuid_row, with_immediate_transaction,
6};
7use chrono::Utc;
8use r2d2::Pool;
9use r2d2_sqlite::SqliteConnectionManager;
10use rusqlite::OptionalExtension;
11use rust_decimal::Decimal;
12use stateset_core::{
13 CommerceError, CreateTransferOrder, Result, TransferOrder, TransferOrderFilter,
14 TransferOrderId, TransferOrderItem, TransferOrderItemId, TransferOrderStatus,
15};
16
17#[derive(Debug)]
18pub struct SqliteTransferOrderRepository {
19 pool: Pool<SqliteConnectionManager>,
20}
21
22impl SqliteTransferOrderRepository {
23 #[must_use]
24 pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
25 Self { pool }
26 }
27
28 fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
29 self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
30 }
31
32 fn row_to_item(row: &rusqlite::Row<'_>) -> rusqlite::Result<TransferOrderItem> {
33 Ok(TransferOrderItem {
34 id: parse_uuid_row(&row.get::<_, String>("id")?, "transfer_order_item", "id")?.into(),
35 transfer_order_id: parse_uuid_row(
36 &row.get::<_, String>("transfer_order_id")?,
37 "transfer_order_item",
38 "transfer_order_id",
39 )?
40 .into(),
41 product_id: parse_uuid_row(
42 &row.get::<_, String>("product_id")?,
43 "transfer_order_item",
44 "product_id",
45 )?
46 .into(),
47 sku: row.get("sku")?,
48 quantity: parse_decimal_row(
49 &row.get::<_, String>("quantity")?,
50 "transfer_order_item",
51 "quantity",
52 )?,
53 quantity_shipped: parse_decimal_row(
54 &row.get::<_, String>("quantity_shipped")?,
55 "transfer_order_item",
56 "quantity_shipped",
57 )?,
58 quantity_received: parse_decimal_row(
59 &row.get::<_, String>("quantity_received")?,
60 "transfer_order_item",
61 "quantity_received",
62 )?,
63 })
64 }
65
66 fn row_to_order_head(row: &rusqlite::Row<'_>) -> rusqlite::Result<TransferOrder> {
67 Ok(TransferOrder {
68 id: parse_uuid_row(&row.get::<_, String>("id")?, "transfer_order", "id")?.into(),
69 number: row.get("number")?,
70 source_warehouse_id: parse_uuid_row(
71 &row.get::<_, String>("source_warehouse_id")?,
72 "transfer_order",
73 "source_warehouse_id",
74 )?
75 .into(),
76 destination_warehouse_id: parse_uuid_row(
77 &row.get::<_, String>("destination_warehouse_id")?,
78 "transfer_order",
79 "destination_warehouse_id",
80 )?
81 .into(),
82 status: parse_enum_row::<TransferOrderStatus>(
83 &row.get::<_, String>("status")?,
84 "transfer_order",
85 "status",
86 )?,
87 items: Vec::new(),
88 expected_at: parse_datetime_opt_row(
89 row.get::<_, Option<String>>("expected_at")?,
90 "transfer_order",
91 "expected_at",
92 )?,
93 shipped_at: parse_datetime_opt_row(
94 row.get::<_, Option<String>>("shipped_at")?,
95 "transfer_order",
96 "shipped_at",
97 )?,
98 received_at: parse_datetime_opt_row(
99 row.get::<_, Option<String>>("received_at")?,
100 "transfer_order",
101 "received_at",
102 )?,
103 notes: row.get("notes")?,
104 created_at: parse_datetime_row(
105 &row.get::<_, String>("created_at")?,
106 "transfer_order",
107 "created_at",
108 )?,
109 updated_at: parse_datetime_row(
110 &row.get::<_, String>("updated_at")?,
111 "transfer_order",
112 "updated_at",
113 )?,
114 })
115 }
116
117 fn load_items(
118 conn: &rusqlite::Connection,
119 order_id: &str,
120 ) -> rusqlite::Result<Vec<TransferOrderItem>> {
121 let mut stmt = conn.prepare(
122 "SELECT * FROM transfer_order_items WHERE transfer_order_id = ? ORDER BY sku",
123 )?;
124 let items = stmt
125 .query_map([order_id], Self::row_to_item)?
126 .collect::<std::result::Result<Vec<_>, _>>()?;
127 Ok(items)
128 }
129
130 fn load_items_batch(
131 conn: &rusqlite::Connection,
132 ids: &[String],
133 ) -> rusqlite::Result<std::collections::HashMap<String, Vec<TransferOrderItem>>> {
134 let mut map: std::collections::HashMap<String, Vec<TransferOrderItem>> =
135 std::collections::HashMap::with_capacity(ids.len());
136 for chunk in ids.chunks(500) {
137 let placeholders = super::build_in_clause(chunk.len());
138 let sql = format!(
139 "SELECT * FROM transfer_order_items WHERE transfer_order_id IN ({placeholders}) ORDER BY sku"
140 );
141 let mut stmt = conn.prepare(&sql)?;
142 let param_refs: Vec<&dyn rusqlite::types::ToSql> =
143 chunk.iter().map(|s| s as &dyn rusqlite::types::ToSql).collect();
144 let rows = stmt.query_map(param_refs.as_slice(), |row| {
145 let parent: String = row.get("transfer_order_id")?;
146 Ok((parent, Self::row_to_item(row)?))
147 })?;
148 for row in rows {
149 let (parent, item) = row?;
150 map.entry(parent).or_default().push(item);
151 }
152 }
153 Ok(map)
154 }
155
156 fn load_full(conn: &rusqlite::Connection, id: &str) -> rusqlite::Result<TransferOrder> {
157 let mut order = conn.query_row(
158 "SELECT * FROM transfer_orders WHERE id = ?",
159 [id],
160 Self::row_to_order_head,
161 )?;
162 order.items = Self::load_items(conn, id)?;
163 Ok(order)
164 }
165}
166
167impl stateset_core::TransferOrderRepository for SqliteTransferOrderRepository {
168 fn create(&self, input: CreateTransferOrder) -> Result<TransferOrder> {
169 if input.source_warehouse_id == input.destination_warehouse_id {
170 return Err(CommerceError::ValidationError(
171 "source and destination warehouses must differ".into(),
172 ));
173 }
174 if input.items.is_empty() {
175 return Err(CommerceError::ValidationError(
176 "a transfer order requires at least one item".into(),
177 ));
178 }
179 let id = TransferOrderId::new();
180 let id_str = id.to_string();
181 let now = Utc::now();
182 let now_str = now.to_rfc3339();
183 let number = format!("TO-{}", &id_str[..8]);
185
186 with_immediate_transaction(&self.pool, |tx| {
187 tx.execute(
188 "INSERT INTO transfer_orders (id, number, source_warehouse_id, destination_warehouse_id, status, expected_at, notes, created_at, updated_at)
189 VALUES (?, ?, ?, ?, 'draft', ?, ?, ?, ?)",
190 rusqlite::params![
191 &id_str,
192 &number,
193 input.source_warehouse_id.to_string(),
194 input.destination_warehouse_id.to_string(),
195 input.expected_at.map(|d| d.to_rfc3339()),
196 &input.notes,
197 &now_str,
198 &now_str,
199 ],
200 )?;
201
202 for item in &input.items {
203 let item_id = TransferOrderItemId::new().to_string();
204 tx.execute(
205 "INSERT INTO transfer_order_items (id, transfer_order_id, product_id, sku, quantity, quantity_shipped, quantity_received)
206 VALUES (?, ?, ?, '', ?, '0', '0')",
207 rusqlite::params![
208 &item_id,
209 &id_str,
210 item.product_id.to_string(),
211 item.quantity.to_string(),
212 ],
213 )?;
214 }
215
216 Self::load_full(tx, &id_str)
217 })
218 }
219
220 fn get(&self, id: TransferOrderId) -> Result<Option<TransferOrder>> {
221 let conn = self.conn()?;
222 match Self::load_full(&conn, &id.to_string()) {
223 Ok(o) => Ok(Some(o)),
224 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
225 Err(e) => Err(map_db_error(e)),
226 }
227 }
228
229 fn list(&self, filter: TransferOrderFilter) -> Result<Vec<TransferOrder>> {
230 let conn = self.conn()?;
231 let mut sql = "SELECT * FROM transfer_orders WHERE 1=1".to_string();
232 let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
233 if let Some(status) = filter.status {
234 sql.push_str(" AND status = ?");
235 params.push(Box::new(status.to_string()));
236 }
237 if let Some(src) = filter.source_warehouse_id {
238 sql.push_str(" AND source_warehouse_id = ?");
239 params.push(Box::new(src.to_string()));
240 }
241 if let Some(dest) = filter.destination_warehouse_id {
242 sql.push_str(" AND destination_warehouse_id = ?");
243 params.push(Box::new(dest.to_string()));
244 }
245 sql.push_str(" ORDER BY created_at DESC");
246 crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
247
248 let param_refs: Vec<&dyn rusqlite::types::ToSql> =
249 params.iter().map(|p| p.as_ref()).collect();
250 let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
251 let heads = stmt
252 .query_map(param_refs.as_slice(), Self::row_to_order_head)
253 .map_err(map_db_error)?
254 .collect::<std::result::Result<Vec<_>, _>>()
255 .map_err(map_db_error)?;
256 let ids: Vec<String> = heads.iter().map(|h| h.id.to_string()).collect();
257 let mut items_by_id = Self::load_items_batch(&conn, &ids).map_err(map_db_error)?;
258 let mut out = Vec::with_capacity(heads.len());
259 for mut head in heads {
260 head.items = items_by_id.remove(&head.id.to_string()).unwrap_or_default();
261 out.push(head);
262 }
263 Ok(out)
264 }
265
266 fn ship(&self, id: TransferOrderId) -> Result<TransferOrder> {
267 let id_str = id.to_string();
268 let now = Utc::now().to_rfc3339();
269 with_immediate_transaction(&self.pool, |tx| {
270 tx.execute(
272 "UPDATE transfer_order_items SET quantity_shipped = quantity WHERE transfer_order_id = ?",
273 [&id_str],
274 )?;
275 tx.execute(
276 "UPDATE transfer_orders SET status = 'in_transit', shipped_at = ?, updated_at = ? WHERE id = ?",
277 rusqlite::params![&now, &now, &id_str],
278 )?;
279 Self::load_full(tx, &id_str)
280 })
281 }
282
283 fn receive_line(
284 &self,
285 id: TransferOrderId,
286 item_id: TransferOrderItemId,
287 quantity: Decimal,
288 ) -> Result<TransferOrder> {
289 if quantity <= Decimal::ZERO {
290 return Err(CommerceError::ValidationError("receive quantity must be positive".into()));
291 }
292 let id_str = id.to_string();
293 let item_str = item_id.to_string();
294 let now = Utc::now().to_rfc3339();
295 with_immediate_transaction(&self.pool, |tx| {
296 let row: Option<(String, String)> = tx
297 .query_row(
298 "SELECT quantity, quantity_received FROM transfer_order_items WHERE id = ? AND transfer_order_id = ?",
299 rusqlite::params![&item_str, &id_str],
300 |r| Ok((r.get(0)?, r.get(1)?)),
301 )
302 .optional()?;
303 let Some((expected, current)) = row else {
304 return Err(rusqlite::Error::QueryReturnedNoRows);
305 };
306 let expected: Decimal = expected.parse().unwrap_or(Decimal::ZERO);
307 let current: Decimal = current.parse().unwrap_or(Decimal::ZERO);
308 let new_received = current + quantity;
309 if new_received > expected {
310 return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
311 CommerceError::ValidationError(format!(
312 "receiving {quantity} would exceed the {expected} expected on this line ({current} already received)"
313 )),
314 )));
315 }
316 tx.execute(
317 "UPDATE transfer_order_items SET quantity_received = ? WHERE id = ?",
318 rusqlite::params![new_received.to_string(), &item_str],
319 )?;
320
321 let order = Self::load_full(tx, &id_str)?;
323 let derived = order.derive_receipt_status();
324 let received_at =
325 if derived == TransferOrderStatus::Received { Some(now.clone()) } else { None };
326 tx.execute(
327 "UPDATE transfer_orders SET status = ?, received_at = COALESCE(?, received_at), updated_at = ? WHERE id = ?",
328 rusqlite::params![derived.to_string(), received_at, &now, &id_str],
329 )?;
330 Self::load_full(tx, &id_str)
331 })
332 }
333
334 fn cancel(&self, id: TransferOrderId) -> Result<TransferOrder> {
335 let id_str = id.to_string();
336 let now = Utc::now().to_rfc3339();
337 let conn = self.conn()?;
338 let status: String = conn
339 .query_row("SELECT status FROM transfer_orders WHERE id = ?", [&id_str], |row| {
340 row.get(0)
341 })
342 .map_err(|e| match e {
343 rusqlite::Error::QueryReturnedNoRows => CommerceError::NotFound,
344 other => map_db_error(other),
345 })?;
346 drop(conn);
347 if matches!(status.as_str(), "received" | "cancelled") {
348 return Err(CommerceError::ValidationError(format!(
349 "Cannot cancel a transfer order in status {status}"
350 )));
351 }
352 with_immediate_transaction(&self.pool, |tx| {
353 tx.execute(
354 "UPDATE transfer_orders SET status = 'cancelled', updated_at = ? WHERE id = ?",
355 rusqlite::params![&now, &id_str],
356 )?;
357 Self::load_full(tx, &id_str)
358 })
359 }
360}
361
362#[cfg(test)]
363mod tests {
364 use super::*;
365 use crate::DatabaseConfig;
366 use crate::sqlite::SqliteDatabase;
367 use rust_decimal_macros::dec;
368 use stateset_core::{CreateTransferOrderItem, ProductId, TransferOrderRepository, WarehouseId};
369
370 fn test_repo() -> SqliteTransferOrderRepository {
371 let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
372 SqliteTransferOrderRepository::new(db.pool().clone())
373 }
374
375 fn new_order(repo: &SqliteTransferOrderRepository) -> TransferOrder {
376 repo.create(CreateTransferOrder {
377 source_warehouse_id: WarehouseId::new(),
378 destination_warehouse_id: WarehouseId::new(),
379 items: vec![CreateTransferOrderItem {
380 product_id: ProductId::new(),
381 quantity: dec!(10),
382 }],
383 expected_at: None,
384 notes: Some("restock".into()),
385 })
386 .expect("create order")
387 }
388
389 #[test]
390 fn create_rejects_same_warehouse() {
391 let repo = test_repo();
392 let w = WarehouseId::new();
393 let res = repo.create(CreateTransferOrder {
394 source_warehouse_id: w,
395 destination_warehouse_id: w,
396 items: vec![CreateTransferOrderItem {
397 product_id: ProductId::new(),
398 quantity: dec!(1),
399 }],
400 expected_at: None,
401 notes: None,
402 });
403 assert!(res.is_err());
404 }
405
406 #[test]
407 fn create_and_get_with_items() {
408 let repo = test_repo();
409 let o = new_order(&repo);
410 assert_eq!(o.status, TransferOrderStatus::Draft);
411 assert_eq!(o.items.len(), 1);
412 let fetched = repo.get(o.id).expect("get").expect("found");
413 assert_eq!(fetched.total_quantity(), dec!(10));
414 }
415
416 #[test]
417 fn ship_then_receive_transitions_status() {
418 let repo = test_repo();
419 let o = new_order(&repo);
420 let shipped = repo.ship(o.id).expect("ship");
421 assert_eq!(shipped.status, TransferOrderStatus::InTransit);
422 assert_eq!(shipped.items[0].quantity_shipped, dec!(10));
423
424 let item_id = shipped.items[0].id;
425 let partial = repo.receive_line(o.id, item_id, dec!(4)).expect("receive partial");
426 assert_eq!(partial.status, TransferOrderStatus::PartiallyReceived);
427
428 let full = repo.receive_line(o.id, item_id, dec!(6)).expect("receive rest");
429 assert_eq!(full.status, TransferOrderStatus::Received);
430 assert!(full.received_at.is_some());
431 }
432
433 #[test]
434 fn receive_line_rejects_over_receipt() {
435 let repo = test_repo();
436 let o = new_order(&repo);
437 let shipped = repo.ship(o.id).expect("ship");
438 let item_id = shipped.items[0].id;
439 assert!(repo.receive_line(o.id, item_id, dec!(11)).is_err());
441 assert!(repo.receive_line(o.id, item_id, dec!(0)).is_err());
443 repo.receive_line(o.id, item_id, dec!(7)).expect("receive 7");
445 assert!(repo.receive_line(o.id, item_id, dec!(4)).is_err());
446 let full = repo.receive_line(o.id, item_id, dec!(3)).expect("receive rest");
448 assert_eq!(full.status, TransferOrderStatus::Received);
449 }
450
451 #[test]
452 fn cancel_sets_status() {
453 let repo = test_repo();
454 let o = new_order(&repo);
455 let cancelled = repo.cancel(o.id).expect("cancel");
456 assert_eq!(cancelled.status, TransferOrderStatus::Cancelled);
457
458 let err = repo.cancel(o.id).expect_err("already cancelled");
460 assert!(matches!(err, CommerceError::ValidationError(_)));
461 }
462
463 #[test]
464 fn list_filters_by_status() {
465 let repo = test_repo();
466 let a = new_order(&repo);
467 new_order(&repo);
468 repo.cancel(a.id).expect("cancel");
469 let cancelled = repo
470 .list(TransferOrderFilter {
471 status: Some(TransferOrderStatus::Cancelled),
472 ..Default::default()
473 })
474 .expect("list");
475 assert_eq!(cancelled.len(), 1);
476 }
477}