stateset_db/sqlite/
price_schedules.rs

1//! SQLite implementation of the price schedule repository
2
3use super::{
4    map_db_error, parse_datetime_opt_row, parse_datetime_row, parse_decimal_row, parse_enum_row,
5    parse_uuid_row, with_immediate_transaction,
6};
7use chrono::{DateTime, Utc};
8use r2d2::Pool;
9use r2d2_sqlite::SqliteConnectionManager;
10use rust_decimal::Decimal;
11use stateset_core::{
12    CommerceError, CreatePriceSchedule, CurrencyCode, PriceSchedule, PriceScheduleEntry,
13    PriceScheduleFilter, PriceScheduleId, PriceScheduleRepository, ProductId, Result,
14    UpdatePriceSchedule,
15};
16
17#[derive(Debug)]
18pub struct SqlitePriceScheduleRepository {
19    pool: Pool<SqliteConnectionManager>,
20}
21
22impl SqlitePriceScheduleRepository {
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_schedule(row: &rusqlite::Row<'_>) -> rusqlite::Result<PriceSchedule> {
33        Ok(PriceSchedule {
34            id: parse_uuid_row(&row.get::<_, String>("id")?, "price_schedule", "id")?.into(),
35            name: row.get("name")?,
36            code: row.get("code")?,
37            currency: parse_enum_row::<CurrencyCode>(
38                &row.get::<_, String>("currency")?,
39                "price_schedule",
40                "currency",
41            )?,
42            starts_at: parse_datetime_opt_row(
43                row.get::<_, Option<String>>("starts_at")?,
44                "price_schedule",
45                "starts_at",
46            )?,
47            ends_at: parse_datetime_opt_row(
48                row.get::<_, Option<String>>("ends_at")?,
49                "price_schedule",
50                "ends_at",
51            )?,
52            is_active: row.get::<_, i32>("is_active")? != 0,
53            priority: row.get("priority")?,
54            created_at: parse_datetime_row(
55                &row.get::<_, String>("created_at")?,
56                "price_schedule",
57                "created_at",
58            )?,
59            updated_at: parse_datetime_row(
60                &row.get::<_, String>("updated_at")?,
61                "price_schedule",
62                "updated_at",
63            )?,
64        })
65    }
66
67    fn row_to_entry(row: &rusqlite::Row<'_>) -> rusqlite::Result<PriceScheduleEntry> {
68        Ok(PriceScheduleEntry {
69            price_schedule_id: parse_uuid_row(
70                &row.get::<_, String>("price_schedule_id")?,
71                "price_schedule_entry",
72                "price_schedule_id",
73            )?
74            .into(),
75            product_id: parse_uuid_row(
76                &row.get::<_, String>("product_id")?,
77                "price_schedule_entry",
78                "product_id",
79            )?
80            .into(),
81            price: parse_decimal_row(
82                &row.get::<_, String>("price")?,
83                "price_schedule_entry",
84                "price",
85            )?,
86            created_at: parse_datetime_row(
87                &row.get::<_, String>("created_at")?,
88                "price_schedule_entry",
89                "created_at",
90            )?,
91            updated_at: parse_datetime_row(
92                &row.get::<_, String>("updated_at")?,
93                "price_schedule_entry",
94                "updated_at",
95            )?,
96        })
97    }
98}
99
100impl PriceScheduleRepository for SqlitePriceScheduleRepository {
101    fn create(&self, input: CreatePriceSchedule) -> Result<PriceSchedule> {
102        let id = PriceScheduleId::new();
103        let id_str = id.to_string();
104        let now_str = Utc::now().to_rfc3339();
105        let currency = input.currency.unwrap_or(CurrencyCode::USD);
106        with_immediate_transaction(&self.pool, |tx| {
107            tx.execute(
108                "INSERT INTO price_schedules (id, name, code, currency, starts_at, ends_at, is_active, priority, created_at, updated_at)
109                 VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?, ?)",
110                rusqlite::params![
111                    &id_str,
112                    &input.name,
113                    &input.code,
114                    currency.to_string(),
115                    input.starts_at.map(|d| d.to_rfc3339()),
116                    input.ends_at.map(|d| d.to_rfc3339()),
117                    input.priority,
118                    &now_str,
119                    &now_str,
120                ],
121            )?;
122            tx.query_row(
123                "SELECT * FROM price_schedules WHERE id = ?",
124                [&id_str],
125                Self::row_to_schedule,
126            )
127        })
128    }
129
130    fn get(&self, id: PriceScheduleId) -> Result<Option<PriceSchedule>> {
131        let conn = self.conn()?;
132        match conn.query_row(
133            "SELECT * FROM price_schedules WHERE id = ?",
134            [id.to_string()],
135            Self::row_to_schedule,
136        ) {
137            Ok(s) => Ok(Some(s)),
138            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
139            Err(e) => Err(map_db_error(e)),
140        }
141    }
142
143    fn update(&self, id: PriceScheduleId, input: UpdatePriceSchedule) -> Result<PriceSchedule> {
144        let id_str = id.to_string();
145        let now_str = Utc::now().to_rfc3339();
146        with_immediate_transaction(&self.pool, |tx| {
147            let mut sets = vec!["updated_at = ?".to_string()];
148            let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(now_str.clone())];
149
150            if let Some(ref name) = input.name {
151                sets.push("name = ?".into());
152                params.push(Box::new(name.clone()));
153            }
154            if let Some(ref code) = input.code {
155                sets.push("code = ?".into());
156                params.push(Box::new(code.clone()));
157            }
158            if let Some(starts_at) = input.starts_at {
159                sets.push("starts_at = ?".into());
160                params.push(Box::new(starts_at.to_rfc3339()));
161            }
162            if let Some(ends_at) = input.ends_at {
163                sets.push("ends_at = ?".into());
164                params.push(Box::new(ends_at.to_rfc3339()));
165            }
166            if let Some(is_active) = input.is_active {
167                sets.push("is_active = ?".into());
168                params.push(Box::new(is_active as i32));
169            }
170            if let Some(priority) = input.priority {
171                sets.push("priority = ?".into());
172                params.push(Box::new(priority));
173            }
174
175            let sql = format!("UPDATE price_schedules SET {} WHERE id = ?", sets.join(", "));
176            params.push(Box::new(id_str.clone()));
177            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
178                params.iter().map(|p| p.as_ref()).collect();
179            tx.execute(&sql, param_refs.as_slice())?;
180
181            tx.query_row(
182                "SELECT * FROM price_schedules WHERE id = ?",
183                [&id_str],
184                Self::row_to_schedule,
185            )
186        })
187    }
188
189    fn list(&self, filter: PriceScheduleFilter) -> Result<Vec<PriceSchedule>> {
190        let conn = self.conn()?;
191        let mut sql = "SELECT * FROM price_schedules WHERE 1=1".to_string();
192        let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
193        if let Some(active) = filter.is_active {
194            sql.push_str(" AND is_active = ?");
195            params.push(Box::new(active as i32));
196        }
197        sql.push_str(" ORDER BY priority DESC, created_at DESC");
198        crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
199        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
200            params.iter().map(|p| p.as_ref()).collect();
201        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
202        let rows = stmt
203            .query_map(param_refs.as_slice(), Self::row_to_schedule)
204            .map_err(map_db_error)?
205            .collect::<std::result::Result<Vec<_>, _>>()
206            .map_err(map_db_error)?;
207        Ok(rows)
208    }
209
210    fn delete(&self, id: PriceScheduleId) -> Result<()> {
211        let id_str = id.to_string();
212        with_immediate_transaction(&self.pool, |tx| {
213            tx.execute(
214                "DELETE FROM price_schedule_entries WHERE price_schedule_id = ?",
215                [&id_str],
216            )?;
217            tx.execute("DELETE FROM price_schedules WHERE id = ?", [&id_str])?;
218            Ok(())
219        })
220    }
221
222    fn set_entry(
223        &self,
224        id: PriceScheduleId,
225        product_id: ProductId,
226        price: Decimal,
227    ) -> Result<PriceScheduleEntry> {
228        let id_str = id.to_string();
229        let product_str = product_id.to_string();
230        let now_str = Utc::now().to_rfc3339();
231        with_immediate_transaction(&self.pool, |tx| {
232            tx.execute(
233                "INSERT INTO price_schedule_entries (price_schedule_id, product_id, price, created_at, updated_at)
234                 VALUES (?, ?, ?, ?, ?)
235                 ON CONFLICT(price_schedule_id, product_id) DO UPDATE SET
236                    price = excluded.price,
237                    updated_at = excluded.updated_at",
238                rusqlite::params![&id_str, &product_str, price.to_string(), &now_str, &now_str],
239            )?;
240            tx.query_row(
241                "SELECT * FROM price_schedule_entries WHERE price_schedule_id = ? AND product_id = ?",
242                rusqlite::params![&id_str, &product_str],
243                Self::row_to_entry,
244            )
245        })
246    }
247
248    fn delete_entry(&self, id: PriceScheduleId, product_id: ProductId) -> Result<()> {
249        let conn = self.conn()?;
250        conn.execute(
251            "DELETE FROM price_schedule_entries WHERE price_schedule_id = ? AND product_id = ?",
252            rusqlite::params![id.to_string(), product_id.to_string()],
253        )
254        .map_err(map_db_error)?;
255        Ok(())
256    }
257
258    fn list_entries(&self, id: PriceScheduleId) -> Result<Vec<PriceScheduleEntry>> {
259        let conn = self.conn()?;
260        let mut stmt = conn
261            .prepare("SELECT * FROM price_schedule_entries WHERE price_schedule_id = ? ORDER BY product_id")
262            .map_err(map_db_error)?;
263        let rows = stmt
264            .query_map([id.to_string()], Self::row_to_entry)
265            .map_err(map_db_error)?
266            .collect::<std::result::Result<Vec<_>, _>>()
267            .map_err(map_db_error)?;
268        Ok(rows)
269    }
270
271    fn resolve_price(&self, product_id: ProductId, at: DateTime<Utc>) -> Result<Option<Decimal>> {
272        // Active schedules ordered by priority then recency; first whose window
273        // contains `at` and that has an entry for the product wins.
274        let schedules =
275            self.list(PriceScheduleFilter { is_active: Some(true), ..Default::default() })?;
276        let conn = self.conn()?;
277        for schedule in schedules {
278            if !schedule.is_effective_at(at) {
279                continue;
280            }
281            let price: Option<String> = conn
282                .query_row(
283                    "SELECT price FROM price_schedule_entries WHERE price_schedule_id = ? AND product_id = ?",
284                    rusqlite::params![schedule.id.to_string(), product_id.to_string()],
285                    |r| r.get(0),
286                )
287                .ok();
288            if let Some(price_str) = price {
289                let price = price_str
290                    .parse::<Decimal>()
291                    .map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
292                return Ok(Some(price));
293            }
294        }
295        Ok(None)
296    }
297}
298
299#[cfg(test)]
300mod tests {
301    use super::*;
302    use crate::DatabaseConfig;
303    use crate::sqlite::SqliteDatabase;
304    use chrono::TimeZone;
305    use rust_decimal_macros::dec;
306
307    fn at(y: i32, m: u32, d: u32) -> DateTime<Utc> {
308        Utc.with_ymd_and_hms(y, m, d, 12, 0, 0).unwrap()
309    }
310
311    fn test_repo() -> SqlitePriceScheduleRepository {
312        let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
313        SqlitePriceScheduleRepository::new(db.pool().clone())
314    }
315
316    fn new_schedule(
317        repo: &SqlitePriceScheduleRepository,
318        starts: DateTime<Utc>,
319        ends: DateTime<Utc>,
320        priority: i32,
321    ) -> PriceSchedule {
322        repo.create(CreatePriceSchedule {
323            name: "Sale".into(),
324            code: None,
325            currency: Some(CurrencyCode::USD),
326            starts_at: Some(starts),
327            ends_at: Some(ends),
328            priority,
329        })
330        .expect("create schedule")
331    }
332
333    #[test]
334    fn create_update_entries() {
335        let repo = test_repo();
336        let s = new_schedule(&repo, at(2026, 6, 1), at(2026, 6, 30), 0);
337        let product = ProductId::new();
338        let entry = repo.set_entry(s.id, product, dec!(19.99)).expect("entry");
339        assert_eq!(entry.price, dec!(19.99));
340        repo.set_entry(s.id, product, dec!(15.00)).expect("upsert");
341        assert_eq!(repo.list_entries(s.id).expect("entries").len(), 1);
342        repo.delete_entry(s.id, product).expect("delete entry");
343        assert_eq!(repo.list_entries(s.id).expect("entries").len(), 0);
344    }
345
346    #[test]
347    fn resolve_within_window() {
348        let repo = test_repo();
349        let s = new_schedule(&repo, at(2026, 6, 1), at(2026, 6, 30), 0);
350        let product = ProductId::new();
351        repo.set_entry(s.id, product, dec!(9.99)).expect("entry");
352        assert_eq!(
353            repo.resolve_price(product, at(2026, 6, 15)).expect("resolve"),
354            Some(dec!(9.99))
355        );
356        // outside window → None
357        assert_eq!(repo.resolve_price(product, at(2026, 7, 15)).expect("resolve"), None);
358    }
359
360    #[test]
361    fn resolve_highest_priority_wins() {
362        let repo = test_repo();
363        let low = new_schedule(&repo, at(2026, 6, 1), at(2026, 6, 30), 1);
364        let high = new_schedule(&repo, at(2026, 6, 1), at(2026, 6, 30), 10);
365        let product = ProductId::new();
366        repo.set_entry(low.id, product, dec!(20)).expect("low");
367        repo.set_entry(high.id, product, dec!(12)).expect("high");
368        assert_eq!(repo.resolve_price(product, at(2026, 6, 15)).expect("resolve"), Some(dec!(12)));
369    }
370
371    #[test]
372    fn delete_cascades_entries() {
373        let repo = test_repo();
374        let s = new_schedule(&repo, at(2026, 6, 1), at(2026, 6, 30), 0);
375        repo.set_entry(s.id, ProductId::new(), dec!(5)).expect("entry");
376        repo.delete(s.id).expect("delete");
377        assert!(repo.get(s.id).expect("get").is_none());
378        assert_eq!(repo.list_entries(s.id).expect("entries").len(), 0);
379    }
380}
stateset_db/sqlite/price_schedules.rs

stateset_db/sqlite/
price_schedules.rs
