stateset_db/sqlite/
purgatory.rs

1//! SQLite implementation of the purgatory (order ingestion staging) repository
2
3use super::{
4    map_db_error, parse_datetime_row, parse_decimal_row, parse_json_row, parse_uuid_opt_row,
5    parse_uuid_row, with_immediate_transaction,
6};
7use chrono::Utc;
8use r2d2::Pool;
9use r2d2_sqlite::SqliteConnectionManager;
10use stateset_core::{
11    CommerceError, IngestOrder, MapPurgatoryLine, PurgatoryFilter, PurgatoryLineItem,
12    PurgatoryLineItemId, PurgatoryOrder, PurgatoryOrderId, PurgatoryRepository, Result,
13};
14
15#[derive(Debug)]
16pub struct SqlitePurgatoryRepository {
17    pool: Pool<SqliteConnectionManager>,
18}
19
20impl SqlitePurgatoryRepository {
21    #[must_use]
22    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
23        Self { pool }
24    }
25
26    fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
27        self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
28    }
29
30    fn row_to_line(row: &rusqlite::Row<'_>) -> rusqlite::Result<PurgatoryLineItem> {
31        Ok(PurgatoryLineItem {
32            id: parse_uuid_row(&row.get::<_, String>("id")?, "purgatory_line", "id")?.into(),
33            purgatory_order_id: parse_uuid_row(
34                &row.get::<_, String>("purgatory_order_id")?,
35                "purgatory_line",
36                "purgatory_order_id",
37            )?
38            .into(),
39            external_sku: row.get("external_sku")?,
40            product_id: parse_uuid_opt_row(
41                row.get::<_, Option<String>>("product_id")?,
42                "purgatory_line",
43                "product_id",
44            )?
45            .map(Into::into),
46            quantity: parse_decimal_row(
47                &row.get::<_, String>("quantity")?,
48                "purgatory_line",
49                "quantity",
50            )?,
51            ignore_item: row.get::<_, i32>("ignore_item")? != 0,
52            non_physical: row.get::<_, i32>("non_physical")? != 0,
53        })
54    }
55
56    fn row_to_head(row: &rusqlite::Row<'_>) -> rusqlite::Result<PurgatoryOrder> {
57        let metadata_json: String = row.get("metadata")?;
58        Ok(PurgatoryOrder {
59            id: parse_uuid_row(&row.get::<_, String>("id")?, "purgatory_order", "id")?.into(),
60            channel_id: parse_uuid_opt_row(
61                row.get::<_, Option<String>>("channel_id")?,
62                "purgatory_order",
63                "channel_id",
64            )?
65            .map(Into::into),
66            external_order_id: row.get("external_order_id")?,
67            external_status: row.get("external_status")?,
68            is_posted: row.get::<_, i32>("is_posted")? != 0,
69            hold_reason: row.get("hold_reason")?,
70            metadata: parse_json_row(&metadata_json, "purgatory_order", "metadata")?,
71            items: Vec::new(),
72            created_at: parse_datetime_row(
73                &row.get::<_, String>("created_at")?,
74                "purgatory_order",
75                "created_at",
76            )?,
77            updated_at: parse_datetime_row(
78                &row.get::<_, String>("updated_at")?,
79                "purgatory_order",
80                "updated_at",
81            )?,
82        })
83    }
84
85    fn load_items(
86        conn: &rusqlite::Connection,
87        id: &str,
88    ) -> rusqlite::Result<Vec<PurgatoryLineItem>> {
89        let mut stmt = conn.prepare(
90            "SELECT * FROM purgatory_line_items WHERE purgatory_order_id = ? ORDER BY external_sku",
91        )?;
92        stmt.query_map([id], Self::row_to_line)?.collect()
93    }
94
95    fn load_items_batch(
96        conn: &rusqlite::Connection,
97        ids: &[String],
98    ) -> rusqlite::Result<std::collections::HashMap<String, Vec<PurgatoryLineItem>>> {
99        let mut map: std::collections::HashMap<String, Vec<PurgatoryLineItem>> =
100            std::collections::HashMap::with_capacity(ids.len());
101        for chunk in ids.chunks(500) {
102            let placeholders = super::build_in_clause(chunk.len());
103            let sql = format!(
104                "SELECT * FROM purgatory_line_items WHERE purgatory_order_id IN ({placeholders}) ORDER BY external_sku"
105            );
106            let mut stmt = conn.prepare(&sql)?;
107            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
108                chunk.iter().map(|s| s as &dyn rusqlite::types::ToSql).collect();
109            let rows = stmt.query_map(param_refs.as_slice(), |row| {
110                let parent: String = row.get("purgatory_order_id")?;
111                Ok((parent, Self::row_to_line(row)?))
112            })?;
113            for row in rows {
114                let (parent, item) = row?;
115                map.entry(parent).or_default().push(item);
116            }
117        }
118        Ok(map)
119    }
120
121    fn load_full(conn: &rusqlite::Connection, id: &str) -> rusqlite::Result<PurgatoryOrder> {
122        let mut head =
123            conn.query_row("SELECT * FROM purgatory_orders WHERE id = ?", [id], Self::row_to_head)?;
124        head.items = Self::load_items(conn, id)?;
125        Ok(head)
126    }
127}
128
129impl PurgatoryRepository for SqlitePurgatoryRepository {
130    fn ingest(&self, input: IngestOrder) -> Result<PurgatoryOrder> {
131        if input.items.is_empty() {
132            return Err(CommerceError::ValidationError(
133                "an ingested order requires at least one line".into(),
134            ));
135        }
136        let id = PurgatoryOrderId::new();
137        let id_str = id.to_string();
138        let now = Utc::now().to_rfc3339();
139        let metadata_json = serde_json::to_string(&input.metadata)
140            .map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
141        with_immediate_transaction(&self.pool, |tx| {
142            tx.execute(
143                "INSERT INTO purgatory_orders (id, channel_id, external_order_id, external_status, is_posted, metadata, created_at, updated_at)
144                 VALUES (?, ?, ?, ?, 0, ?, ?, ?)",
145                rusqlite::params![
146                    &id_str,
147                    input.channel_id.map(|c| c.to_string()),
148                    &input.external_order_id,
149                    &input.external_status,
150                    &metadata_json,
151                    &now,
152                    &now,
153                ],
154            )?;
155            for item in &input.items {
156                tx.execute(
157                    "INSERT INTO purgatory_line_items (id, purgatory_order_id, external_sku, product_id, quantity, ignore_item, non_physical)
158                     VALUES (?, ?, ?, ?, ?, 0, 0)",
159                    rusqlite::params![
160                        PurgatoryLineItemId::new().to_string(),
161                        &id_str,
162                        &item.external_sku,
163                        item.product_id.map(|p| p.to_string()),
164                        item.quantity.to_string(),
165                    ],
166                )?;
167            }
168            Self::load_full(tx, &id_str)
169        })
170    }
171
172    fn get(&self, id: PurgatoryOrderId) -> Result<Option<PurgatoryOrder>> {
173        let conn = self.conn()?;
174        match Self::load_full(&conn, &id.to_string()) {
175            Ok(o) => Ok(Some(o)),
176            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
177            Err(e) => Err(map_db_error(e)),
178        }
179    }
180
181    fn list(&self, filter: PurgatoryFilter) -> Result<Vec<PurgatoryOrder>> {
182        let conn = self.conn()?;
183        let mut sql = "SELECT * FROM purgatory_orders WHERE 1=1".to_string();
184        let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
185        if let Some(channel) = filter.channel_id {
186            sql.push_str(" AND channel_id = ?");
187            params.push(Box::new(channel.to_string()));
188        }
189        // Defaults to non-posted when not specified.
190        let is_posted = filter.is_posted.unwrap_or(false);
191        sql.push_str(" AND is_posted = ?");
192        params.push(Box::new(is_posted as i32));
193        sql.push_str(" ORDER BY created_at DESC");
194        crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
195        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
196            params.iter().map(|p| p.as_ref()).collect();
197        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
198        let heads = stmt
199            .query_map(param_refs.as_slice(), Self::row_to_head)
200            .map_err(map_db_error)?
201            .collect::<std::result::Result<Vec<_>, _>>()
202            .map_err(map_db_error)?;
203        let ids: Vec<String> = heads.iter().map(|h| h.id.to_string()).collect();
204        let mut items_by_id = Self::load_items_batch(&conn, &ids).map_err(map_db_error)?;
205        let mut out = Vec::with_capacity(heads.len());
206        for mut head in heads {
207            head.items = items_by_id.remove(&head.id.to_string()).unwrap_or_default();
208            out.push(head);
209        }
210        Ok(out)
211    }
212
213    fn map_line(
214        &self,
215        id: PurgatoryOrderId,
216        line_id: PurgatoryLineItemId,
217        input: MapPurgatoryLine,
218    ) -> Result<PurgatoryOrder> {
219        let id_str = id.to_string();
220        let line_str = line_id.to_string();
221        let now = Utc::now().to_rfc3339();
222        with_immediate_transaction(&self.pool, |tx| {
223            let mut sets: Vec<String> = vec![];
224            let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
225            if let Some(product_id) = input.product_id {
226                sets.push("product_id = ?".into());
227                params.push(Box::new(product_id.to_string()));
228            }
229            if let Some(ignore) = input.ignore_item {
230                sets.push("ignore_item = ?".into());
231                params.push(Box::new(ignore as i32));
232            }
233            if let Some(non_physical) = input.non_physical {
234                sets.push("non_physical = ?".into());
235                params.push(Box::new(non_physical as i32));
236            }
237            if !sets.is_empty() {
238                let sql = format!(
239                    "UPDATE purgatory_line_items SET {} WHERE id = ? AND purgatory_order_id = ?",
240                    sets.join(", ")
241                );
242                params.push(Box::new(line_str.clone()));
243                params.push(Box::new(id_str.clone()));
244                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
245                    params.iter().map(|p| p.as_ref()).collect();
246                tx.execute(&sql, param_refs.as_slice())?;
247            }
248            tx.execute(
249                "UPDATE purgatory_orders SET updated_at = ? WHERE id = ?",
250                rusqlite::params![&now, &id_str],
251            )?;
252            Self::load_full(tx, &id_str)
253        })
254    }
255
256    fn post(&self, id: PurgatoryOrderId) -> Result<PurgatoryOrder> {
257        let id_str = id.to_string();
258        let now = Utc::now().to_rfc3339();
259        with_immediate_transaction(&self.pool, |tx| {
260            let order = Self::load_full(tx, &id_str)?;
261            if order.is_posted {
262                return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
263                    CommerceError::Conflict("order is already posted".into()),
264                )));
265            }
266            if !order.is_ready_to_post() {
267                return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
268                    CommerceError::ValidationError(format!(
269                        "{} line(s) still unresolved",
270                        order.unresolved_count()
271                    )),
272                )));
273            }
274            tx.execute(
275                "UPDATE purgatory_orders SET is_posted = 1, hold_reason = NULL, updated_at = ? WHERE id = ?",
276                rusqlite::params![&now, &id_str],
277            )?;
278            Self::load_full(tx, &id_str)
279        })
280    }
281
282    fn delete(&self, id: PurgatoryOrderId) -> Result<()> {
283        let id_str = id.to_string();
284        with_immediate_transaction(&self.pool, |tx| {
285            tx.execute("DELETE FROM purgatory_line_items WHERE purgatory_order_id = ?", [&id_str])?;
286            tx.execute("DELETE FROM purgatory_orders WHERE id = ?", [&id_str])?;
287            Ok(())
288        })
289    }
290}
291
292#[cfg(test)]
293mod tests {
294    use super::*;
295    use crate::DatabaseConfig;
296    use crate::sqlite::SqliteDatabase;
297    use rust_decimal_macros::dec;
298    use stateset_core::{IngestLineItem, ProductId};
299
300    fn test_repo() -> SqlitePurgatoryRepository {
301        let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
302        SqlitePurgatoryRepository::new(db.pool().clone())
303    }
304
305    fn ingest(repo: &SqlitePurgatoryRepository, mapped: bool) -> PurgatoryOrder {
306        repo.ingest(IngestOrder {
307            channel_id: None,
308            external_order_id: "SHOP-1001".into(),
309            external_status: Some("paid".into()),
310            metadata: serde_json::json!({"src": "shopify"}),
311            items: vec![IngestLineItem {
312                external_sku: "EXT-1".into(),
313                quantity: dec!(2),
314                product_id: if mapped { Some(ProductId::new()) } else { None },
315            }],
316        })
317        .expect("ingest")
318    }
319
320    #[test]
321    fn ingest_rejects_empty() {
322        let repo = test_repo();
323        let res = repo.ingest(IngestOrder {
324            channel_id: None,
325            external_order_id: "X".into(),
326            external_status: None,
327            metadata: serde_json::Value::Null,
328            items: vec![],
329        });
330        assert!(res.is_err());
331    }
332
333    #[test]
334    fn unmapped_order_cannot_post() {
335        let repo = test_repo();
336        let o = ingest(&repo, false);
337        assert!(!o.is_ready_to_post());
338        assert!(repo.post(o.id).is_err());
339    }
340
341    #[test]
342    fn map_line_then_post() {
343        let repo = test_repo();
344        let o = ingest(&repo, false);
345        let line = o.items[0].id;
346        let mapped = repo
347            .map_line(
348                o.id,
349                line,
350                MapPurgatoryLine { product_id: Some(ProductId::new()), ..Default::default() },
351            )
352            .expect("map");
353        assert!(mapped.is_ready_to_post());
354        let posted = repo.post(o.id).expect("post");
355        assert!(posted.is_posted);
356        // double-post fails
357        assert!(repo.post(o.id).is_err());
358    }
359
360    #[test]
361    fn ignore_flag_resolves_line() {
362        let repo = test_repo();
363        let o = ingest(&repo, false);
364        let line = o.items[0].id;
365        repo.map_line(
366            o.id,
367            line,
368            MapPurgatoryLine { ignore_item: Some(true), ..Default::default() },
369        )
370        .expect("ignore");
371        assert!(repo.post(o.id).is_ok());
372    }
373
374    #[test]
375    fn list_defaults_to_non_posted() {
376        let repo = test_repo();
377        let a = ingest(&repo, true);
378        ingest(&repo, true);
379        repo.post(a.id).expect("post");
380        // default filter → only non-posted
381        let pending = repo.list(PurgatoryFilter::default()).expect("list");
382        assert_eq!(pending.len(), 1);
383        let posted = repo
384            .list(PurgatoryFilter { is_posted: Some(true), ..Default::default() })
385            .expect("list");
386        assert_eq!(posted.len(), 1);
387    }
388}
stateset_db/sqlite/purgatory.rs

stateset_db/sqlite/
purgatory.rs
