stateset_db/sqlite/
loyalty.rs

1//! SQLite implementation of loyalty program repository
2
3use super::{
4    map_db_error, parse_datetime_row, parse_enum_row, parse_uuid_row, with_immediate_transaction,
5};
6use chrono::Utc;
7use r2d2::Pool;
8use r2d2_sqlite::SqliteConnectionManager;
9use stateset_core::{
10    AdjustPoints, CommerceError, CreateLoyaltyProgram, CustomerId, EnrollCustomer, LoyaltyAccount,
11    LoyaltyAccountFilter, LoyaltyAccountId, LoyaltyProgram, LoyaltyProgramId,
12    LoyaltyProgramRepository, LoyaltyTransaction, LoyaltyTransactionId, Result,
13};
14
15#[derive(Debug)]
16pub struct SqliteLoyaltyProgramRepository {
17    pool: Pool<SqliteConnectionManager>,
18}
19
20impl SqliteLoyaltyProgramRepository {
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_program(row: &rusqlite::Row<'_>) -> rusqlite::Result<LoyaltyProgram> {
31        // `tiers` is a JSON array column (migration 057); older rows or a
32        // parse failure degrade to an empty tier list rather than erroring.
33        let tiers = row
34            .get::<_, Option<String>>("tiers")?
35            .and_then(|s| serde_json::from_str(&s).ok())
36            .unwrap_or_default();
37        Ok(LoyaltyProgram {
38            id: parse_uuid_row(&row.get::<_, String>("id")?, "loyalty_program", "id")?.into(),
39            name: row.get("name")?,
40            description: row.get("description")?,
41            points_per_dollar: row.get::<_, i32>("points_per_dollar")? as u32,
42            tiers,
43            status: parse_enum_row(&row.get::<_, String>("status")?, "loyalty_program", "status")?,
44            created_at: parse_datetime_row(
45                &row.get::<_, String>("created_at")?,
46                "loyalty_program",
47                "created_at",
48            )?,
49            updated_at: parse_datetime_row(
50                &row.get::<_, String>("updated_at")?,
51                "loyalty_program",
52                "updated_at",
53            )?,
54        })
55    }
56
57    fn row_to_account(row: &rusqlite::Row<'_>) -> rusqlite::Result<LoyaltyAccount> {
58        Ok(LoyaltyAccount {
59            id: parse_uuid_row(&row.get::<_, String>("id")?, "loyalty_account", "id")?.into(),
60            program_id: parse_uuid_row(
61                &row.get::<_, String>("program_id")?,
62                "loyalty_account",
63                "program_id",
64            )?
65            .into(),
66            customer_id: parse_uuid_row(
67                &row.get::<_, String>("customer_id")?,
68                "loyalty_account",
69                "customer_id",
70            )?
71            .into(),
72            points_balance: row.get::<_, i64>("points_balance")?,
73            lifetime_points: row.get::<_, i64>("lifetime_points").unwrap_or(0) as u64,
74            tier: row.get("tier")?,
75            created_at: parse_datetime_row(
76                &row.get::<_, String>("created_at")?,
77                "loyalty_account",
78                "created_at",
79            )?,
80            updated_at: parse_datetime_row(
81                &row.get::<_, String>("updated_at")?,
82                "loyalty_account",
83                "updated_at",
84            )?,
85        })
86    }
87
88    fn row_to_transaction(row: &rusqlite::Row<'_>) -> rusqlite::Result<LoyaltyTransaction> {
89        Ok(LoyaltyTransaction {
90            id: parse_uuid_row(&row.get::<_, String>("id")?, "loyalty_transaction", "id")?.into(),
91            account_id: parse_uuid_row(
92                &row.get::<_, String>("account_id")?,
93                "loyalty_transaction",
94                "account_id",
95            )?
96            .into(),
97            points: row.get("points")?,
98            transaction_type: parse_enum_row(
99                &row.get::<_, String>("type")?,
100                "loyalty_transaction",
101                "type",
102            )?,
103            reference_id: row.get("reference_id")?,
104            description: row.get("notes")?,
105            created_at: parse_datetime_row(
106                &row.get::<_, String>("created_at")?,
107                "loyalty_transaction",
108                "created_at",
109            )?,
110        })
111    }
112}
113
114impl LoyaltyProgramRepository for SqliteLoyaltyProgramRepository {
115    fn create(&self, input: CreateLoyaltyProgram) -> Result<LoyaltyProgram> {
116        let id = LoyaltyProgramId::new();
117        let now = Utc::now();
118        let id_str = id.to_string();
119        let now_str = now.to_rfc3339();
120
121        let tiers_json = serde_json::to_string(&input.tiers).unwrap_or_else(|_| "[]".to_string());
122
123        with_immediate_transaction(&self.pool, |tx| {
124            tx.execute(
125                "INSERT INTO loyalty_programs (id, name, description, points_per_dollar, status, tiers, created_at, updated_at)
126                 VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
127                rusqlite::params![
128                    &id_str,
129                    &input.name,
130                    &input.description,
131                    input.points_per_dollar as i32,
132                    "active",
133                    &tiers_json,
134                    &now_str,
135                    &now_str,
136                ],
137            )?;
138
139            tx.query_row(
140                "SELECT * FROM loyalty_programs WHERE id = ?",
141                [&id_str],
142                Self::row_to_program,
143            )
144        })
145    }
146
147    fn get(&self, id: LoyaltyProgramId) -> Result<Option<LoyaltyProgram>> {
148        let conn = self.conn()?;
149        match conn.query_row(
150            "SELECT * FROM loyalty_programs WHERE id = ?",
151            [id.to_string()],
152            Self::row_to_program,
153        ) {
154            Ok(program) => Ok(Some(program)),
155            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
156            Err(e) => Err(map_db_error(e)),
157        }
158    }
159
160    fn list(&self) -> Result<Vec<LoyaltyProgram>> {
161        let conn = self.conn()?;
162        let mut stmt = conn
163            .prepare("SELECT * FROM loyalty_programs ORDER BY created_at DESC")
164            .map_err(map_db_error)?;
165        let programs = stmt
166            .query_map([], Self::row_to_program)
167            .map_err(map_db_error)?
168            .collect::<std::result::Result<Vec<_>, _>>()
169            .map_err(map_db_error)?;
170        Ok(programs)
171    }
172
173    fn enroll(&self, input: EnrollCustomer) -> Result<LoyaltyAccount> {
174        let id = LoyaltyAccountId::new();
175        let now = Utc::now();
176        let id_str = id.to_string();
177        let now_str = now.to_rfc3339();
178
179        with_immediate_transaction(&self.pool, |tx| {
180            tx.execute(
181                "INSERT INTO loyalty_accounts (id, program_id, customer_id, points_balance, lifetime_points, tier, status, created_at, updated_at)
182                 VALUES (?, ?, ?, 0, 0, 'bronze', 'active', ?, ?)",
183                rusqlite::params![
184                    &id_str,
185                    input.program_id.to_string(),
186                    input.customer_id.to_string(),
187                    &now_str,
188                    &now_str,
189                ],
190            )?;
191
192            tx.query_row(
193                "SELECT * FROM loyalty_accounts WHERE id = ?",
194                [&id_str],
195                Self::row_to_account,
196            )
197        })
198    }
199
200    fn get_account(&self, id: LoyaltyAccountId) -> Result<Option<LoyaltyAccount>> {
201        let conn = self.conn()?;
202        match conn.query_row(
203            "SELECT * FROM loyalty_accounts WHERE id = ?",
204            [id.to_string()],
205            Self::row_to_account,
206        ) {
207            Ok(account) => Ok(Some(account)),
208            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
209            Err(e) => Err(map_db_error(e)),
210        }
211    }
212
213    fn get_account_by_customer(
214        &self,
215        customer_id: CustomerId,
216        program_id: LoyaltyProgramId,
217    ) -> Result<Option<LoyaltyAccount>> {
218        let conn = self.conn()?;
219        match conn.query_row(
220            "SELECT * FROM loyalty_accounts WHERE customer_id = ? AND program_id = ?",
221            rusqlite::params![customer_id.to_string(), program_id.to_string()],
222            Self::row_to_account,
223        ) {
224            Ok(account) => Ok(Some(account)),
225            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
226            Err(e) => Err(map_db_error(e)),
227        }
228    }
229
230    fn list_accounts(&self, filter: LoyaltyAccountFilter) -> Result<Vec<LoyaltyAccount>> {
231        let conn = self.conn()?;
232        let mut sql = "SELECT * FROM loyalty_accounts WHERE 1=1".to_string();
233        let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
234
235        if let Some(customer_id) = filter.customer_id {
236            sql.push_str(" AND customer_id = ?");
237            params.push(Box::new(customer_id.to_string()));
238        }
239        if let Some(program_id) = filter.program_id {
240            sql.push_str(" AND program_id = ?");
241            params.push(Box::new(program_id.to_string()));
242        }
243        if let Some(ref tier) = filter.tier {
244            sql.push_str(" AND tier = ?");
245            params.push(Box::new(tier.clone()));
246        }
247
248        sql.push_str(" ORDER BY created_at DESC");
249
250        crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
251
252        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
253            params.iter().map(|p| p.as_ref()).collect();
254        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
255        let accounts = stmt
256            .query_map(param_refs.as_slice(), Self::row_to_account)
257            .map_err(map_db_error)?
258            .collect::<std::result::Result<Vec<_>, _>>()
259            .map_err(map_db_error)?;
260        Ok(accounts)
261    }
262
263    fn adjust_points(&self, input: AdjustPoints) -> Result<LoyaltyTransaction> {
264        let tx_id = LoyaltyTransactionId::new();
265        let now = Utc::now();
266        let tx_id_str = tx_id.to_string();
267        let now_str = now.to_rfc3339();
268        let account_id_str = input.account_id.to_string();
269
270        with_immediate_transaction(&self.pool, |tx| {
271            // Fetch the current balance inside the transaction (errors if the
272            // account does not exist).
273            let current_balance: i64 = tx.query_row(
274                "SELECT points_balance FROM loyalty_accounts WHERE id = ?",
275                [&account_id_str],
276                |row| row.get(0),
277            )?;
278
279            let new_balance = current_balance.checked_add(input.points).ok_or_else(|| {
280                rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::ValidationError(
281                    "Points adjustment overflows".to_string(),
282                )))
283            })?;
284            if new_balance < 0 {
285                return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
286                    CommerceError::ValidationError("Insufficient points balance".to_string()),
287                )));
288            }
289
290            // Update the account balance
291            tx.execute(
292                "UPDATE loyalty_accounts SET points_balance = ?, updated_at = ? WHERE id = ?",
293                rusqlite::params![new_balance, &now_str, &account_id_str],
294            )?;
295
296            // If earning points, also increment lifetime_points
297            if input.points > 0 {
298                tx.execute(
299                    "UPDATE loyalty_accounts SET lifetime_points = lifetime_points + ? WHERE id = ?",
300                    rusqlite::params![input.points, &account_id_str],
301                )?;
302            }
303
304            // Insert the transaction record
305            tx.execute(
306                "INSERT INTO loyalty_transactions (id, account_id, points, type, reference_id, notes, created_at)
307                 VALUES (?, ?, ?, ?, ?, ?, ?)",
308                rusqlite::params![
309                    &tx_id_str,
310                    &account_id_str,
311                    input.points,
312                    input.transaction_type.to_string(),
313                    &input.reference_id,
314                    &input.description,
315                    &now_str,
316                ],
317            )?;
318
319            tx.query_row(
320                "SELECT * FROM loyalty_transactions WHERE id = ?",
321                [&tx_id_str],
322                Self::row_to_transaction,
323            )
324        })
325    }
326
327    fn get_transactions(
328        &self,
329        account_id: LoyaltyAccountId,
330        limit: Option<u32>,
331    ) -> Result<Vec<LoyaltyTransaction>> {
332        let conn = self.conn()?;
333        let mut sql =
334            "SELECT * FROM loyalty_transactions WHERE account_id = ? ORDER BY created_at DESC"
335                .to_string();
336
337        if let Some(limit) = limit {
338            sql.push_str(&format!(" LIMIT {limit}"));
339        }
340
341        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
342        let transactions = stmt
343            .query_map([account_id.to_string()], Self::row_to_transaction)
344            .map_err(map_db_error)?
345            .collect::<std::result::Result<Vec<_>, _>>()
346            .map_err(map_db_error)?;
347        Ok(transactions)
348    }
349}
350
351#[cfg(test)]
352mod tests {
353    use super::*;
354    use crate::DatabaseConfig;
355    use crate::sqlite::SqliteDatabase;
356    use stateset_core::{CustomerId, LoyaltyProgramStatus, LoyaltyTransactionType};
357
358    fn test_repo() -> SqliteLoyaltyProgramRepository {
359        let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
360        let conn = db.conn().expect("connection");
361        conn.execute_batch(
362            "CREATE TABLE IF NOT EXISTS loyalty_programs (
363                id TEXT PRIMARY KEY,
364                name TEXT NOT NULL,
365                description TEXT,
366                points_per_dollar INTEGER NOT NULL DEFAULT 1,
367                status TEXT NOT NULL DEFAULT 'active',
368                created_at TEXT NOT NULL DEFAULT (datetime('now')),
369                updated_at TEXT NOT NULL DEFAULT (datetime('now'))
370            );
371            CREATE TABLE IF NOT EXISTS loyalty_accounts (
372                id TEXT PRIMARY KEY,
373                program_id TEXT NOT NULL,
374                customer_id TEXT NOT NULL,
375                points_balance INTEGER NOT NULL DEFAULT 0,
376                lifetime_points INTEGER NOT NULL DEFAULT 0,
377                tier TEXT NOT NULL DEFAULT 'bronze',
378                status TEXT NOT NULL DEFAULT 'active',
379                created_at TEXT NOT NULL DEFAULT (datetime('now')),
380                updated_at TEXT NOT NULL DEFAULT (datetime('now'))
381            );
382            CREATE TABLE IF NOT EXISTS loyalty_transactions (
383                id TEXT PRIMARY KEY,
384                account_id TEXT NOT NULL,
385                points INTEGER NOT NULL,
386                type TEXT NOT NULL,
387                reference_id TEXT,
388                notes TEXT,
389                created_at TEXT NOT NULL DEFAULT (datetime('now'))
390            );",
391        )
392        .expect("create tables");
393        SqliteLoyaltyProgramRepository::new(db.pool().clone())
394    }
395
396    #[test]
397    fn create_program() {
398        let repo = test_repo();
399        let program = repo
400            .create(CreateLoyaltyProgram {
401                name: "Rewards Plus".into(),
402                description: Some("Earn points on every purchase".into()),
403                points_per_dollar: 2,
404                tiers: vec![],
405            })
406            .expect("create program");
407
408        assert_eq!(program.name, "Rewards Plus");
409        assert_eq!(program.points_per_dollar, 2);
410        assert_eq!(program.status, LoyaltyProgramStatus::Active);
411        assert_eq!(program.description.as_deref(), Some("Earn points on every purchase"));
412
413        let fetched = repo.get(program.id).expect("get program").expect("found");
414        assert_eq!(fetched.id, program.id);
415        assert_eq!(fetched.name, "Rewards Plus");
416
417        let all = repo.list().expect("list programs");
418        assert_eq!(all.len(), 1);
419    }
420
421    #[test]
422    fn enroll_customer() {
423        let repo = test_repo();
424        let program = repo
425            .create(CreateLoyaltyProgram {
426                name: "Test Program".into(),
427                description: None,
428                points_per_dollar: 1,
429                tiers: vec![],
430            })
431            .expect("create program");
432
433        let customer_id = CustomerId::new();
434        let account = repo
435            .enroll(EnrollCustomer { customer_id, program_id: program.id })
436            .expect("enroll customer");
437
438        assert_eq!(account.customer_id, customer_id);
439        assert_eq!(account.program_id, program.id);
440        assert_eq!(account.points_balance, 0);
441        assert_eq!(account.tier, "bronze");
442
443        let fetched = repo.get_account(account.id).expect("get account").expect("found");
444        assert_eq!(fetched.id, account.id);
445
446        let by_customer = repo
447            .get_account_by_customer(customer_id, program.id)
448            .expect("get by customer")
449            .expect("found");
450        assert_eq!(by_customer.id, account.id);
451
452        let accounts = repo
453            .list_accounts(LoyaltyAccountFilter {
454                program_id: Some(program.id),
455                ..Default::default()
456            })
457            .expect("list accounts");
458        assert_eq!(accounts.len(), 1);
459    }
460
461    #[test]
462    fn adjust_points() {
463        let repo = test_repo();
464        let program = repo
465            .create(CreateLoyaltyProgram {
466                name: "Points Program".into(),
467                description: None,
468                points_per_dollar: 1,
469                tiers: vec![],
470            })
471            .expect("create program");
472
473        let account = repo
474            .enroll(EnrollCustomer { customer_id: CustomerId::new(), program_id: program.id })
475            .expect("enroll");
476
477        // Earn 100 points
478        let earn_tx = repo
479            .adjust_points(AdjustPoints {
480                account_id: account.id,
481                points: 100,
482                transaction_type: LoyaltyTransactionType::Earn,
483                reference_id: Some("order-123".into()),
484                description: Some("Purchase points".into()),
485            })
486            .expect("earn points");
487
488        assert_eq!(earn_tx.points, 100);
489        assert_eq!(earn_tx.transaction_type, LoyaltyTransactionType::Earn);
490        assert_eq!(earn_tx.reference_id.as_deref(), Some("order-123"));
491
492        let updated = repo.get_account(account.id).expect("get").expect("found");
493        assert_eq!(updated.points_balance, 100);
494        assert_eq!(updated.lifetime_points, 100);
495
496        // Redeem 30 points
497        repo.adjust_points(AdjustPoints {
498            account_id: account.id,
499            points: -30,
500            transaction_type: LoyaltyTransactionType::Redeem,
501            reference_id: None,
502            description: Some("Reward redemption".into()),
503        })
504        .expect("redeem points");
505
506        let after_redeem = repo.get_account(account.id).expect("get").expect("found");
507        assert_eq!(after_redeem.points_balance, 70);
508        // lifetime_points should NOT decrease on redemption
509        assert_eq!(after_redeem.lifetime_points, 100);
510
511        // Verify transaction history
512        let txns = repo.get_transactions(account.id, None).expect("get transactions");
513        assert_eq!(txns.len(), 2);
514        // Most recent first (ORDER BY created_at DESC)
515        assert_eq!(txns[0].transaction_type, LoyaltyTransactionType::Redeem);
516        assert_eq!(txns[1].transaction_type, LoyaltyTransactionType::Earn);
517    }
518
519    #[test]
520    fn adjust_points_rejects_overdraft() {
521        let repo = test_repo();
522        let program = repo
523            .create(CreateLoyaltyProgram {
524                name: "Overdraft Program".into(),
525                description: None,
526                points_per_dollar: 1,
527                tiers: vec![],
528            })
529            .expect("create program");
530        let account = repo
531            .enroll(EnrollCustomer { customer_id: CustomerId::new(), program_id: program.id })
532            .expect("enroll");
533
534        repo.adjust_points(AdjustPoints {
535            account_id: account.id,
536            points: 50,
537            transaction_type: LoyaltyTransactionType::Earn,
538            reference_id: None,
539            description: None,
540        })
541        .expect("earn");
542
543        // Redeeming more than the balance must fail, not go negative.
544        assert!(
545            repo.adjust_points(AdjustPoints {
546                account_id: account.id,
547                points: -100,
548                transaction_type: LoyaltyTransactionType::Redeem,
549                reference_id: None,
550                description: None,
551            })
552            .is_err()
553        );
554
555        let fetched = repo.get_account(account.id).expect("get").expect("found");
556        assert_eq!(fetched.points_balance, 50);
557        // No transaction record for the rejected redemption.
558        assert_eq!(repo.get_transactions(account.id, None).expect("txns").len(), 1);
559    }
560
561    #[test]
562    fn concurrent_redemptions_cannot_overdraw() {
563        use std::sync::{Arc, Barrier};
564        use std::thread;
565
566        let db = Arc::new(SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("db"));
567        let repo = SqliteLoyaltyProgramRepository::new(db.pool().clone());
568        let program = repo
569            .create(CreateLoyaltyProgram {
570                name: "Race Program".into(),
571                description: None,
572                points_per_dollar: 1,
573                tiers: vec![],
574            })
575            .expect("create program");
576        let account = repo
577            .enroll(EnrollCustomer { customer_id: CustomerId::new(), program_id: program.id })
578            .expect("enroll");
579        repo.adjust_points(AdjustPoints {
580            account_id: account.id,
581            points: 50,
582            transaction_type: LoyaltyTransactionType::Earn,
583            reference_id: None,
584            description: None,
585        })
586        .expect("earn");
587
588        let thread_count = 10;
589        let barrier = Arc::new(Barrier::new(thread_count));
590        let mut handles = Vec::new();
591        for _ in 0..thread_count {
592            let db = Arc::clone(&db);
593            let barrier = Arc::clone(&barrier);
594            let account_id = account.id;
595            handles.push(thread::spawn(move || {
596                let repo = SqliteLoyaltyProgramRepository::new(db.pool().clone());
597                barrier.wait();
598                repo.adjust_points(AdjustPoints {
599                    account_id,
600                    points: -30,
601                    transaction_type: LoyaltyTransactionType::Redeem,
602                    reference_id: None,
603                    description: None,
604                })
605            }));
606        }
607
608        let results: Vec<_> = handles.into_iter().map(|h| h.join().expect("thread")).collect();
609        let successes = results.iter().filter(|r| r.is_ok()).count();
610        // The 50-point balance can fund at most one 30-point redemption — the
611        // safety invariant is that no more than one succeeds (never overdrawn).
612        // Under extreme lock contention the sole winner can fail with a retryable
613        // "table is locked" error, so zero successes is acceptable; two or more
614        // would be an overdraw bug.
615        assert!(successes <= 1, "points overdrawn under concurrency: {results:?}");
616
617        let fetched = repo.get_account(account.id).expect("get").expect("found");
618        assert_eq!(
619            fetched.points_balance,
620            50 - 30 * successes as i64,
621            "points must reflect exactly the successful redemption: {results:?}"
622        );
623    }
624
625    #[test]
626    fn adjust_points_rejects_unknown_account() {
627        let repo = test_repo();
628        let ghost = LoyaltyAccountId::new();
629
630        assert!(
631            repo.adjust_points(AdjustPoints {
632                account_id: ghost,
633                points: 100,
634                transaction_type: LoyaltyTransactionType::Earn,
635                reference_id: None,
636                description: None,
637            })
638            .is_err()
639        );
640
641        // No orphaned transaction record for the nonexistent account.
642        assert!(repo.get_transactions(ghost, None).expect("txns").is_empty());
643    }
644}
stateset_db/sqlite/loyalty.rs

stateset_db/sqlite/
loyalty.rs
