stateset_db/sqlite/
vendor_returns.rs

1//! SQLite implementation of the vendor return 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 stateset_core::{
12    CommerceError, CreateVendorReturn, CurrencyCode, Result, VendorReturn, VendorReturnFilter,
13    VendorReturnId, VendorReturnItem, VendorReturnItemId, VendorReturnReason, VendorReturnStatus,
14};
15
16#[derive(Debug)]
17pub struct SqliteVendorReturnRepository {
18    pool: Pool<SqliteConnectionManager>,
19}
20
21impl SqliteVendorReturnRepository {
22    #[must_use]
23    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
24        Self { pool }
25    }
26
27    fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
28        self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
29    }
30
31    fn row_to_item(row: &rusqlite::Row<'_>) -> rusqlite::Result<VendorReturnItem> {
32        Ok(VendorReturnItem {
33            id: parse_uuid_row(&row.get::<_, String>("id")?, "vendor_return_item", "id")?.into(),
34            vendor_return_id: parse_uuid_row(
35                &row.get::<_, String>("vendor_return_id")?,
36                "vendor_return_item",
37                "vendor_return_id",
38            )?
39            .into(),
40            product_id: parse_uuid_row(
41                &row.get::<_, String>("product_id")?,
42                "vendor_return_item",
43                "product_id",
44            )?
45            .into(),
46            sku: row.get("sku")?,
47            quantity: parse_decimal_row(
48                &row.get::<_, String>("quantity")?,
49                "vendor_return_item",
50                "quantity",
51            )?,
52            unit_cost: parse_decimal_row(
53                &row.get::<_, String>("unit_cost")?,
54                "vendor_return_item",
55                "unit_cost",
56            )?,
57            reason: parse_enum_row::<VendorReturnReason>(
58                &row.get::<_, String>("reason")?,
59                "vendor_return_item",
60                "reason",
61            )?,
62        })
63    }
64
65    fn row_to_head(row: &rusqlite::Row<'_>) -> rusqlite::Result<VendorReturn> {
66        Ok(VendorReturn {
67            id: parse_uuid_row(&row.get::<_, String>("id")?, "vendor_return", "id")?.into(),
68            number: row.get("number")?,
69            supplier_id: parse_uuid_row(
70                &row.get::<_, String>("supplier_id")?,
71                "vendor_return",
72                "supplier_id",
73            )?,
74            purchase_order_id: parse_uuid_opt_row(
75                row.get::<_, Option<String>>("purchase_order_id")?,
76                "vendor_return",
77                "purchase_order_id",
78            )?,
79            status: parse_enum_row::<VendorReturnStatus>(
80                &row.get::<_, String>("status")?,
81                "vendor_return",
82                "status",
83            )?,
84            currency: parse_enum_row::<CurrencyCode>(
85                &row.get::<_, String>("currency")?,
86                "vendor_return",
87                "currency",
88            )?,
89            items: Vec::new(),
90            credit_generated: row.get::<_, i32>("credit_generated")? != 0,
91            notes: row.get("notes")?,
92            processed_at: parse_datetime_opt_row(
93                row.get::<_, Option<String>>("processed_at")?,
94                "vendor_return",
95                "processed_at",
96            )?,
97            created_at: parse_datetime_row(
98                &row.get::<_, String>("created_at")?,
99                "vendor_return",
100                "created_at",
101            )?,
102            updated_at: parse_datetime_row(
103                &row.get::<_, String>("updated_at")?,
104                "vendor_return",
105                "updated_at",
106            )?,
107        })
108    }
109
110    fn load_items(
111        conn: &rusqlite::Connection,
112        id: &str,
113    ) -> rusqlite::Result<Vec<VendorReturnItem>> {
114        let mut stmt = conn
115            .prepare("SELECT * FROM vendor_return_items WHERE vendor_return_id = ? ORDER BY sku")?;
116        stmt.query_map([id], Self::row_to_item)?.collect()
117    }
118
119    fn load_items_batch(
120        conn: &rusqlite::Connection,
121        ids: &[String],
122    ) -> rusqlite::Result<std::collections::HashMap<String, Vec<VendorReturnItem>>> {
123        let mut map: std::collections::HashMap<String, Vec<VendorReturnItem>> =
124            std::collections::HashMap::with_capacity(ids.len());
125        for chunk in ids.chunks(500) {
126            let placeholders = super::build_in_clause(chunk.len());
127            let sql = format!(
128                "SELECT * FROM vendor_return_items WHERE vendor_return_id IN ({placeholders}) ORDER BY sku"
129            );
130            let mut stmt = conn.prepare(&sql)?;
131            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
132                chunk.iter().map(|s| s as &dyn rusqlite::types::ToSql).collect();
133            let rows = stmt.query_map(param_refs.as_slice(), |row| {
134                let parent: String = row.get("vendor_return_id")?;
135                Ok((parent, Self::row_to_item(row)?))
136            })?;
137            for row in rows {
138                let (parent, item) = row?;
139                map.entry(parent).or_default().push(item);
140            }
141        }
142        Ok(map)
143    }
144
145    fn load_full(conn: &rusqlite::Connection, id: &str) -> rusqlite::Result<VendorReturn> {
146        let mut head =
147            conn.query_row("SELECT * FROM vendor_returns WHERE id = ?", [id], Self::row_to_head)?;
148        head.items = Self::load_items(conn, id)?;
149        Ok(head)
150    }
151
152    /// Guard a status transition, returning the current status or `NoRows`.
153    fn current_status(tx: &rusqlite::Connection, id: &str) -> rusqlite::Result<VendorReturnStatus> {
154        let s: Option<String> = tx
155            .query_row("SELECT status FROM vendor_returns WHERE id = ?", [id], |r| r.get(0))
156            .optional()?;
157        let s = s.ok_or(rusqlite::Error::QueryReturnedNoRows)?;
158        parse_enum_row::<VendorReturnStatus>(&s, "vendor_return", "status")
159    }
160
161    fn conflict(msg: &str) -> rusqlite::Error {
162        rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::Conflict(msg.to_string())))
163    }
164}
165
166impl stateset_core::VendorReturnRepository for SqliteVendorReturnRepository {
167    fn create(&self, input: CreateVendorReturn) -> Result<VendorReturn> {
168        if input.items.is_empty() {
169            return Err(CommerceError::ValidationError(
170                "a vendor return requires at least one item".into(),
171            ));
172        }
173        let id = VendorReturnId::new();
174        let id_str = id.to_string();
175        let now_str = Utc::now().to_rfc3339();
176        let number = format!("VR-{}", &id_str[..8]);
177        let currency = input.currency.unwrap_or(CurrencyCode::USD);
178
179        with_immediate_transaction(&self.pool, |tx| {
180            tx.execute(
181                "INSERT INTO vendor_returns (id, number, supplier_id, purchase_order_id, status, currency, credit_generated, notes, created_at, updated_at)
182                 VALUES (?, ?, ?, ?, 'draft', ?, 0, ?, ?, ?)",
183                rusqlite::params![
184                    &id_str,
185                    &number,
186                    input.supplier_id.to_string(),
187                    input.purchase_order_id.map(|p| p.to_string()),
188                    currency.to_string(),
189                    &input.notes,
190                    &now_str,
191                    &now_str,
192                ],
193            )?;
194            for item in &input.items {
195                tx.execute(
196                    "INSERT INTO vendor_return_items (id, vendor_return_id, product_id, sku, quantity, unit_cost, reason)
197                     VALUES (?, ?, ?, '', ?, ?, ?)",
198                    rusqlite::params![
199                        VendorReturnItemId::new().to_string(),
200                        &id_str,
201                        item.product_id.to_string(),
202                        item.quantity.to_string(),
203                        item.unit_cost.to_string(),
204                        item.reason.to_string(),
205                    ],
206                )?;
207            }
208            Self::load_full(tx, &id_str)
209        })
210    }
211
212    fn get(&self, id: VendorReturnId) -> Result<Option<VendorReturn>> {
213        let conn = self.conn()?;
214        match Self::load_full(&conn, &id.to_string()) {
215            Ok(r) => Ok(Some(r)),
216            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
217            Err(e) => Err(map_db_error(e)),
218        }
219    }
220
221    fn list(&self, filter: VendorReturnFilter) -> Result<Vec<VendorReturn>> {
222        let conn = self.conn()?;
223        let mut sql = "SELECT * FROM vendor_returns WHERE 1=1".to_string();
224        let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
225        if let Some(supplier) = filter.supplier_id {
226            sql.push_str(" AND supplier_id = ?");
227            params.push(Box::new(supplier.to_string()));
228        }
229        if let Some(status) = filter.status {
230            sql.push_str(" AND status = ?");
231            params.push(Box::new(status.to_string()));
232        }
233        sql.push_str(" ORDER BY created_at DESC");
234        crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
235        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
236            params.iter().map(|p| p.as_ref()).collect();
237        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
238        let heads = stmt
239            .query_map(param_refs.as_slice(), Self::row_to_head)
240            .map_err(map_db_error)?
241            .collect::<std::result::Result<Vec<_>, _>>()
242            .map_err(map_db_error)?;
243        let ids: Vec<String> = heads.iter().map(|h| h.id.to_string()).collect();
244        let mut items_by_id = Self::load_items_batch(&conn, &ids).map_err(map_db_error)?;
245        let mut out = Vec::with_capacity(heads.len());
246        for mut head in heads {
247            head.items = items_by_id.remove(&head.id.to_string()).unwrap_or_default();
248            out.push(head);
249        }
250        Ok(out)
251    }
252
253    fn submit(&self, id: VendorReturnId) -> Result<VendorReturn> {
254        let id_str = id.to_string();
255        let now = Utc::now().to_rfc3339();
256        with_immediate_transaction(&self.pool, |tx| {
257            if Self::current_status(tx, &id_str)? != VendorReturnStatus::Draft {
258                return Err(Self::conflict("only draft vendor returns can be submitted"));
259            }
260            tx.execute(
261                "UPDATE vendor_returns SET status = 'pending', updated_at = ? WHERE id = ?",
262                rusqlite::params![&now, &id_str],
263            )?;
264            Self::load_full(tx, &id_str)
265        })
266    }
267
268    fn process(&self, id: VendorReturnId, generate_credit: bool) -> Result<VendorReturn> {
269        let id_str = id.to_string();
270        let now = Utc::now().to_rfc3339();
271        with_immediate_transaction(&self.pool, |tx| {
272            let status = Self::current_status(tx, &id_str)?;
273            if status.is_terminal() {
274                return Err(Self::conflict("vendor return is already in a terminal state"));
275            }
276            tx.execute(
277                "UPDATE vendor_returns SET status = 'processed', credit_generated = ?, processed_at = ?, updated_at = ? WHERE id = ?",
278                rusqlite::params![generate_credit as i32, &now, &now, &id_str],
279            )?;
280            Self::load_full(tx, &id_str)
281        })
282    }
283
284    fn cancel(&self, id: VendorReturnId) -> Result<VendorReturn> {
285        let id_str = id.to_string();
286        let now = Utc::now().to_rfc3339();
287        with_immediate_transaction(&self.pool, |tx| {
288            if Self::current_status(tx, &id_str)? == VendorReturnStatus::Processed {
289                return Err(Self::conflict("processed vendor returns cannot be cancelled"));
290            }
291            tx.execute(
292                "UPDATE vendor_returns SET status = 'cancelled', updated_at = ? WHERE id = ?",
293                rusqlite::params![&now, &id_str],
294            )?;
295            Self::load_full(tx, &id_str)
296        })
297    }
298}
299
300#[cfg(test)]
301mod tests {
302    use super::*;
303    use crate::DatabaseConfig;
304    use crate::sqlite::SqliteDatabase;
305    use rust_decimal_macros::dec;
306    use stateset_core::{CreateVendorReturnItem, ProductId, VendorReturnRepository};
307    use uuid::Uuid;
308
309    fn test_repo() -> SqliteVendorReturnRepository {
310        let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
311        SqliteVendorReturnRepository::new(db.pool().clone())
312    }
313
314    fn new_return(repo: &SqliteVendorReturnRepository) -> VendorReturn {
315        repo.create(CreateVendorReturn {
316            supplier_id: Uuid::new_v4(),
317            purchase_order_id: None,
318            currency: Some(CurrencyCode::USD),
319            items: vec![CreateVendorReturnItem {
320                product_id: ProductId::new(),
321                quantity: dec!(3),
322                unit_cost: dec!(10),
323                reason: VendorReturnReason::Defective,
324            }],
325            notes: Some("defective batch".into()),
326        })
327        .expect("create vendor return")
328    }
329
330    #[test]
331    fn create_get_with_items() {
332        let repo = test_repo();
333        let r = new_return(&repo);
334        assert_eq!(r.status, VendorReturnStatus::Draft);
335        assert_eq!(r.items.len(), 1);
336        let fetched = repo.get(r.id).expect("get").expect("found");
337        assert_eq!(fetched.total_credit(), dec!(30));
338    }
339
340    #[test]
341    fn create_rejects_empty_items() {
342        let repo = test_repo();
343        let res = repo.create(CreateVendorReturn {
344            supplier_id: Uuid::new_v4(),
345            purchase_order_id: None,
346            currency: None,
347            items: vec![],
348            notes: None,
349        });
350        assert!(res.is_err());
351    }
352
353    #[test]
354    fn submit_then_process_generates_credit() {
355        let repo = test_repo();
356        let r = new_return(&repo);
357        let submitted = repo.submit(r.id).expect("submit");
358        assert_eq!(submitted.status, VendorReturnStatus::Pending);
359        let processed = repo.process(r.id, true).expect("process");
360        assert_eq!(processed.status, VendorReturnStatus::Processed);
361        assert!(processed.credit_generated);
362        assert!(processed.processed_at.is_some());
363    }
364
365    #[test]
366    fn submit_twice_conflicts() {
367        let repo = test_repo();
368        let r = new_return(&repo);
369        repo.submit(r.id).expect("submit");
370        assert!(repo.submit(r.id).is_err());
371    }
372
373    #[test]
374    fn processed_cannot_be_cancelled() {
375        let repo = test_repo();
376        let r = new_return(&repo);
377        repo.process(r.id, false).expect("process");
378        assert!(repo.cancel(r.id).is_err());
379    }
380
381    #[test]
382    fn list_filters_by_status() {
383        let repo = test_repo();
384        let a = new_return(&repo);
385        new_return(&repo);
386        repo.cancel(a.id).expect("cancel");
387        let cancelled = repo
388            .list(VendorReturnFilter {
389                status: Some(VendorReturnStatus::Cancelled),
390                ..Default::default()
391            })
392            .expect("list");
393        assert_eq!(cancelled.len(), 1);
394    }
395}
stateset_db/sqlite/vendor_returns.rs

stateset_db/sqlite/
vendor_returns.rs
