1use super::{
4 map_db_error, parse_datetime_opt_row, parse_datetime_row, parse_enum_row, parse_json_row,
5 parse_uuid_opt_row, parse_uuid_row, with_immediate_transaction,
6};
7use chrono::Utc;
8use r2d2::Pool;
9use r2d2_sqlite::SqliteConnectionManager;
10use stateset_core::{
11 CommerceError, CreateProductionBatch, ProductionBatch, ProductionBatchFilter,
12 ProductionBatchId, ProductionBatchRepository, ProductionBatchStatus, Result,
13 UpdateProductionBatch,
14};
15use uuid::Uuid;
16
17#[derive(Debug)]
18pub struct SqliteProductionBatchRepository {
19 pool: Pool<SqliteConnectionManager>,
20}
21
22impl SqliteProductionBatchRepository {
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_batch(row: &rusqlite::Row<'_>) -> rusqlite::Result<ProductionBatch> {
33 let work_order_ids_json: String = row.get("work_order_ids")?;
34 Ok(ProductionBatch {
35 id: parse_uuid_row(&row.get::<_, String>("id")?, "production_batch", "id")?.into(),
36 name: row.get("name")?,
37 status: parse_enum_row::<ProductionBatchStatus>(
38 &row.get::<_, String>("status")?,
39 "production_batch",
40 "status",
41 )?,
42 vendor_id: parse_uuid_opt_row(
43 row.get::<_, Option<String>>("vendor_id")?,
44 "production_batch",
45 "vendor_id",
46 )?,
47 work_order_ids: parse_json_row(
48 &work_order_ids_json,
49 "production_batch",
50 "work_order_ids",
51 )?,
52 notes: row.get("notes")?,
53 scheduled_start: parse_datetime_opt_row(
54 row.get::<_, Option<String>>("scheduled_start")?,
55 "production_batch",
56 "scheduled_start",
57 )?,
58 scheduled_end: parse_datetime_opt_row(
59 row.get::<_, Option<String>>("scheduled_end")?,
60 "production_batch",
61 "scheduled_end",
62 )?,
63 created_at: parse_datetime_row(
64 &row.get::<_, String>("created_at")?,
65 "production_batch",
66 "created_at",
67 )?,
68 updated_at: parse_datetime_row(
69 &row.get::<_, String>("updated_at")?,
70 "production_batch",
71 "updated_at",
72 )?,
73 })
74 }
75
76 fn json_err(e: serde_json::Error) -> rusqlite::Error {
77 rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::DatabaseError(
78 e.to_string(),
79 )))
80 }
81}
82
83impl ProductionBatchRepository for SqliteProductionBatchRepository {
84 fn create(&self, input: CreateProductionBatch) -> Result<ProductionBatch> {
85 let id = ProductionBatchId::new();
86 let id_str = id.to_string();
87 let now_str = Utc::now().to_rfc3339();
88 let work_order_ids_json = serde_json::to_string(&input.work_order_ids)
89 .map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
90 with_immediate_transaction(&self.pool, |tx| {
91 tx.execute(
92 "INSERT INTO production_batches (id, name, status, vendor_id, work_order_ids, notes, scheduled_start, scheduled_end, created_at, updated_at)
93 VALUES (?, ?, 'planned', ?, ?, ?, ?, ?, ?, ?)",
94 rusqlite::params![
95 &id_str,
96 &input.name,
97 input.vendor_id.map(|v| v.to_string()),
98 &work_order_ids_json,
99 &input.notes,
100 input.scheduled_start.map(|d| d.to_rfc3339()),
101 input.scheduled_end.map(|d| d.to_rfc3339()),
102 &now_str,
103 &now_str,
104 ],
105 )?;
106 tx.query_row(
107 "SELECT * FROM production_batches WHERE id = ?",
108 [&id_str],
109 Self::row_to_batch,
110 )
111 })
112 }
113
114 fn get(&self, id: ProductionBatchId) -> Result<Option<ProductionBatch>> {
115 let conn = self.conn()?;
116 match conn.query_row(
117 "SELECT * FROM production_batches WHERE id = ?",
118 [id.to_string()],
119 Self::row_to_batch,
120 ) {
121 Ok(b) => Ok(Some(b)),
122 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
123 Err(e) => Err(map_db_error(e)),
124 }
125 }
126
127 fn update(
128 &self,
129 id: ProductionBatchId,
130 input: UpdateProductionBatch,
131 ) -> Result<ProductionBatch> {
132 let id_str = id.to_string();
133 let now_str = Utc::now().to_rfc3339();
134 with_immediate_transaction(&self.pool, |tx| {
135 let mut sets = vec!["updated_at = ?".to_string()];
136 let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(now_str.clone())];
137
138 if let Some(ref name) = input.name {
139 sets.push("name = ?".into());
140 params.push(Box::new(name.clone()));
141 }
142 if let Some(vendor) = input.vendor_id {
143 sets.push("vendor_id = ?".into());
144 params.push(Box::new(vendor.to_string()));
145 }
146 if let Some(status) = input.status {
147 sets.push("status = ?".into());
148 params.push(Box::new(status.to_string()));
149 }
150 if let Some(ref notes) = input.notes {
151 sets.push("notes = ?".into());
152 params.push(Box::new(notes.clone()));
153 }
154 if let Some(start) = input.scheduled_start {
155 sets.push("scheduled_start = ?".into());
156 params.push(Box::new(start.to_rfc3339()));
157 }
158 if let Some(end) = input.scheduled_end {
159 sets.push("scheduled_end = ?".into());
160 params.push(Box::new(end.to_rfc3339()));
161 }
162
163 let sql = format!("UPDATE production_batches SET {} WHERE id = ?", sets.join(", "));
164 params.push(Box::new(id_str.clone()));
165 let param_refs: Vec<&dyn rusqlite::types::ToSql> =
166 params.iter().map(|p| p.as_ref()).collect();
167 tx.execute(&sql, param_refs.as_slice())?;
168
169 tx.query_row(
170 "SELECT * FROM production_batches WHERE id = ?",
171 [&id_str],
172 Self::row_to_batch,
173 )
174 })
175 }
176
177 fn list(&self, filter: ProductionBatchFilter) -> Result<Vec<ProductionBatch>> {
178 let conn = self.conn()?;
179 let mut sql = "SELECT * FROM production_batches WHERE 1=1".to_string();
180 let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
181 if let Some(status) = filter.status {
182 sql.push_str(" AND status = ?");
183 params.push(Box::new(status.to_string()));
184 }
185 if let Some(vendor) = filter.vendor_id {
186 sql.push_str(" AND vendor_id = ?");
187 params.push(Box::new(vendor.to_string()));
188 }
189 sql.push_str(" ORDER BY created_at DESC");
190 crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
191
192 let param_refs: Vec<&dyn rusqlite::types::ToSql> =
193 params.iter().map(|p| p.as_ref()).collect();
194 let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
195 let rows = stmt
196 .query_map(param_refs.as_slice(), Self::row_to_batch)
197 .map_err(map_db_error)?
198 .collect::<std::result::Result<Vec<_>, _>>()
199 .map_err(map_db_error)?;
200 Ok(rows)
201 }
202
203 fn delete(&self, id: ProductionBatchId) -> Result<()> {
204 let conn = self.conn()?;
205 conn.execute("DELETE FROM production_batches WHERE id = ?", [id.to_string()])
206 .map_err(map_db_error)?;
207 Ok(())
208 }
209
210 fn add_work_orders(
211 &self,
212 id: ProductionBatchId,
213 work_order_ids: Vec<Uuid>,
214 ) -> Result<ProductionBatch> {
215 let id_str = id.to_string();
216 let now_str = Utc::now().to_rfc3339();
217 with_immediate_transaction(&self.pool, |tx| {
218 let current_json: String = tx.query_row(
219 "SELECT work_order_ids FROM production_batches WHERE id = ?",
220 [&id_str],
221 |r| r.get(0),
222 )?;
223 let mut ids: Vec<Uuid> = serde_json::from_str(¤t_json).map_err(Self::json_err)?;
224 for wo in &work_order_ids {
225 if !ids.contains(wo) {
226 ids.push(*wo);
227 }
228 }
229 let json = serde_json::to_string(&ids).map_err(Self::json_err)?;
230 tx.execute(
231 "UPDATE production_batches SET work_order_ids = ?, updated_at = ? WHERE id = ?",
232 rusqlite::params![&json, &now_str, &id_str],
233 )?;
234 tx.query_row(
235 "SELECT * FROM production_batches WHERE id = ?",
236 [&id_str],
237 Self::row_to_batch,
238 )
239 })
240 }
241
242 fn remove_work_order(
243 &self,
244 id: ProductionBatchId,
245 work_order_id: Uuid,
246 ) -> Result<ProductionBatch> {
247 let id_str = id.to_string();
248 let now_str = Utc::now().to_rfc3339();
249 with_immediate_transaction(&self.pool, |tx| {
250 let current_json: String = tx.query_row(
251 "SELECT work_order_ids FROM production_batches WHERE id = ?",
252 [&id_str],
253 |r| r.get(0),
254 )?;
255 let mut ids: Vec<Uuid> = serde_json::from_str(¤t_json).map_err(Self::json_err)?;
256 ids.retain(|w| *w != work_order_id);
257 let json = serde_json::to_string(&ids).map_err(Self::json_err)?;
258 tx.execute(
259 "UPDATE production_batches SET work_order_ids = ?, updated_at = ? WHERE id = ?",
260 rusqlite::params![&json, &now_str, &id_str],
261 )?;
262 tx.query_row(
263 "SELECT * FROM production_batches WHERE id = ?",
264 [&id_str],
265 Self::row_to_batch,
266 )
267 })
268 }
269}
270
271#[cfg(test)]
272mod tests {
273 use super::*;
274 use crate::DatabaseConfig;
275 use crate::sqlite::SqliteDatabase;
276
277 fn test_repo() -> SqliteProductionBatchRepository {
278 let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
279 SqliteProductionBatchRepository::new(db.pool().clone())
280 }
281
282 fn new_batch(repo: &SqliteProductionBatchRepository, name: &str) -> ProductionBatch {
283 repo.create(CreateProductionBatch {
284 name: name.into(),
285 vendor_id: None,
286 work_order_ids: vec![],
287 notes: None,
288 scheduled_start: None,
289 scheduled_end: None,
290 })
291 .expect("create batch")
292 }
293
294 #[test]
295 fn create_get_update() {
296 let repo = test_repo();
297 let b = new_batch(&repo, "Batch A");
298 assert_eq!(b.status, ProductionBatchStatus::Planned);
299 let updated = repo
300 .update(
301 b.id,
302 UpdateProductionBatch {
303 status: Some(ProductionBatchStatus::InProgress),
304 ..Default::default()
305 },
306 )
307 .expect("update");
308 assert_eq!(updated.status, ProductionBatchStatus::InProgress);
309 }
310
311 #[test]
312 fn add_and_remove_work_orders_idempotent() {
313 let repo = test_repo();
314 let b = new_batch(&repo, "Batch B");
315 let wo1 = Uuid::from_bytes([1; 16]);
316 let wo2 = Uuid::from_bytes([2; 16]);
317 let updated = repo.add_work_orders(b.id, vec![wo1, wo2, wo1]).expect("add");
318 assert_eq!(updated.work_order_count(), 2);
319 let again = repo.add_work_orders(b.id, vec![wo1]).expect("add again");
321 assert_eq!(again.work_order_count(), 2);
322 let removed = repo.remove_work_order(b.id, wo1).expect("remove");
323 assert_eq!(removed.work_order_count(), 1);
324 assert_eq!(removed.work_order_ids, vec![wo2]);
325 }
326
327 #[test]
328 fn list_filters_by_status() {
329 let repo = test_repo();
330 let a = new_batch(&repo, "A");
331 new_batch(&repo, "B");
332 repo.update(
333 a.id,
334 UpdateProductionBatch {
335 status: Some(ProductionBatchStatus::Completed),
336 ..Default::default()
337 },
338 )
339 .expect("update");
340 let completed = repo
341 .list(ProductionBatchFilter {
342 status: Some(ProductionBatchStatus::Completed),
343 ..Default::default()
344 })
345 .expect("list");
346 assert_eq!(completed.len(), 1);
347 }
348
349 #[test]
350 fn delete_removes_batch() {
351 let repo = test_repo();
352 let b = new_batch(&repo, "Doomed");
353 repo.delete(b.id).expect("delete");
354 assert!(repo.get(b.id).expect("get").is_none());
355 }
356}