stateset_db/sqlite/
customers.rs

1//! SQLite customer repository implementation
2
3use super::{
4    build_in_clause, json1_available, map_db_error, params_refs, parse_datetime_row, parse_enum,
5    parse_enum_row, parse_json_row, parse_uuid_opt_row, parse_uuid_row, uuid_params,
6    with_immediate_transaction,
7};
8use chrono::Utc;
9use r2d2::Pool;
10use r2d2_sqlite::SqliteConnectionManager;
11use rusqlite::OptionalExtension;
12use stateset_core::{
13    AddressType, BatchResult, CommerceError, CreateCustomer, CreateCustomerAddress, Customer,
14    CustomerAddress, CustomerFilter, CustomerId, CustomerRepository, CustomerStatus, Result,
15    UpdateCustomer, validate_batch_size, validate_email, validate_phone, validate_postal_code,
16    validate_required_text, validate_required_uuid,
17};
18use uuid::Uuid;
19
20/// SQLite implementation of `CustomerRepository`
21#[derive(Debug)]
22pub struct SqliteCustomerRepository {
23    pool: Pool<SqliteConnectionManager>,
24}
25
26impl SqliteCustomerRepository {
27    #[must_use]
28    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
29        Self { pool }
30    }
31
32    fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
33        self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
34    }
35
36    fn row_to_customer(row: &rusqlite::Row<'_>) -> rusqlite::Result<Customer> {
37        let tags_json: String = row.get("tags")?;
38        let metadata_json: Option<String> = row.get("metadata")?;
39
40        Ok(Customer {
41            id: CustomerId::from(parse_uuid_row(&row.get::<_, String>("id")?, "customer", "id")?),
42            email: row.get("email")?,
43            first_name: row.get("first_name")?,
44            last_name: row.get("last_name")?,
45            phone: row.get("phone")?,
46            status: parse_enum_row(&row.get::<_, String>("status")?, "customer", "status")?,
47            accepts_marketing: row.get::<_, i32>("accepts_marketing")? != 0,
48            email_verified: row.get::<_, i32>("email_verified")? != 0,
49            tags: parse_json_row(&tags_json, "customer", "tags")?,
50            metadata: metadata_json
51                .map(|s| parse_json_row(&s, "customer", "metadata"))
52                .transpose()?,
53            default_shipping_address_id: parse_uuid_opt_row(
54                row.get::<_, Option<String>>("default_shipping_address_id")?,
55                "customer",
56                "default_shipping_address_id",
57            )?,
58            default_billing_address_id: parse_uuid_opt_row(
59                row.get::<_, Option<String>>("default_billing_address_id")?,
60                "customer",
61                "default_billing_address_id",
62            )?,
63            created_at: parse_datetime_row(
64                &row.get::<_, String>("created_at")?,
65                "customer",
66                "created_at",
67            )?,
68            updated_at: parse_datetime_row(
69                &row.get::<_, String>("updated_at")?,
70                "customer",
71                "updated_at",
72            )?,
73        })
74    }
75
76    fn row_to_address(row: &rusqlite::Row<'_>) -> rusqlite::Result<CustomerAddress> {
77        Ok(CustomerAddress {
78            id: parse_uuid_row(&row.get::<_, String>("id")?, "customer_address", "id")?,
79            customer_id: CustomerId::from(parse_uuid_row(
80                &row.get::<_, String>("customer_id")?,
81                "customer_address",
82                "customer_id",
83            )?),
84            address_type: parse_enum_row(
85                &row.get::<_, String>("address_type")?,
86                "customer_address",
87                "address_type",
88            )?,
89            first_name: row.get("first_name")?,
90            last_name: row.get("last_name")?,
91            company: row.get("company")?,
92            line1: row.get("line1")?,
93            line2: row.get("line2")?,
94            city: row.get("city")?,
95            state: row.get("state")?,
96            postal_code: row.get("postal_code")?,
97            country: row.get("country")?,
98            phone: row.get("phone")?,
99            is_default: row.get::<_, i32>("is_default")? != 0,
100            created_at: parse_datetime_row(
101                &row.get::<_, String>("created_at")?,
102                "customer_address",
103                "created_at",
104            )?,
105            updated_at: parse_datetime_row(
106                &row.get::<_, String>("updated_at")?,
107                "customer_address",
108                "updated_at",
109            )?,
110        })
111    }
112
113    fn validate_address_input(input: &CreateCustomerAddress) -> Result<()> {
114        validate_required_uuid("customer_address.customer_id", input.customer_id.into_uuid())?;
115        validate_required_text("customer_address.first_name", &input.first_name, 100)?;
116        validate_required_text("customer_address.last_name", &input.last_name, 100)?;
117        validate_required_text("customer_address.line1", &input.line1, 255)?;
118        validate_required_text("customer_address.city", &input.city, 255)?;
119        validate_postal_code(&input.postal_code)?;
120        validate_required_text("customer_address.country", &input.country, 64)?;
121
122        if let Some(line2) = &input.line2 {
123            validate_required_text("customer_address.line2", line2, 255)?;
124        }
125        if let Some(state) = &input.state {
126            validate_required_text("customer_address.state", state, 64)?;
127        }
128        if let Some(phone) = &input.phone {
129            validate_phone(phone)?;
130        }
131
132        Ok(())
133    }
134}
135
136impl CustomerRepository for SqliteCustomerRepository {
137    fn create(&self, input: CreateCustomer) -> Result<Customer> {
138        // Validate email format
139        validate_email(&input.email)?;
140        validate_required_text("customer.first_name", &input.first_name, 100)?;
141        validate_required_text("customer.last_name", &input.last_name, 100)?;
142        if let Some(phone) = &input.phone {
143            validate_phone(phone)?;
144        }
145
146        let id = CustomerId::new();
147        let tags = input.tags.clone().unwrap_or_default();
148        let metadata = input.metadata.clone();
149        let email = input.email.clone();
150        let first_name = input.first_name.clone();
151        let last_name = input.last_name.clone();
152        let phone = input.phone.clone();
153        let accepts_marketing = input.accepts_marketing.unwrap_or(false);
154
155        with_immediate_transaction(&self.pool, |tx| {
156            let now = Utc::now();
157
158            // Check email uniqueness (EXISTS is faster than COUNT(*))
159            let exists: bool = tx.query_row(
160                "SELECT EXISTS(SELECT 1 FROM customers WHERE email = ? LIMIT 1)",
161                [&email],
162                |row| row.get(0),
163            )?;
164
165            if exists {
166                return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
167                    CommerceError::EmailAlreadyExists(email.clone()),
168                )));
169            }
170
171            let tags_json = serde_json::to_string(&tags).unwrap_or_default();
172            let metadata_json =
173                metadata.as_ref().map(|m| serde_json::to_string(m).unwrap_or_default());
174            let id_str = id.to_string();
175            let now_str = now.to_rfc3339();
176
177            tx.prepare_cached(
178                "INSERT INTO customers (id, email, first_name, last_name, phone, status,
179                                        accepts_marketing, email_verified, tags, metadata,
180                                        created_at, updated_at)
181                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
182            )?
183            .execute(rusqlite::params![
184                &id_str,
185                &email,
186                &first_name,
187                &last_name,
188                &phone,
189                "active",
190                i32::from(accepts_marketing),
191                0,
192                tags_json,
193                metadata_json,
194                &now_str,
195                &now_str,
196            ])?;
197
198            // Clone values for the return since Fn closure may be called multiple times
199            Ok(Customer {
200                id,
201                email: email.clone(),
202                first_name: first_name.clone(),
203                last_name: last_name.clone(),
204                phone: phone.clone(),
205                status: CustomerStatus::Active,
206                accepts_marketing,
207                email_verified: false,
208                tags: tags.clone(),
209                metadata: metadata.clone(),
210                default_shipping_address_id: None,
211                default_billing_address_id: None,
212                created_at: now,
213                updated_at: now,
214            })
215        })
216    }
217
218    fn get(&self, id: CustomerId) -> Result<Option<Customer>> {
219        let conn = self.conn()?;
220        let result = conn.query_row(
221            "SELECT * FROM customers WHERE id = ?",
222            [id.to_string()],
223            Self::row_to_customer,
224        );
225
226        match result {
227            Ok(customer) => Ok(Some(customer)),
228            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
229            Err(e) => Err(map_db_error(e)),
230        }
231    }
232
233    fn get_by_email(&self, email: &str) -> Result<Option<Customer>> {
234        let conn = self.conn()?;
235        let result = conn.query_row(
236            "SELECT * FROM customers WHERE email = ?",
237            [email],
238            Self::row_to_customer,
239        );
240
241        match result {
242            Ok(customer) => Ok(Some(customer)),
243            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
244            Err(e) => Err(map_db_error(e)),
245        }
246    }
247
248    fn update(&self, id: CustomerId, input: UpdateCustomer) -> Result<Customer> {
249        let conn = self.conn()?;
250        let now = Utc::now();
251        let current_version: i32 = conn
252            .query_row("SELECT version FROM customers WHERE id = ?", [id.to_string()], |row| {
253                row.get(0)
254            })
255            .map_err(|e| match e {
256                rusqlite::Error::QueryReturnedNoRows => {
257                    CommerceError::CustomerNotFound(id.into_uuid())
258                }
259                e => map_db_error(e),
260            })?;
261
262        let mut updates = vec!["updated_at = ?"];
263        let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(now.to_rfc3339())];
264
265        if let Some(email) = &input.email {
266            validate_email(email)?;
267            let existing_id: Option<String> = conn
268                .query_row("SELECT id FROM customers WHERE email = ?", [email], |row| row.get(0))
269                .optional()
270                .map_err(map_db_error)?;
271            if let Some(existing_id) = existing_id {
272                if existing_id != id.to_string() {
273                    return Err(CommerceError::EmailAlreadyExists(email.clone()));
274                }
275            }
276            updates.push("email = ?");
277            params.push(Box::new(email.clone()));
278        }
279        if let Some(first_name) = &input.first_name {
280            validate_required_text("customer.first_name", first_name, 100)?;
281            updates.push("first_name = ?");
282            params.push(Box::new(first_name.clone()));
283        }
284        if let Some(last_name) = &input.last_name {
285            validate_required_text("customer.last_name", last_name, 100)?;
286            updates.push("last_name = ?");
287            params.push(Box::new(last_name.clone()));
288        }
289        if let Some(phone) = &input.phone {
290            validate_phone(phone)?;
291            updates.push("phone = ?");
292            params.push(Box::new(phone.clone()));
293        }
294        if let Some(status) = &input.status {
295            updates.push("status = ?");
296            params.push(Box::new(status.to_string()));
297        }
298        if let Some(accepts_marketing) = &input.accepts_marketing {
299            updates.push("accepts_marketing = ?");
300            params.push(Box::new(i32::from(*accepts_marketing)));
301        }
302        if let Some(tags) = &input.tags {
303            updates.push("tags = ?");
304            params.push(Box::new(serde_json::to_string(tags).unwrap_or_default()));
305        }
306        if let Some(metadata) = &input.metadata {
307            updates.push("metadata = ?");
308            params.push(Box::new(serde_json::to_string(metadata).unwrap_or_default()));
309        }
310
311        updates.push("version = version + 1");
312        params.push(Box::new(id.to_string()));
313        params.push(Box::new(current_version));
314
315        let sql =
316            format!("UPDATE customers SET {} WHERE id = ? AND version = ?", updates.join(", "));
317        let params_refs: Vec<&dyn rusqlite::ToSql> =
318            params.iter().map(std::convert::AsRef::as_ref).collect();
319
320        let rows_affected = conn.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
321        if rows_affected == 0 {
322            return Err(CommerceError::VersionConflict {
323                entity: "customer".to_string(),
324                id: id.to_string(),
325                expected_version: current_version,
326            });
327        }
328
329        self.get(id)?.ok_or(CommerceError::CustomerNotFound(id.into_uuid()))
330    }
331
332    fn list(&self, filter: CustomerFilter) -> Result<Vec<Customer>> {
333        let conn = self.conn()?;
334        let use_json = json1_available(&conn);
335        let mut sql = "SELECT * FROM customers WHERE 1=1".to_string();
336        let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![];
337
338        if let Some(email) = &filter.email {
339            sql.push_str(" AND email LIKE ?");
340            params.push(Box::new(format!("%{email}%")));
341        }
342        if let Some(status) = &filter.status {
343            sql.push_str(" AND status = ?");
344            params.push(Box::new(status.to_string()));
345        } else {
346            sql.push_str(" AND status != 'deleted'");
347        }
348        if let Some(tag) = &filter.tag {
349            if use_json {
350                sql.push_str(" AND EXISTS (SELECT 1 FROM json_each(tags) WHERE value = ?)");
351                params.push(Box::new(tag.clone()));
352            } else {
353                sql.push_str(" AND tags LIKE ?");
354                params.push(Box::new(format!("%\"{tag}\"%")));
355            }
356        }
357        if let Some(accepts_marketing) = &filter.accepts_marketing {
358            sql.push_str(" AND accepts_marketing = ?");
359            params.push(Box::new(i32::from(*accepts_marketing)));
360        }
361
362        // Keyset cursor: (created_at, id) for stable DESC ordering
363        if let Some((cursor_date, cursor_id)) = &filter.after_cursor {
364            sql.push_str(" AND (created_at < ? OR (created_at = ? AND id < ?))");
365            params.push(Box::new(cursor_date.clone()));
366            params.push(Box::new(cursor_date.clone()));
367            params.push(Box::new(cursor_id.clone()));
368        }
369
370        sql.push_str(" ORDER BY created_at DESC, id DESC");
371
372        // Offset pagination applies only in non-cursor mode; the helper emits
373        // `LIMIT -1 OFFSET n` when an offset is set without a limit (SQLite rejects
374        // a bare OFFSET).
375        let offset = if filter.after_cursor.is_none() { filter.offset } else { None };
376        crate::sqlite::append_limit_offset(&mut sql, filter.limit, offset);
377
378        let params_refs: Vec<&dyn rusqlite::ToSql> =
379            params.iter().map(std::convert::AsRef::as_ref).collect();
380        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
381
382        let customers = stmt
383            .query_map(params_refs.as_slice(), Self::row_to_customer)
384            .map_err(map_db_error)?
385            .collect::<rusqlite::Result<Vec<_>>>()
386            .map_err(map_db_error)?;
387
388        Ok(customers)
389    }
390
391    fn delete(&self, id: CustomerId) -> Result<()> {
392        let conn = self.conn()?;
393        conn.execute(
394            "UPDATE customers SET status = ?, updated_at = ? WHERE id = ?",
395            rusqlite::params!["deleted", Utc::now().to_rfc3339(), id.to_string()],
396        )
397        .map_err(map_db_error)?;
398        Ok(())
399    }
400
401    fn add_address(&self, input: CreateCustomerAddress) -> Result<CustomerAddress> {
402        Self::validate_address_input(&input)?;
403
404        let mut conn = self.conn()?;
405        let id = Uuid::new_v4();
406        let now = Utc::now();
407        let address_type = input.address_type.unwrap_or_default();
408        let is_default = input.is_default.unwrap_or(false);
409
410        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
411
412        tx.execute(
413            "INSERT INTO customer_addresses (id, customer_id, address_type, first_name, last_name,
414                                             company, line1, line2, city, state, postal_code,
415                                             country, phone, is_default, created_at, updated_at)
416             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
417            rusqlite::params![
418                id.to_string(),
419                input.customer_id.to_string(),
420                address_type.to_string(),
421                input.first_name,
422                input.last_name,
423                input.company,
424                input.line1,
425                input.line2,
426                input.city,
427                input.state,
428                input.postal_code,
429                input.country,
430                input.phone,
431                i32::from(is_default),
432                now.to_rfc3339(),
433                now.to_rfc3339(),
434            ],
435        )
436        .map_err(map_db_error)?;
437
438        if is_default {
439            let clear_sql = match address_type {
440                AddressType::Shipping => {
441                    "UPDATE customer_addresses SET is_default = 0 WHERE customer_id = ? AND (address_type = 'shipping' OR address_type = 'both')"
442                }
443                AddressType::Billing => {
444                    "UPDATE customer_addresses SET is_default = 0 WHERE customer_id = ? AND (address_type = 'billing' OR address_type = 'both')"
445                }
446                AddressType::Both => {
447                    "UPDATE customer_addresses SET is_default = 0 WHERE customer_id = ?"
448                }
449                _ => "UPDATE customer_addresses SET is_default = 0 WHERE customer_id = ?",
450            };
451
452            // Clear defaults for the relevant address types.
453            tx.execute(clear_sql, [input.customer_id.to_string()]).map_err(map_db_error)?;
454
455            // Set new default
456            tx.execute(
457                "UPDATE customer_addresses SET is_default = 1 WHERE id = ?",
458                [id.to_string()],
459            )
460            .map_err(map_db_error)?;
461
462            match address_type {
463                AddressType::Shipping => {
464                    tx.execute(
465                        "UPDATE customers SET default_shipping_address_id = ?, updated_at = ? WHERE id = ?",
466                        rusqlite::params![id.to_string(), now.to_rfc3339(), input.customer_id.to_string()],
467                    )
468                    .map_err(map_db_error)?;
469                }
470                AddressType::Billing => {
471                    tx.execute(
472                        "UPDATE customers SET default_billing_address_id = ?, updated_at = ? WHERE id = ?",
473                        rusqlite::params![id.to_string(), now.to_rfc3339(), input.customer_id.to_string()],
474                    )
475                    .map_err(map_db_error)?;
476                }
477                AddressType::Both | _ => {
478                    tx.execute(
479                        "UPDATE customers SET default_shipping_address_id = ?, default_billing_address_id = ?, updated_at = ? WHERE id = ?",
480                        rusqlite::params![id.to_string(), id.to_string(), now.to_rfc3339(), input.customer_id.to_string()],
481                    )
482                    .map_err(map_db_error)?;
483                }
484            }
485        }
486
487        let addr = tx
488            .query_row(
489                "SELECT * FROM customer_addresses WHERE id = ?",
490                [id.to_string()],
491                Self::row_to_address,
492            )
493            .map_err(map_db_error)?;
494
495        tx.commit().map_err(map_db_error)?;
496        Ok(addr)
497    }
498
499    fn get_addresses(&self, customer_id: CustomerId) -> Result<Vec<CustomerAddress>> {
500        let conn = self.conn()?;
501        let mut stmt = conn
502            .prepare("SELECT * FROM customer_addresses WHERE customer_id = ?")
503            .map_err(map_db_error)?;
504
505        let addresses = stmt
506            .query_map([customer_id.to_string()], Self::row_to_address)
507            .map_err(map_db_error)?
508            .collect::<rusqlite::Result<Vec<_>>>()
509            .map_err(map_db_error)?;
510
511        Ok(addresses)
512    }
513
514    fn update_address(
515        &self,
516        address_id: Uuid,
517        input: CreateCustomerAddress,
518    ) -> Result<CustomerAddress> {
519        Self::validate_address_input(&input)?;
520
521        let conn = self.conn()?;
522        let now = Utc::now();
523
524        let exists: i32 = conn
525            .query_row(
526                "SELECT COUNT(*) FROM customer_addresses WHERE id = ? AND customer_id = ?",
527                rusqlite::params![address_id.to_string(), input.customer_id.to_string()],
528                |row| row.get(0),
529            )
530            .map_err(map_db_error)?;
531        if exists == 0 {
532            return Err(CommerceError::ValidationError(
533                "Address does not belong to customer".into(),
534            ));
535        }
536
537        conn.execute(
538            "UPDATE customer_addresses SET first_name = ?, last_name = ?, company = ?,
539                     line1 = ?, line2 = ?, city = ?, state = ?, postal_code = ?,
540                     country = ?, phone = ?, updated_at = ? WHERE id = ?",
541            rusqlite::params![
542                input.first_name,
543                input.last_name,
544                input.company,
545                input.line1,
546                input.line2,
547                input.city,
548                input.state,
549                input.postal_code,
550                input.country,
551                input.phone,
552                now.to_rfc3339(),
553                address_id.to_string(),
554            ],
555        )
556        .map_err(map_db_error)?;
557
558        let addr = conn
559            .query_row(
560                "SELECT * FROM customer_addresses WHERE id = ?",
561                [address_id.to_string()],
562                Self::row_to_address,
563            )
564            .map_err(map_db_error)?;
565
566        Ok(addr)
567    }
568
569    fn delete_address(&self, address_id: Uuid) -> Result<()> {
570        let mut conn = self.conn()?;
571        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
572        let now = Utc::now();
573
574        let row: Option<(String, String, i32)> = tx
575            .query_row(
576                "SELECT customer_id, address_type, is_default FROM customer_addresses WHERE id = ?",
577                [address_id.to_string()],
578                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
579            )
580            .optional()
581            .map_err(map_db_error)?;
582
583        let (customer_id, address_type, is_default) = match row {
584            Some(values) => values,
585            None => return Err(CommerceError::NotFound),
586        };
587
588        if is_default != 0 {
589            let parsed_type: AddressType =
590                parse_enum(&address_type, "customer_address", "address_type")?;
591            match parsed_type {
592                AddressType::Shipping => {
593                    tx.execute(
594                        "UPDATE customers SET default_shipping_address_id = NULL, updated_at = ? WHERE id = ?",
595                        rusqlite::params![now.to_rfc3339(), customer_id],
596                    )
597                    .map_err(map_db_error)?;
598                }
599                AddressType::Billing => {
600                    tx.execute(
601                        "UPDATE customers SET default_billing_address_id = NULL, updated_at = ? WHERE id = ?",
602                        rusqlite::params![now.to_rfc3339(), customer_id],
603                    )
604                    .map_err(map_db_error)?;
605                }
606                AddressType::Both | _ => {
607                    tx.execute(
608                        "UPDATE customers SET default_shipping_address_id = NULL, default_billing_address_id = NULL, updated_at = ? WHERE id = ?",
609                        rusqlite::params![now.to_rfc3339(), customer_id],
610                    )
611                    .map_err(map_db_error)?;
612                }
613            }
614        }
615
616        tx.execute("DELETE FROM customer_addresses WHERE id = ?", [address_id.to_string()])
617            .map_err(map_db_error)?;
618        tx.commit().map_err(map_db_error)?;
619        Ok(())
620    }
621
622    fn set_default_address(
623        &self,
624        customer_id: CustomerId,
625        address_id: Uuid,
626        address_type: AddressType,
627    ) -> Result<()> {
628        let mut conn = self.conn()?;
629        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
630        let now = Utc::now();
631
632        let owner: Option<String> = tx
633            .query_row(
634                "SELECT customer_id FROM customer_addresses WHERE id = ?",
635                [address_id.to_string()],
636                |row| row.get(0),
637            )
638            .optional()
639            .map_err(map_db_error)?;
640        match owner {
641            Some(owner_id) if owner_id != customer_id.to_string() => {
642                return Err(CommerceError::ValidationError(
643                    "Address does not belong to customer".into(),
644                ));
645            }
646            None => {
647                return Err(CommerceError::NotFound);
648            }
649            _ => {}
650        }
651
652        // Clear other defaults
653        let clear_sql = match address_type {
654            AddressType::Shipping => {
655                "UPDATE customer_addresses SET is_default = 0 WHERE customer_id = ? AND (address_type = 'shipping' OR address_type = 'both')"
656            }
657            AddressType::Billing => {
658                "UPDATE customer_addresses SET is_default = 0 WHERE customer_id = ? AND (address_type = 'billing' OR address_type = 'both')"
659            }
660            AddressType::Both | _ => {
661                "UPDATE customer_addresses SET is_default = 0 WHERE customer_id = ?"
662            }
663        };
664        tx.execute(clear_sql, [customer_id.to_string()]).map_err(map_db_error)?;
665
666        // Set new default
667        tx.execute(
668            "UPDATE customer_addresses SET is_default = 1 WHERE id = ?",
669            [address_id.to_string()],
670        )
671        .map_err(map_db_error)?;
672
673        // Update customer
674        match address_type {
675            AddressType::Shipping => {
676                tx.execute(
677                    "UPDATE customers SET default_shipping_address_id = ?, updated_at = ? WHERE id = ?",
678                    rusqlite::params![address_id.to_string(), now.to_rfc3339(), customer_id.to_string()],
679                )
680                .map_err(map_db_error)?;
681            }
682            AddressType::Billing => {
683                tx.execute(
684                    "UPDATE customers SET default_billing_address_id = ?, updated_at = ? WHERE id = ?",
685                    rusqlite::params![address_id.to_string(), now.to_rfc3339(), customer_id.to_string()],
686                )
687                .map_err(map_db_error)?;
688            }
689            AddressType::Both | _ => {
690                tx.execute(
691                    "UPDATE customers SET default_shipping_address_id = ?, default_billing_address_id = ?, updated_at = ? WHERE id = ?",
692                    rusqlite::params![address_id.to_string(), address_id.to_string(), now.to_rfc3339(), customer_id.to_string()],
693                )
694                .map_err(map_db_error)?;
695            }
696        }
697
698        tx.commit().map_err(map_db_error)?;
699
700        Ok(())
701    }
702
703    fn count(&self, filter: CustomerFilter) -> Result<u64> {
704        let conn = self.conn()?;
705        let use_json = json1_available(&conn);
706        let mut sql = "SELECT COUNT(*) FROM customers WHERE 1=1".to_string();
707        let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![];
708
709        if let Some(email) = &filter.email {
710            sql.push_str(" AND email LIKE ?");
711            params.push(Box::new(format!("%{email}%")));
712        }
713        if let Some(status) = &filter.status {
714            sql.push_str(" AND status = ?");
715            params.push(Box::new(status.to_string()));
716        } else {
717            sql.push_str(" AND status != 'deleted'");
718        }
719        if let Some(tag) = &filter.tag {
720            if use_json {
721                sql.push_str(" AND EXISTS (SELECT 1 FROM json_each(tags) WHERE value = ?)");
722                params.push(Box::new(tag.clone()));
723            } else {
724                sql.push_str(" AND tags LIKE ?");
725                params.push(Box::new(format!("%\"{tag}\"%")));
726            }
727        }
728        if let Some(accepts_marketing) = &filter.accepts_marketing {
729            sql.push_str(" AND accepts_marketing = ?");
730            params.push(Box::new(i32::from(*accepts_marketing)));
731        }
732
733        let params_refs: Vec<&dyn rusqlite::ToSql> =
734            params.iter().map(std::convert::AsRef::as_ref).collect();
735        let count: i64 =
736            conn.query_row(&sql, params_refs.as_slice(), |row| row.get(0)).map_err(map_db_error)?;
737
738        Ok(count as u64)
739    }
740
741    // === Batch Operations ===
742
743    fn create_batch(&self, inputs: Vec<CreateCustomer>) -> Result<BatchResult<Customer>> {
744        validate_batch_size(&inputs)?;
745        let mut result = BatchResult::with_capacity(inputs.len());
746
747        for (index, input) in inputs.into_iter().enumerate() {
748            match self.create(input) {
749                Ok(customer) => result.record_success(customer),
750                Err(e) => result.record_failure(index, None, &e),
751            }
752        }
753
754        Ok(result)
755    }
756
757    fn create_batch_atomic(&self, inputs: Vec<CreateCustomer>) -> Result<Vec<Customer>> {
758        validate_batch_size(&inputs)?;
759        if inputs.is_empty() {
760            return Ok(vec![]);
761        }
762
763        let mut conn = self.conn()?;
764        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
765        let mut results = Vec::with_capacity(inputs.len());
766
767        for input in inputs {
768            let id = CustomerId::new();
769            let now = Utc::now();
770            let tags = input.tags.clone().unwrap_or_default();
771            let metadata = input.metadata.clone();
772            let email = input.email.clone();
773            let first_name = input.first_name.clone();
774            let last_name = input.last_name.clone();
775            let phone = input.phone.clone();
776            let accepts_marketing = input.accepts_marketing.unwrap_or(false);
777
778            // Check email uniqueness
779            let exists: i32 = tx
780                .query_row(
781                    "SELECT COUNT(*) FROM customers WHERE email = ?",
782                    [&input.email],
783                    |row| row.get(0),
784                )
785                .map_err(map_db_error)?;
786
787            if exists > 0 {
788                return Err(CommerceError::EmailAlreadyExists(input.email));
789            }
790
791            let tags_json = serde_json::to_string(&tags).unwrap_or_default();
792            let metadata_json =
793                metadata.as_ref().map(|m| serde_json::to_string(m).unwrap_or_default());
794
795            tx.execute(
796                "INSERT INTO customers (id, email, first_name, last_name, phone, status,
797                                        accepts_marketing, email_verified, tags, metadata,
798                                        created_at, updated_at)
799                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
800                rusqlite::params![
801                    id.to_string(),
802                    &email,
803                    &first_name,
804                    &last_name,
805                    &phone,
806                    "active",
807                    i32::from(accepts_marketing),
808                    0,
809                    tags_json,
810                    metadata_json,
811                    now.to_rfc3339(),
812                    now.to_rfc3339(),
813                ],
814            )
815            .map_err(map_db_error)?;
816
817            results.push(Customer {
818                id,
819                email,
820                first_name,
821                last_name,
822                phone,
823                status: CustomerStatus::Active,
824                accepts_marketing,
825                email_verified: false,
826                tags,
827                metadata,
828                default_shipping_address_id: None,
829                default_billing_address_id: None,
830                created_at: now,
831                updated_at: now,
832            });
833        }
834
835        tx.commit().map_err(map_db_error)?;
836        Ok(results)
837    }
838
839    fn update_batch(
840        &self,
841        updates: Vec<(CustomerId, UpdateCustomer)>,
842    ) -> Result<BatchResult<Customer>> {
843        validate_batch_size(&updates)?;
844        let mut result = BatchResult::with_capacity(updates.len());
845
846        for (index, (id, input)) in updates.into_iter().enumerate() {
847            match self.update(id, input) {
848                Ok(customer) => result.record_success(customer),
849                Err(e) => result.record_failure(index, Some(id.to_string()), &e),
850            }
851        }
852
853        Ok(result)
854    }
855
856    fn update_batch_atomic(
857        &self,
858        updates: Vec<(CustomerId, UpdateCustomer)>,
859    ) -> Result<Vec<Customer>> {
860        validate_batch_size(&updates)?;
861        if updates.is_empty() {
862            return Ok(vec![]);
863        }
864
865        let mut conn = self.conn()?;
866        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
867        let mut results = Vec::with_capacity(updates.len());
868
869        for (id, input) in updates {
870            let now = Utc::now();
871            let current_version: i32 = tx
872                .query_row("SELECT version FROM customers WHERE id = ?", [id.to_string()], |row| {
873                    row.get(0)
874                })
875                .map_err(|e| match e {
876                    rusqlite::Error::QueryReturnedNoRows => {
877                        CommerceError::CustomerNotFound(id.into_uuid())
878                    }
879                    e => map_db_error(e),
880                })?;
881
882            let mut update_parts = vec!["updated_at = ?"];
883            let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(now.to_rfc3339())];
884
885            if let Some(email) = &input.email {
886                validate_email(email)?;
887                let existing_id: Option<String> = tx
888                    .query_row("SELECT id FROM customers WHERE email = ?", [email], |row| {
889                        row.get(0)
890                    })
891                    .optional()
892                    .map_err(map_db_error)?;
893                if let Some(existing_id) = existing_id {
894                    if existing_id != id.to_string() {
895                        return Err(CommerceError::EmailAlreadyExists(email.clone()));
896                    }
897                }
898                update_parts.push("email = ?");
899                params.push(Box::new(email.clone()));
900            }
901            if let Some(first_name) = &input.first_name {
902                validate_required_text("customer.first_name", first_name, 100)?;
903                update_parts.push("first_name = ?");
904                params.push(Box::new(first_name.clone()));
905            }
906            if let Some(last_name) = &input.last_name {
907                validate_required_text("customer.last_name", last_name, 100)?;
908                update_parts.push("last_name = ?");
909                params.push(Box::new(last_name.clone()));
910            }
911            if let Some(phone) = &input.phone {
912                validate_phone(phone)?;
913                update_parts.push("phone = ?");
914                params.push(Box::new(phone.clone()));
915            }
916            if let Some(status) = &input.status {
917                update_parts.push("status = ?");
918                params.push(Box::new(status.to_string()));
919            }
920            if let Some(accepts_marketing) = &input.accepts_marketing {
921                update_parts.push("accepts_marketing = ?");
922                params.push(Box::new(i32::from(*accepts_marketing)));
923            }
924            if let Some(tags) = &input.tags {
925                update_parts.push("tags = ?");
926                params.push(Box::new(serde_json::to_string(tags).unwrap_or_default()));
927            }
928            if let Some(metadata) = &input.metadata {
929                update_parts.push("metadata = ?");
930                params.push(Box::new(serde_json::to_string(metadata).unwrap_or_default()));
931            }
932
933            update_parts.push("version = version + 1");
934            params.push(Box::new(id.to_string()));
935            params.push(Box::new(current_version));
936
937            let sql = format!(
938                "UPDATE customers SET {} WHERE id = ? AND version = ?",
939                update_parts.join(", ")
940            );
941
942            let params_refs: Vec<&dyn rusqlite::ToSql> =
943                params.iter().map(std::convert::AsRef::as_ref).collect();
944            let rows_affected = tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
945            if rows_affected == 0 {
946                return Err(CommerceError::VersionConflict {
947                    entity: "customer".to_string(),
948                    id: id.to_string(),
949                    expected_version: current_version,
950                });
951            }
952
953            let customer = tx
954                .query_row(
955                    "SELECT * FROM customers WHERE id = ?",
956                    [id.to_string()],
957                    Self::row_to_customer,
958                )
959                .map_err(map_db_error)?;
960
961            results.push(customer);
962        }
963
964        tx.commit().map_err(map_db_error)?;
965        Ok(results)
966    }
967
968    fn delete_batch(&self, ids: Vec<CustomerId>) -> Result<BatchResult<CustomerId>> {
969        validate_batch_size(&ids)?;
970        let mut result = BatchResult::with_capacity(ids.len());
971
972        for (index, id) in ids.into_iter().enumerate() {
973            match self.delete(id) {
974                Ok(()) => result.record_success(id),
975                Err(e) => result.record_failure(index, Some(id.to_string()), &e),
976            }
977        }
978
979        Ok(result)
980    }
981
982    fn delete_batch_atomic(&self, ids: Vec<CustomerId>) -> Result<()> {
983        validate_batch_size(&ids)?;
984        if ids.is_empty() {
985            return Ok(());
986        }
987
988        let mut conn = self.conn()?;
989        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
990
991        let placeholders = build_in_clause(ids.len());
992
993        // Soft delete by setting status to 'deleted'
994        let now = Utc::now();
995        let sql = format!(
996            "UPDATE customers SET status = 'deleted', updated_at = ? WHERE id IN ({placeholders})"
997        );
998
999        let mut all_params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(now.to_rfc3339())];
1000        for id in &ids {
1001            all_params.push(Box::new(id.to_string()));
1002        }
1003        let all_params_refs: Vec<&dyn rusqlite::ToSql> =
1004            all_params.iter().map(std::convert::AsRef::as_ref).collect();
1005
1006        tx.execute(&sql, all_params_refs.as_slice()).map_err(map_db_error)?;
1007
1008        tx.commit().map_err(map_db_error)?;
1009        Ok(())
1010    }
1011
1012    fn get_batch(&self, ids: Vec<CustomerId>) -> Result<Vec<Customer>> {
1013        validate_batch_size(&ids)?;
1014        if ids.is_empty() {
1015            return Ok(vec![]);
1016        }
1017
1018        let conn = self.conn()?;
1019        let placeholders = build_in_clause(ids.len());
1020        let sql = format!("SELECT * FROM customers WHERE id IN ({placeholders})");
1021
1022        let raw_ids: Vec<Uuid> = ids.iter().map(|id| id.into_uuid()).collect();
1023        let params = uuid_params(&raw_ids);
1024        let params_refs = params_refs(&params);
1025
1026        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
1027        let customers = stmt
1028            .query_map(params_refs.as_slice(), Self::row_to_customer)
1029            .map_err(map_db_error)?
1030            .collect::<rusqlite::Result<Vec<_>>>()
1031            .map_err(map_db_error)?;
1032
1033        Ok(customers)
1034    }
1035}
stateset_db/sqlite/customers.rs

stateset_db/sqlite/
customers.rs
