stateset_db/sqlite/
transfer_orders.rs

1//! SQLite implementation of the transfer order repository
2
3use 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        // Human-readable number derived from the timestamp + short id fragment.
184        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            // Shipping sets quantity_shipped = quantity on each line.
271            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            // Recompute order status from line receipts.
322            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        // Line expects 10; receiving 11 at once is rejected.
440        assert!(repo.receive_line(o.id, item_id, dec!(11)).is_err());
441        // Non-positive quantities are rejected.
442        assert!(repo.receive_line(o.id, item_id, dec!(0)).is_err());
443        // After receiving 7, receiving another 4 (total 11) is rejected.
444        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        // Exact remaining (3) still succeeds.
447        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        // Terminal-state guard: cancelling again is rejected.
459        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}
stateset_db/sqlite/transfer_orders.rs

stateset_db/sqlite/
transfer_orders.rs
