1use 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 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}