stateset_db/sqlite/
inbound_shipments.rs

1//! SQLite implementation of the inbound shipment repository
2
3use super::{
4    map_db_error, parse_datetime_opt_row, parse_datetime_row, parse_decimal_row, parse_enum_row,
5    parse_uuid_opt_row, 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, CreateInboundShipment, InboundShipment, InboundShipmentFilter,
14    InboundShipmentId, InboundShipmentItem, InboundShipmentItemId, InboundShipmentStatus, Result,
15};
16
17#[derive(Debug)]
18pub struct SqliteInboundShipmentRepository {
19    pool: Pool<SqliteConnectionManager>,
20}
21
22impl SqliteInboundShipmentRepository {
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<InboundShipmentItem> {
33        Ok(InboundShipmentItem {
34            id: parse_uuid_row(&row.get::<_, String>("id")?, "inbound_shipment_item", "id")?.into(),
35            inbound_shipment_id: parse_uuid_row(
36                &row.get::<_, String>("inbound_shipment_id")?,
37                "inbound_shipment_item",
38                "inbound_shipment_id",
39            )?
40            .into(),
41            product_id: parse_uuid_row(
42                &row.get::<_, String>("product_id")?,
43                "inbound_shipment_item",
44                "product_id",
45            )?
46            .into(),
47            sku: row.get("sku")?,
48            quantity_expected: parse_decimal_row(
49                &row.get::<_, String>("quantity_expected")?,
50                "inbound_shipment_item",
51                "quantity_expected",
52            )?,
53            quantity_received: parse_decimal_row(
54                &row.get::<_, String>("quantity_received")?,
55                "inbound_shipment_item",
56                "quantity_received",
57            )?,
58        })
59    }
60
61    fn row_to_head(row: &rusqlite::Row<'_>) -> rusqlite::Result<InboundShipment> {
62        Ok(InboundShipment {
63            id: parse_uuid_row(&row.get::<_, String>("id")?, "inbound_shipment", "id")?.into(),
64            number: row.get("number")?,
65            supplier_id: parse_uuid_row(
66                &row.get::<_, String>("supplier_id")?,
67                "inbound_shipment",
68                "supplier_id",
69            )?,
70            purchase_order_id: parse_uuid_opt_row(
71                row.get::<_, Option<String>>("purchase_order_id")?,
72                "inbound_shipment",
73                "purchase_order_id",
74            )?,
75            warehouse_id: parse_uuid_opt_row(
76                row.get::<_, Option<String>>("warehouse_id")?,
77                "inbound_shipment",
78                "warehouse_id",
79            )?
80            .map(Into::into),
81            carrier: row.get("carrier")?,
82            tracking_number: row.get("tracking_number")?,
83            status: parse_enum_row::<InboundShipmentStatus>(
84                &row.get::<_, String>("status")?,
85                "inbound_shipment",
86                "status",
87            )?,
88            items: Vec::new(),
89            expected_at: parse_datetime_opt_row(
90                row.get::<_, Option<String>>("expected_at")?,
91                "inbound_shipment",
92                "expected_at",
93            )?,
94            received_at: parse_datetime_opt_row(
95                row.get::<_, Option<String>>("received_at")?,
96                "inbound_shipment",
97                "received_at",
98            )?,
99            notes: row.get("notes")?,
100            created_at: parse_datetime_row(
101                &row.get::<_, String>("created_at")?,
102                "inbound_shipment",
103                "created_at",
104            )?,
105            updated_at: parse_datetime_row(
106                &row.get::<_, String>("updated_at")?,
107                "inbound_shipment",
108                "updated_at",
109            )?,
110        })
111    }
112
113    fn load_items(
114        conn: &rusqlite::Connection,
115        id: &str,
116    ) -> rusqlite::Result<Vec<InboundShipmentItem>> {
117        let mut stmt = conn.prepare(
118            "SELECT * FROM inbound_shipment_items WHERE inbound_shipment_id = ? ORDER BY sku",
119        )?;
120        stmt.query_map([id], Self::row_to_item)?.collect()
121    }
122
123    fn load_items_batch(
124        conn: &rusqlite::Connection,
125        ids: &[String],
126    ) -> rusqlite::Result<std::collections::HashMap<String, Vec<InboundShipmentItem>>> {
127        let mut map: std::collections::HashMap<String, Vec<InboundShipmentItem>> =
128            std::collections::HashMap::with_capacity(ids.len());
129        for chunk in ids.chunks(500) {
130            let placeholders = super::build_in_clause(chunk.len());
131            let sql = format!(
132                "SELECT * FROM inbound_shipment_items WHERE inbound_shipment_id IN ({placeholders}) ORDER BY sku"
133            );
134            let mut stmt = conn.prepare(&sql)?;
135            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
136                chunk.iter().map(|s| s as &dyn rusqlite::types::ToSql).collect();
137            let rows = stmt.query_map(param_refs.as_slice(), |row| {
138                let parent: String = row.get("inbound_shipment_id")?;
139                Ok((parent, Self::row_to_item(row)?))
140            })?;
141            for row in rows {
142                let (parent, item) = row?;
143                map.entry(parent).or_default().push(item);
144            }
145        }
146        Ok(map)
147    }
148
149    fn load_full(conn: &rusqlite::Connection, id: &str) -> rusqlite::Result<InboundShipment> {
150        let mut head = conn.query_row(
151            "SELECT * FROM inbound_shipments WHERE id = ?",
152            [id],
153            Self::row_to_head,
154        )?;
155        head.items = Self::load_items(conn, id)?;
156        Ok(head)
157    }
158
159    fn set_status(
160        &self,
161        id: InboundShipmentId,
162        status: InboundShipmentStatus,
163    ) -> Result<InboundShipment> {
164        let id_str = id.to_string();
165        let now = Utc::now().to_rfc3339();
166        with_immediate_transaction(&self.pool, |tx| {
167            tx.execute(
168                "UPDATE inbound_shipments SET status = ?, updated_at = ? WHERE id = ?",
169                rusqlite::params![status.to_string(), &now, &id_str],
170            )?;
171            Self::load_full(tx, &id_str)
172        })
173    }
174}
175
176impl stateset_core::InboundShipmentRepository for SqliteInboundShipmentRepository {
177    fn create(&self, input: CreateInboundShipment) -> Result<InboundShipment> {
178        if input.items.is_empty() {
179            return Err(CommerceError::ValidationError(
180                "an inbound shipment requires at least one item".into(),
181            ));
182        }
183        let id = InboundShipmentId::new();
184        let id_str = id.to_string();
185        let now = Utc::now().to_rfc3339();
186        let number = format!("ASN-{}", &id_str[..8]);
187        with_immediate_transaction(&self.pool, |tx| {
188            tx.execute(
189                "INSERT INTO inbound_shipments (id, number, supplier_id, purchase_order_id, warehouse_id, carrier, tracking_number, status, expected_at, notes, created_at, updated_at)
190                 VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?, ?, ?)",
191                rusqlite::params![
192                    &id_str,
193                    &number,
194                    input.supplier_id.to_string(),
195                    input.purchase_order_id.map(|p| p.to_string()),
196                    input.warehouse_id.map(|w| w.to_string()),
197                    &input.carrier,
198                    &input.tracking_number,
199                    input.expected_at.map(|d| d.to_rfc3339()),
200                    &input.notes,
201                    &now,
202                    &now,
203                ],
204            )?;
205            for item in &input.items {
206                tx.execute(
207                    "INSERT INTO inbound_shipment_items (id, inbound_shipment_id, product_id, sku, quantity_expected, quantity_received)
208                     VALUES (?, ?, ?, ?, ?, '0')",
209                    rusqlite::params![
210                        InboundShipmentItemId::new().to_string(),
211                        &id_str,
212                        item.product_id.to_string(),
213                        &item.sku,
214                        item.quantity_expected.to_string(),
215                    ],
216                )?;
217            }
218            Self::load_full(tx, &id_str)
219        })
220    }
221
222    fn get(&self, id: InboundShipmentId) -> Result<Option<InboundShipment>> {
223        let conn = self.conn()?;
224        match Self::load_full(&conn, &id.to_string()) {
225            Ok(s) => Ok(Some(s)),
226            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
227            Err(e) => Err(map_db_error(e)),
228        }
229    }
230
231    fn list(&self, filter: InboundShipmentFilter) -> Result<Vec<InboundShipment>> {
232        let conn = self.conn()?;
233        let mut sql = "SELECT * FROM inbound_shipments WHERE 1=1".to_string();
234        let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
235        if let Some(supplier) = filter.supplier_id {
236            sql.push_str(" AND supplier_id = ?");
237            params.push(Box::new(supplier.to_string()));
238        }
239        if let Some(warehouse) = filter.warehouse_id {
240            sql.push_str(" AND warehouse_id = ?");
241            params.push(Box::new(warehouse.to_string()));
242        }
243        if let Some(status) = filter.status {
244            sql.push_str(" AND status = ?");
245            params.push(Box::new(status.to_string()));
246        }
247        sql.push_str(" ORDER BY created_at DESC");
248        crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
249        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
250            params.iter().map(|p| p.as_ref()).collect();
251        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
252        let heads = stmt
253            .query_map(param_refs.as_slice(), Self::row_to_head)
254            .map_err(map_db_error)?
255            .collect::<std::result::Result<Vec<_>, _>>()
256            .map_err(map_db_error)?;
257        let ids: Vec<String> = heads.iter().map(|h| h.id.to_string()).collect();
258        let mut items_by_id = Self::load_items_batch(&conn, &ids).map_err(map_db_error)?;
259        let mut out = Vec::with_capacity(heads.len());
260        for mut head in heads {
261            head.items = items_by_id.remove(&head.id.to_string()).unwrap_or_default();
262            out.push(head);
263        }
264        Ok(out)
265    }
266
267    fn mark_in_transit(&self, id: InboundShipmentId) -> Result<InboundShipment> {
268        self.set_status(id, InboundShipmentStatus::InTransit)
269    }
270
271    fn mark_arrived(&self, id: InboundShipmentId) -> Result<InboundShipment> {
272        self.set_status(id, InboundShipmentStatus::Arrived)
273    }
274
275    fn receive_line(
276        &self,
277        id: InboundShipmentId,
278        item_id: InboundShipmentItemId,
279        quantity: Decimal,
280    ) -> Result<InboundShipment> {
281        if quantity <= Decimal::ZERO {
282            return Err(CommerceError::ValidationError("receive quantity must be positive".into()));
283        }
284        let id_str = id.to_string();
285        let item_str = item_id.to_string();
286        let now = Utc::now().to_rfc3339();
287        with_immediate_transaction(&self.pool, |tx| {
288            let row: Option<(String, String)> = tx
289                .query_row(
290                    "SELECT quantity_expected, quantity_received FROM inbound_shipment_items WHERE id = ? AND inbound_shipment_id = ?",
291                    rusqlite::params![&item_str, &id_str],
292                    |r| Ok((r.get(0)?, r.get(1)?)),
293                )
294                .optional()?;
295            let Some((expected, current)) = row else {
296                return Err(rusqlite::Error::QueryReturnedNoRows);
297            };
298            let expected: Decimal = expected.parse().unwrap_or(Decimal::ZERO);
299            let new_received: Decimal =
300                current.parse::<Decimal>().unwrap_or(Decimal::ZERO) + quantity;
301            if new_received > expected {
302                return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
303                    CommerceError::ValidationError(format!(
304                        "receiving {quantity} would exceed the {expected} expected on this line"
305                    )),
306                )));
307            }
308            tx.execute(
309                "UPDATE inbound_shipment_items SET quantity_received = ? WHERE id = ?",
310                rusqlite::params![new_received.to_string(), &item_str],
311            )?;
312
313            let shipment = Self::load_full(tx, &id_str)?;
314            let derived = shipment.derive_receipt_status();
315            let received_at =
316                if derived == InboundShipmentStatus::Received { Some(now.clone()) } else { None };
317            tx.execute(
318                "UPDATE inbound_shipments SET status = ?, received_at = COALESCE(?, received_at), updated_at = ? WHERE id = ?",
319                rusqlite::params![derived.to_string(), received_at, &now, &id_str],
320            )?;
321            Self::load_full(tx, &id_str)
322        })
323    }
324
325    fn cancel(&self, id: InboundShipmentId) -> Result<InboundShipment> {
326        let current = self.get(id)?.ok_or(CommerceError::NotFound)?;
327        if matches!(
328            current.status,
329            InboundShipmentStatus::Received | InboundShipmentStatus::Cancelled
330        ) {
331            return Err(CommerceError::ValidationError(format!(
332                "Cannot cancel an inbound shipment in status {}",
333                current.status
334            )));
335        }
336        self.set_status(id, InboundShipmentStatus::Cancelled)
337    }
338}
339
340#[cfg(test)]
341mod tests {
342    use super::*;
343    use crate::DatabaseConfig;
344    use crate::sqlite::SqliteDatabase;
345    use rust_decimal_macros::dec;
346    use stateset_core::{CreateInboundShipmentItem, InboundShipmentRepository, ProductId};
347    use uuid::Uuid;
348
349    fn test_repo() -> SqliteInboundShipmentRepository {
350        let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
351        SqliteInboundShipmentRepository::new(db.pool().clone())
352    }
353
354    fn new_shipment(repo: &SqliteInboundShipmentRepository) -> InboundShipment {
355        repo.create(CreateInboundShipment {
356            supplier_id: Uuid::new_v4(),
357            purchase_order_id: None,
358            warehouse_id: None,
359            carrier: Some("DHL".into()),
360            tracking_number: Some("1Z999".into()),
361            expected_at: None,
362            items: vec![CreateInboundShipmentItem {
363                product_id: ProductId::new(),
364                sku: "SKU-1".into(),
365                quantity_expected: dec!(10),
366            }],
367            notes: None,
368        })
369        .expect("create shipment")
370    }
371
372    #[test]
373    fn create_rejects_empty_items() {
374        let repo = test_repo();
375        let res = repo.create(CreateInboundShipment {
376            supplier_id: Uuid::new_v4(),
377            purchase_order_id: None,
378            warehouse_id: None,
379            carrier: None,
380            tracking_number: None,
381            expected_at: None,
382            items: vec![],
383            notes: None,
384        });
385        assert!(res.is_err());
386    }
387
388    #[test]
389    fn create_and_get() {
390        let repo = test_repo();
391        let s = new_shipment(&repo);
392        assert_eq!(s.status, InboundShipmentStatus::Pending);
393        assert_eq!(s.items.len(), 1);
394        let fetched = repo.get(s.id).expect("get").expect("found");
395        assert_eq!(fetched.total_expected(), dec!(10));
396    }
397
398    #[test]
399    fn lifecycle_transitions() {
400        let repo = test_repo();
401        let s = new_shipment(&repo);
402        assert_eq!(
403            repo.mark_in_transit(s.id).expect("transit").status,
404            InboundShipmentStatus::InTransit
405        );
406        assert_eq!(
407            repo.mark_arrived(s.id).expect("arrived").status,
408            InboundShipmentStatus::Arrived
409        );
410        let item = s.items[0].id;
411        let partial = repo.receive_line(s.id, item, dec!(4)).expect("partial");
412        assert_eq!(partial.status, InboundShipmentStatus::PartiallyReceived);
413        let full = repo.receive_line(s.id, item, dec!(6)).expect("full");
414        assert_eq!(full.status, InboundShipmentStatus::Received);
415        assert!(full.received_at.is_some());
416    }
417
418    #[test]
419    fn receive_line_rejects_over_receipt() {
420        let repo = test_repo();
421        let s = new_shipment(&repo);
422        let item = s.items[0].id;
423        // Line expects 10; receiving 11 at once is rejected.
424        assert!(repo.receive_line(s.id, item, dec!(11)).is_err());
425        // After receiving 6, another 5 (total 11) is rejected.
426        repo.receive_line(s.id, item, dec!(6)).expect("receive 6");
427        assert!(repo.receive_line(s.id, item, dec!(5)).is_err());
428        // Exact remaining (4) still succeeds.
429        let full = repo.receive_line(s.id, item, dec!(4)).expect("receive rest");
430        assert_eq!(full.status, InboundShipmentStatus::Received);
431    }
432
433    #[test]
434    fn cancel_sets_status() {
435        let repo = test_repo();
436        let s = new_shipment(&repo);
437        assert_eq!(repo.cancel(s.id).expect("cancel").status, InboundShipmentStatus::Cancelled);
438
439        // Terminal-state guard: cancelling again is rejected.
440        let err = repo.cancel(s.id).expect_err("already cancelled");
441        assert!(matches!(err, CommerceError::ValidationError(_)));
442    }
443
444    #[test]
445    fn list_filters_by_status() {
446        let repo = test_repo();
447        let a = new_shipment(&repo);
448        new_shipment(&repo);
449        repo.cancel(a.id).expect("cancel");
450        let cancelled = repo
451            .list(InboundShipmentFilter {
452                status: Some(InboundShipmentStatus::Cancelled),
453                ..Default::default()
454            })
455            .expect("list");
456        assert_eq!(cancelled.len(), 1);
457    }
458}
stateset_db/sqlite/inbound_shipments.rs

stateset_db/sqlite/
inbound_shipments.rs
