stateset_db/sqlite/
products.rs

1//! SQLite product repository implementation
2
3use super::{
4    build_in_clause, escape_like, json1_available, map_db_error, params_refs, parse_datetime_row,
5    parse_decimal_opt_row, parse_decimal_row, parse_enum_row, parse_json_opt_row, parse_json_row,
6    parse_uuid, parse_uuid_row, uuid_params,
7};
8use chrono::Utc;
9use r2d2::Pool;
10use r2d2_sqlite::SqliteConnectionManager;
11use rusqlite::OptionalExtension;
12use stateset_core::{
13    BatchResult, CommerceError, CreateProduct, CreateProductVariant, Product, ProductFilter,
14    ProductId, ProductRepository, ProductStatus, ProductVariant, Result, UpdateProduct,
15    validate_batch_size, validate_sku,
16};
17use uuid::Uuid;
18
19/// SQLite implementation of `ProductRepository`
20#[derive(Debug)]
21pub struct SqliteProductRepository {
22    pool: Pool<SqliteConnectionManager>,
23}
24
25impl SqliteProductRepository {
26    #[must_use]
27    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
28        Self { pool }
29    }
30
31    fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
32        self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
33    }
34
35    fn row_to_product(row: &rusqlite::Row<'_>) -> rusqlite::Result<Product> {
36        let attributes_json: String = row.get("attributes")?;
37        let seo_json: Option<String> = row.get("seo")?;
38
39        Ok(Product {
40            id: ProductId::from(parse_uuid_row(&row.get::<_, String>("id")?, "product", "id")?),
41            name: row.get("name")?,
42            slug: row.get("slug")?,
43            description: row.get("description")?,
44            status: parse_enum_row(&row.get::<_, String>("status")?, "product", "status")?,
45            product_type: parse_enum_row(
46                &row.get::<_, String>("product_type")?,
47                "product",
48                "product_type",
49            )?,
50            attributes: parse_json_row(&attributes_json, "product", "attributes")?,
51            seo: parse_json_opt_row(seo_json, "product", "seo")?,
52            created_at: parse_datetime_row(
53                &row.get::<_, String>("created_at")?,
54                "product",
55                "created_at",
56            )?,
57            updated_at: parse_datetime_row(
58                &row.get::<_, String>("updated_at")?,
59                "product",
60                "updated_at",
61            )?,
62        })
63    }
64
65    fn row_to_variant(row: &rusqlite::Row<'_>) -> rusqlite::Result<ProductVariant> {
66        let options_json: String = row.get("options")?;
67
68        Ok(ProductVariant {
69            id: parse_uuid_row(&row.get::<_, String>("id")?, "product_variant", "id")?,
70            product_id: ProductId::from(parse_uuid_row(
71                &row.get::<_, String>("product_id")?,
72                "product_variant",
73                "product_id",
74            )?),
75            sku: row.get("sku")?,
76            name: row.get("name")?,
77            price: parse_decimal_row(&row.get::<_, String>("price")?, "product_variant", "price")?,
78            compare_at_price: parse_decimal_opt_row(
79                row.get::<_, Option<String>>("compare_at_price")?,
80                "product_variant",
81                "compare_at_price",
82            )?,
83            cost: parse_decimal_opt_row(
84                row.get::<_, Option<String>>("cost")?,
85                "product_variant",
86                "cost",
87            )?,
88            barcode: row.get("barcode")?,
89            weight: parse_decimal_opt_row(
90                row.get::<_, Option<String>>("weight")?,
91                "product_variant",
92                "weight",
93            )?,
94            weight_unit: row.get("weight_unit")?,
95            options: parse_json_row(&options_json, "product_variant", "options")?,
96            is_default: row.get::<_, i32>("is_default")? != 0,
97            is_active: row.get::<_, i32>("is_active")? != 0,
98            created_at: parse_datetime_row(
99                &row.get::<_, String>("created_at")?,
100                "product_variant",
101                "created_at",
102            )?,
103            updated_at: parse_datetime_row(
104                &row.get::<_, String>("updated_at")?,
105                "product_variant",
106                "updated_at",
107            )?,
108        })
109    }
110}
111
112impl ProductRepository for SqliteProductRepository {
113    fn create(&self, input: CreateProduct) -> Result<Product> {
114        let mut conn = self.conn()?;
115        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
116        let id = ProductId::new();
117        let now = Utc::now();
118        let slug = input.slug.clone().unwrap_or_else(|| Product::generate_slug(&input.name));
119        let name = input.name.clone();
120        let description = input.description.clone().unwrap_or_default();
121        let product_type = input.product_type.unwrap_or_default();
122        let attributes = input.attributes.clone().unwrap_or_default();
123        let seo = input.seo.clone();
124
125        // Check slug uniqueness
126        let exists: i32 = tx
127            .query_row("SELECT COUNT(*) FROM products WHERE slug = ?", [&slug], |row| row.get(0))
128            .map_err(map_db_error)?;
129
130        if exists > 0 {
131            return Err(CommerceError::DuplicateSlug(slug));
132        }
133
134        let attributes_json = serde_json::to_string(&attributes).unwrap_or_default();
135        let seo_json = seo.as_ref().map(|s| serde_json::to_string(s).unwrap_or_default());
136
137        tx.execute(
138            "INSERT INTO products (id, name, slug, description, status, product_type, attributes, seo, created_at, updated_at)
139             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
140            rusqlite::params![
141                id.to_string(),
142                &name,
143                &slug,
144                &description,
145                ProductStatus::Draft.to_string(),
146                product_type.to_string(),
147                attributes_json,
148                seo_json,
149                now.to_rfc3339(),
150                now.to_rfc3339(),
151            ],
152        )
153        .map_err(map_db_error)?;
154
155        // Create variants inline if provided (using the same connection)
156        if let Some(variants) = &input.variants {
157            for (i, variant) in variants.iter().enumerate() {
158                let variant_id = Uuid::new_v4();
159
160                // Check SKU uniqueness
161                let sku_exists: i32 = tx
162                    .query_row(
163                        "SELECT COUNT(*) FROM product_variants WHERE sku = ?",
164                        [&variant.sku],
165                        |row| row.get(0),
166                    )
167                    .map_err(map_db_error)?;
168
169                if sku_exists > 0 {
170                    return Err(CommerceError::DuplicateSku(variant.sku.clone()));
171                }
172
173                let options_json =
174                    serde_json::to_string(&variant.options.clone().unwrap_or_default())
175                        .unwrap_or_default();
176
177                tx.execute(
178                    "INSERT INTO product_variants (id, product_id, sku, name, price, compare_at_price, cost,
179                                                   barcode, weight, weight_unit, options, is_default, is_active,
180                                                   created_at, updated_at)
181                     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)",
182                    rusqlite::params![
183                        variant_id.to_string(),
184                        id.to_string(),
185                        &variant.sku,
186                        variant.name.as_ref().unwrap_or(&variant.sku),
187                        variant.price.to_string(),
188                        variant.compare_at_price.map(|d| d.to_string()),
189                        variant.cost.map(|d| d.to_string()),
190                        &variant.barcode,
191                        variant.weight.map(|d| d.to_string()),
192                        &variant.weight_unit,
193                        options_json,
194                        i32::from(i == 0),  // First variant is default
195                        now.to_rfc3339(),
196                        now.to_rfc3339(),
197                    ],
198                )
199                .map_err(map_db_error)?;
200            }
201        }
202
203        tx.commit().map_err(map_db_error)?;
204
205        Ok(Product {
206            id,
207            name,
208            slug,
209            description,
210            status: ProductStatus::Draft,
211            product_type,
212            attributes,
213            seo,
214            created_at: now,
215            updated_at: now,
216        })
217    }
218
219    fn get(&self, id: ProductId) -> Result<Option<Product>> {
220        let conn = self.conn()?;
221        let result = conn.query_row(
222            "SELECT * FROM products WHERE id = ?",
223            [id.to_string()],
224            Self::row_to_product,
225        );
226
227        match result {
228            Ok(product) => Ok(Some(product)),
229            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
230            Err(e) => Err(map_db_error(e)),
231        }
232    }
233
234    fn get_by_slug(&self, slug: &str) -> Result<Option<Product>> {
235        let conn = self.conn()?;
236        let result =
237            conn.query_row("SELECT * FROM products WHERE slug = ?", [slug], Self::row_to_product);
238
239        match result {
240            Ok(product) => Ok(Some(product)),
241            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
242            Err(e) => Err(map_db_error(e)),
243        }
244    }
245
246    fn update(&self, id: ProductId, input: UpdateProduct) -> Result<Product> {
247        let conn = self.conn()?;
248        let now = Utc::now();
249        let current_version: i32 = conn
250            .query_row("SELECT version FROM products WHERE id = ?", [id.to_string()], |row| {
251                row.get(0)
252            })
253            .map_err(|e| match e {
254                rusqlite::Error::QueryReturnedNoRows => {
255                    CommerceError::ProductNotFound(id.into_uuid())
256                }
257                e => map_db_error(e),
258            })?;
259
260        let mut updates = vec!["updated_at = ?"];
261        let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(now.to_rfc3339())];
262
263        if let Some(name) = &input.name {
264            updates.push("name = ?");
265            params.push(Box::new(name.clone()));
266        }
267        if let Some(slug) = &input.slug {
268            let existing_id: Option<String> = conn
269                .query_row("SELECT id FROM products WHERE slug = ?", [slug], |row| row.get(0))
270                .optional()
271                .map_err(map_db_error)?;
272            if let Some(existing_id) = existing_id {
273                if existing_id != id.to_string() {
274                    return Err(CommerceError::DuplicateSlug(slug.clone()));
275                }
276            }
277            updates.push("slug = ?");
278            params.push(Box::new(slug.clone()));
279        }
280        if let Some(description) = &input.description {
281            updates.push("description = ?");
282            params.push(Box::new(description.clone()));
283        }
284        if let Some(status) = &input.status {
285            updates.push("status = ?");
286            params.push(Box::new(status.to_string()));
287        }
288        if let Some(attributes) = &input.attributes {
289            updates.push("attributes = ?");
290            params.push(Box::new(serde_json::to_string(attributes).unwrap_or_default()));
291        }
292        if let Some(seo) = &input.seo {
293            updates.push("seo = ?");
294            params.push(Box::new(serde_json::to_string(seo).unwrap_or_default()));
295        }
296
297        updates.push("version = version + 1");
298        params.push(Box::new(id.to_string()));
299        params.push(Box::new(current_version));
300
301        let sql =
302            format!("UPDATE products SET {} WHERE id = ? AND version = ?", updates.join(", "));
303        let params_refs: Vec<&dyn rusqlite::ToSql> =
304            params.iter().map(std::convert::AsRef::as_ref).collect();
305
306        let rows_affected = conn.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
307        if rows_affected == 0 {
308            return Err(CommerceError::VersionConflict {
309                entity: "product".to_string(),
310                id: id.to_string(),
311                expected_version: current_version,
312            });
313        }
314
315        // Fetch the updated product with the same connection
316        let result = conn.query_row(
317            "SELECT * FROM products WHERE id = ?",
318            [id.to_string()],
319            Self::row_to_product,
320        );
321
322        match result {
323            Ok(product) => Ok(product),
324            Err(rusqlite::Error::QueryReturnedNoRows) => {
325                Err(CommerceError::ProductNotFound(id.into_uuid()))
326            }
327            Err(e) => Err(map_db_error(e)),
328        }
329    }
330
331    fn list(&self, filter: ProductFilter) -> Result<Vec<Product>> {
332        let ProductFilter {
333            status,
334            product_type,
335            search,
336            category,
337            min_price,
338            max_price,
339            in_stock,
340            limit,
341            offset,
342            after_cursor,
343        } = filter;
344        let conn = self.conn()?;
345        let use_json = json1_available(&conn);
346        let mut sql = "SELECT * FROM products WHERE 1=1".to_string();
347        let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![];
348
349        if let Some(status) = status {
350            sql.push_str(" AND status = ?");
351            params.push(Box::new(status.to_string()));
352        } else {
353            sql.push_str(" AND status != 'archived'");
354        }
355        if let Some(product_type) = product_type {
356            sql.push_str(" AND product_type = ?");
357            params.push(Box::new(product_type.to_string()));
358        }
359        if let Some(search) = search {
360            sql.push_str(" AND (name LIKE ? ESCAPE '\\' OR description LIKE ? ESCAPE '\\')");
361            let escaped = format!("%{}%", escape_like(&search));
362            params.push(Box::new(escaped.clone()));
363            params.push(Box::new(escaped));
364        }
365        if let Some(category) = category {
366            if use_json {
367                sql.push_str(
368                    " AND EXISTS (SELECT 1 FROM json_each(attributes) attr \
369                     WHERE (json_extract(attr.value, '$.name') = 'category' \
370                        OR json_extract(attr.value, '$.group') = 'category') \
371                       AND json_extract(attr.value, '$.value') = ?)",
372                );
373                params.push(Box::new(category));
374            } else {
375                sql.push_str(" AND (attributes LIKE ? OR attributes LIKE ?)");
376                params.push(Box::new(format!("%\"name\":\"category\",\"value\":\"{category}\"%")));
377                params.push(Box::new(format!("%\"value\":\"{category}\",\"group\":\"category\"%")));
378            }
379        }
380        if let Some(in_stock) = in_stock {
381            let stock_clause = if in_stock { "EXISTS" } else { "NOT EXISTS" };
382            sql.push_str(&format!(
383                " AND {stock_clause} (SELECT 1 FROM product_variants pv_stock \
384                 JOIN inventory_items ii ON ii.sku = pv_stock.sku \
385                 JOIN inventory_balances ib ON ib.item_id = ii.id \
386                 WHERE pv_stock.product_id = products.id \
387                   AND pv_stock.is_active = 1 \
388                   AND CAST(ib.quantity_available AS REAL) > 0)"
389            ));
390        }
391
392        // Keyset cursor: (name, id) for stable ASC ordering
393        if let Some((cursor_name, cursor_id)) = &after_cursor {
394            sql.push_str(" AND (name > ? OR (name = ? AND id > ?))");
395            params.push(Box::new(cursor_name.clone()));
396            params.push(Box::new(cursor_name.clone()));
397            params.push(Box::new(cursor_id.clone()));
398        }
399
400        sql.push_str(" ORDER BY name ASC, id ASC");
401
402        let apply_price_filter = min_price.is_some() || max_price.is_some();
403        if !apply_price_filter {
404            // Offset pagination applies only in non-cursor mode; the helper emits
405            // `LIMIT -1 OFFSET n` when an offset is set without a limit (SQLite
406            // rejects a bare OFFSET).
407            let page_offset = if after_cursor.is_none() { offset } else { None };
408            crate::sqlite::append_limit_offset(&mut sql, limit, page_offset);
409        }
410
411        let params_refs: Vec<&dyn rusqlite::ToSql> =
412            params.iter().map(std::convert::AsRef::as_ref).collect();
413        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
414
415        let products = stmt
416            .query_map(params_refs.as_slice(), Self::row_to_product)
417            .map_err(map_db_error)?
418            .collect::<rusqlite::Result<Vec<_>>>()
419            .map_err(map_db_error)?;
420
421        if apply_price_filter {
422            let min_price = min_price.as_ref();
423            let max_price = max_price.as_ref();
424            let mut filtered = Vec::with_capacity(products.len());
425            for product in products {
426                let variants = self.get_variants(product.id)?;
427                let mut matches = false;
428                for variant in variants {
429                    if !variant.is_active {
430                        continue;
431                    }
432                    if let Some(min) = min_price {
433                        if variant.price < *min {
434                            continue;
435                        }
436                    }
437                    if let Some(max) = max_price {
438                        if variant.price > *max {
439                            continue;
440                        }
441                    }
442                    matches = true;
443                    break;
444                }
445                if matches {
446                    filtered.push(product);
447                }
448            }
449            if let Some(offset) = offset {
450                filtered = filtered.into_iter().skip(offset as usize).collect();
451            }
452            if let Some(limit) = limit {
453                filtered.truncate(limit as usize);
454            }
455            return Ok(filtered);
456        }
457
458        Ok(products)
459    }
460
461    fn delete(&self, id: ProductId) -> Result<()> {
462        let conn = self.conn()?;
463        conn.execute(
464            "UPDATE products SET status = 'archived', updated_at = ? WHERE id = ?",
465            rusqlite::params![Utc::now().to_rfc3339(), id.to_string()],
466        )
467        .map_err(map_db_error)?;
468        Ok(())
469    }
470
471    fn add_variant(
472        &self,
473        product_id: ProductId,
474        variant: CreateProductVariant,
475    ) -> Result<ProductVariant> {
476        // Validate SKU format
477        validate_sku(&variant.sku)?;
478
479        let conn = self.conn()?;
480        let id = Uuid::new_v4();
481        let now = Utc::now();
482        let sku = variant.sku.clone();
483        let name = variant.name.clone().unwrap_or_else(|| sku.clone());
484        let options = variant.options.clone().unwrap_or_default();
485
486        // Check SKU uniqueness
487        let exists: i32 = conn
488            .query_row("SELECT COUNT(*) FROM product_variants WHERE sku = ?", [&sku], |row| {
489                row.get(0)
490            })
491            .map_err(map_db_error)?;
492
493        if exists > 0 {
494            return Err(CommerceError::DuplicateSku(sku));
495        }
496
497        let options_json = serde_json::to_string(&options).unwrap_or_default();
498
499        conn.execute(
500            "INSERT INTO product_variants (id, product_id, sku, name, price, compare_at_price, cost,
501                                           barcode, weight, weight_unit, options, is_default, is_active,
502                                           created_at, updated_at)
503             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)",
504            rusqlite::params![
505                id.to_string(),
506                product_id.to_string(),
507                &sku,
508                &name,
509                variant.price.to_string(),
510                variant.compare_at_price.map(|d| d.to_string()),
511                variant.cost.map(|d| d.to_string()),
512                &variant.barcode,
513                variant.weight.map(|d| d.to_string()),
514                &variant.weight_unit,
515                options_json,
516                i32::from(variant.is_default.unwrap_or(false)),
517                now.to_rfc3339(),
518                now.to_rfc3339(),
519            ],
520        )
521        .map_err(map_db_error)?;
522
523        Ok(ProductVariant {
524            id,
525            product_id,
526            sku,
527            name,
528            price: variant.price,
529            compare_at_price: variant.compare_at_price,
530            cost: variant.cost,
531            barcode: variant.barcode,
532            weight: variant.weight,
533            weight_unit: variant.weight_unit,
534            options,
535            is_default: variant.is_default.unwrap_or(false),
536            is_active: true,
537            created_at: now,
538            updated_at: now,
539        })
540    }
541
542    fn get_variant(&self, id: Uuid) -> Result<Option<ProductVariant>> {
543        let conn = self.conn()?;
544        let result = conn.query_row(
545            "SELECT * FROM product_variants WHERE id = ?",
546            [id.to_string()],
547            Self::row_to_variant,
548        );
549
550        match result {
551            Ok(variant) => Ok(Some(variant)),
552            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
553            Err(e) => Err(map_db_error(e)),
554        }
555    }
556
557    fn get_variant_by_sku(&self, sku: &str) -> Result<Option<ProductVariant>> {
558        let conn = self.conn()?;
559        let result = conn.query_row(
560            "SELECT * FROM product_variants WHERE sku = ?",
561            [sku],
562            Self::row_to_variant,
563        );
564
565        match result {
566            Ok(variant) => Ok(Some(variant)),
567            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
568            Err(e) => Err(map_db_error(e)),
569        }
570    }
571
572    fn update_variant(&self, id: Uuid, variant: CreateProductVariant) -> Result<ProductVariant> {
573        let conn = self.conn()?;
574        let now = Utc::now();
575        let current_version: i32 = conn
576            .query_row(
577                "SELECT version FROM product_variants WHERE id = ?",
578                [id.to_string()],
579                |row| row.get(0),
580            )
581            .map_err(|e| match e {
582                rusqlite::Error::QueryReturnedNoRows => CommerceError::ProductVariantNotFound(id),
583                e => map_db_error(e),
584            })?;
585
586        let options_json =
587            serde_json::to_string(&variant.options.clone().unwrap_or_default()).unwrap_or_default();
588
589        let rows_affected = conn.execute(
590            "UPDATE product_variants SET name = ?, price = ?, compare_at_price = ?, cost = ?,
591                     barcode = ?, weight = ?, weight_unit = ?, options = ?, updated_at = ?, version = version + 1
592             WHERE id = ? AND version = ?",
593            rusqlite::params![
594                variant.name.as_ref().unwrap_or(&variant.sku),
595                variant.price.to_string(),
596                variant.compare_at_price.map(|d| d.to_string()),
597                variant.cost.map(|d| d.to_string()),
598                &variant.barcode,
599                variant.weight.map(|d| d.to_string()),
600                &variant.weight_unit,
601                options_json,
602                now.to_rfc3339(),
603                id.to_string(),
604                current_version,
605            ],
606        )
607        .map_err(map_db_error)?;
608        if rows_affected == 0 {
609            return Err(CommerceError::VersionConflict {
610                entity: "product_variant".to_string(),
611                id: id.to_string(),
612                expected_version: current_version,
613            });
614        }
615
616        // Fetch the updated variant with the same connection
617        let result = conn.query_row(
618            "SELECT * FROM product_variants WHERE id = ?",
619            [id.to_string()],
620            Self::row_to_variant,
621        );
622
623        match result {
624            Ok(v) => Ok(v),
625            Err(rusqlite::Error::QueryReturnedNoRows) => {
626                Err(CommerceError::ProductVariantNotFound(id))
627            }
628            Err(e) => Err(map_db_error(e)),
629        }
630    }
631
632    fn delete_variant(&self, id: Uuid) -> Result<()> {
633        let conn = self.conn()?;
634        conn.execute(
635            "UPDATE product_variants SET is_active = 0, updated_at = ? WHERE id = ?",
636            rusqlite::params![Utc::now().to_rfc3339(), id.to_string()],
637        )
638        .map_err(map_db_error)?;
639        Ok(())
640    }
641
642    fn get_variants(&self, product_id: ProductId) -> Result<Vec<ProductVariant>> {
643        let conn = self.conn()?;
644        let mut stmt = conn
645            .prepare("SELECT * FROM product_variants WHERE product_id = ? AND is_active = 1 ORDER BY is_default DESC, sku")
646            .map_err(map_db_error)?;
647
648        let variants = stmt
649            .query_map([product_id.to_string()], Self::row_to_variant)
650            .map_err(map_db_error)?
651            .collect::<rusqlite::Result<Vec<_>>>()
652            .map_err(map_db_error)?;
653
654        Ok(variants)
655    }
656
657    fn count(&self, filter: ProductFilter) -> Result<u64> {
658        let ProductFilter {
659            status,
660            product_type,
661            search,
662            category,
663            min_price,
664            max_price,
665            in_stock,
666            limit: _,
667            offset: _,
668            after_cursor: _,
669        } = filter;
670
671        let conn = self.conn()?;
672        let use_json = json1_available(&conn);
673        let mut sql = "SELECT id FROM products WHERE 1=1".to_string();
674        let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![];
675
676        if let Some(status) = status {
677            sql.push_str(" AND status = ?");
678            params.push(Box::new(status.to_string()));
679        } else {
680            sql.push_str(" AND status != 'archived'");
681        }
682        if let Some(product_type) = product_type {
683            sql.push_str(" AND product_type = ?");
684            params.push(Box::new(product_type.to_string()));
685        }
686        if let Some(search) = search {
687            sql.push_str(" AND (name LIKE ? ESCAPE '\\' OR description LIKE ? ESCAPE '\\')");
688            let escaped = format!("%{}%", escape_like(&search));
689            params.push(Box::new(escaped.clone()));
690            params.push(Box::new(escaped));
691        }
692        if let Some(category) = category {
693            if use_json {
694                sql.push_str(
695                    " AND EXISTS (SELECT 1 FROM json_each(attributes) attr \
696                     WHERE (json_extract(attr.value, '$.name') = 'category' \
697                        OR json_extract(attr.value, '$.group') = 'category') \
698                       AND json_extract(attr.value, '$.value') = ?)",
699                );
700                params.push(Box::new(category));
701            } else {
702                sql.push_str(" AND (attributes LIKE ? OR attributes LIKE ?)");
703                params.push(Box::new(format!("%\"name\":\"category\",\"value\":\"{category}\"%")));
704                params.push(Box::new(format!("%\"value\":\"{category}\",\"group\":\"category\"%")));
705            }
706        }
707        if let Some(in_stock) = in_stock {
708            let stock_clause = if in_stock { "EXISTS" } else { "NOT EXISTS" };
709            sql.push_str(&format!(
710                " AND {stock_clause} (SELECT 1 FROM product_variants pv_stock \
711                 JOIN inventory_items ii ON ii.sku = pv_stock.sku \
712                 JOIN inventory_balances ib ON ib.item_id = ii.id \
713                 WHERE pv_stock.product_id = products.id \
714                   AND pv_stock.is_active = 1 \
715                   AND CAST(ib.quantity_available AS REAL) > 0)"
716            ));
717        }
718
719        let params_refs: Vec<&dyn rusqlite::ToSql> =
720            params.iter().map(std::convert::AsRef::as_ref).collect();
721        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
722        let ids = stmt
723            .query_map(params_refs.as_slice(), |row| row.get::<_, String>(0))
724            .map_err(map_db_error)?
725            .collect::<rusqlite::Result<Vec<_>>>()
726            .map_err(map_db_error)?
727            .into_iter()
728            .map(|id_str| parse_uuid(&id_str, "product", "id").map(ProductId::from))
729            .collect::<Result<Vec<_>>>()?;
730
731        if min_price.is_some() || max_price.is_some() {
732            let min_price = min_price.as_ref();
733            let max_price = max_price.as_ref();
734            let mut count = 0u64;
735            for id in ids {
736                let variants = self.get_variants(id)?;
737                let mut matches = false;
738                for variant in variants {
739                    if !variant.is_active {
740                        continue;
741                    }
742                    if let Some(min) = min_price {
743                        if variant.price < *min {
744                            continue;
745                        }
746                    }
747                    if let Some(max) = max_price {
748                        if variant.price > *max {
749                            continue;
750                        }
751                    }
752                    matches = true;
753                    break;
754                }
755                if matches {
756                    count += 1;
757                }
758            }
759            return Ok(count);
760        }
761
762        Ok(ids.len() as u64)
763    }
764
765    // === Batch Operations ===
766
767    fn create_batch(&self, inputs: Vec<CreateProduct>) -> Result<BatchResult<Product>> {
768        validate_batch_size(&inputs)?;
769        let mut result = BatchResult::with_capacity(inputs.len());
770
771        for (index, input) in inputs.into_iter().enumerate() {
772            match self.create(input) {
773                Ok(product) => result.record_success(product),
774                Err(e) => result.record_failure(index, None, &e),
775            }
776        }
777
778        Ok(result)
779    }
780
781    fn create_batch_atomic(&self, inputs: Vec<CreateProduct>) -> Result<Vec<Product>> {
782        validate_batch_size(&inputs)?;
783        if inputs.is_empty() {
784            return Ok(vec![]);
785        }
786
787        let mut conn = self.conn()?;
788        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
789        let mut results = Vec::with_capacity(inputs.len());
790
791        for input in inputs {
792            let id = ProductId::new();
793            let now = Utc::now();
794            let slug = input.slug.clone().unwrap_or_else(|| Product::generate_slug(&input.name));
795            let name = input.name.clone();
796            let description = input.description.clone().unwrap_or_default();
797            let product_type = input.product_type.unwrap_or_default();
798            let attributes = input.attributes.clone().unwrap_or_default();
799            let seo = input.seo.clone();
800
801            // Check slug uniqueness
802            let exists: i32 = tx
803                .query_row("SELECT COUNT(*) FROM products WHERE slug = ?", [&slug], |row| {
804                    row.get(0)
805                })
806                .map_err(map_db_error)?;
807
808            if exists > 0 {
809                return Err(CommerceError::DuplicateSlug(slug));
810            }
811
812            let attributes_json = serde_json::to_string(&attributes).unwrap_or_default();
813            let seo_json = seo.as_ref().map(|s| serde_json::to_string(s).unwrap_or_default());
814
815            tx.execute(
816                "INSERT INTO products (id, name, slug, description, status, product_type, attributes, seo, created_at, updated_at)
817                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
818                rusqlite::params![
819                    id.to_string(),
820                    &name,
821                    &slug,
822                    &description,
823                    ProductStatus::Draft.to_string(),
824                    product_type.to_string(),
825                    attributes_json,
826                    seo_json,
827                    now.to_rfc3339(),
828                    now.to_rfc3339(),
829                ],
830            )
831            .map_err(map_db_error)?;
832
833            // Create variants inline if provided
834            if let Some(variants) = &input.variants {
835                for (i, variant) in variants.iter().enumerate() {
836                    let variant_id = Uuid::new_v4();
837
838                    // Check SKU uniqueness
839                    let sku_exists: i32 = tx
840                        .query_row(
841                            "SELECT COUNT(*) FROM product_variants WHERE sku = ?",
842                            [&variant.sku],
843                            |row| row.get(0),
844                        )
845                        .map_err(map_db_error)?;
846
847                    if sku_exists > 0 {
848                        return Err(CommerceError::DuplicateSku(variant.sku.clone()));
849                    }
850
851                    let options_json =
852                        serde_json::to_string(&variant.options.clone().unwrap_or_default())
853                            .unwrap_or_default();
854
855                    tx.execute(
856                        "INSERT INTO product_variants (id, product_id, sku, name, price, compare_at_price, cost,
857                                                       barcode, weight, weight_unit, options, is_default, is_active,
858                                                       created_at, updated_at)
859                         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)",
860                        rusqlite::params![
861                            variant_id.to_string(),
862                            id.to_string(),
863                            &variant.sku,
864                            variant.name.as_ref().unwrap_or(&variant.sku),
865                            variant.price.to_string(),
866                            variant.compare_at_price.map(|d| d.to_string()),
867                            variant.cost.map(|d| d.to_string()),
868                            &variant.barcode,
869                            variant.weight.map(|d| d.to_string()),
870                            &variant.weight_unit,
871                            options_json,
872                            i32::from(i == 0),  // First variant is default
873                            now.to_rfc3339(),
874                            now.to_rfc3339(),
875                        ],
876                    )
877                    .map_err(map_db_error)?;
878                }
879            }
880
881            results.push(Product {
882                id,
883                name,
884                slug,
885                description,
886                status: ProductStatus::Draft,
887                product_type,
888                attributes,
889                seo,
890                created_at: now,
891                updated_at: now,
892            });
893        }
894
895        tx.commit().map_err(map_db_error)?;
896        Ok(results)
897    }
898
899    fn update_batch(
900        &self,
901        updates: Vec<(ProductId, UpdateProduct)>,
902    ) -> Result<BatchResult<Product>> {
903        validate_batch_size(&updates)?;
904        let mut result = BatchResult::with_capacity(updates.len());
905
906        for (index, (id, input)) in updates.into_iter().enumerate() {
907            match self.update(id, input) {
908                Ok(product) => result.record_success(product),
909                Err(e) => result.record_failure(index, Some(id.to_string()), &e),
910            }
911        }
912
913        Ok(result)
914    }
915
916    fn update_batch_atomic(
917        &self,
918        updates: Vec<(ProductId, UpdateProduct)>,
919    ) -> Result<Vec<Product>> {
920        validate_batch_size(&updates)?;
921        if updates.is_empty() {
922            return Ok(vec![]);
923        }
924
925        let mut conn = self.conn()?;
926        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
927        let mut results = Vec::with_capacity(updates.len());
928
929        for (id, input) in updates {
930            let now = Utc::now();
931            let current_version: i32 = tx
932                .query_row("SELECT version FROM products WHERE id = ?", [id.to_string()], |row| {
933                    row.get(0)
934                })
935                .map_err(|e| match e {
936                    rusqlite::Error::QueryReturnedNoRows => {
937                        CommerceError::ProductNotFound(id.into_uuid())
938                    }
939                    e => map_db_error(e),
940                })?;
941
942            let mut update_parts = vec!["updated_at = ?"];
943            let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(now.to_rfc3339())];
944
945            if let Some(name) = &input.name {
946                update_parts.push("name = ?");
947                params.push(Box::new(name.clone()));
948            }
949            if let Some(slug) = &input.slug {
950                let existing_id: Option<String> = tx
951                    .query_row("SELECT id FROM products WHERE slug = ?", [slug], |row| row.get(0))
952                    .optional()
953                    .map_err(map_db_error)?;
954                if let Some(existing_id) = existing_id {
955                    if existing_id != id.to_string() {
956                        return Err(CommerceError::DuplicateSlug(slug.clone()));
957                    }
958                }
959                update_parts.push("slug = ?");
960                params.push(Box::new(slug.clone()));
961            }
962            if let Some(description) = &input.description {
963                update_parts.push("description = ?");
964                params.push(Box::new(description.clone()));
965            }
966            if let Some(status) = &input.status {
967                update_parts.push("status = ?");
968                params.push(Box::new(status.to_string()));
969            }
970            if let Some(attributes) = &input.attributes {
971                update_parts.push("attributes = ?");
972                params.push(Box::new(serde_json::to_string(attributes).unwrap_or_default()));
973            }
974            if let Some(seo) = &input.seo {
975                update_parts.push("seo = ?");
976                params.push(Box::new(serde_json::to_string(seo).unwrap_or_default()));
977            }
978
979            update_parts.push("version = version + 1");
980            params.push(Box::new(id.to_string()));
981            params.push(Box::new(current_version));
982
983            let sql = format!(
984                "UPDATE products SET {} WHERE id = ? AND version = ?",
985                update_parts.join(", ")
986            );
987
988            let params_refs: Vec<&dyn rusqlite::ToSql> =
989                params.iter().map(std::convert::AsRef::as_ref).collect();
990            let rows_affected = tx.execute(&sql, params_refs.as_slice()).map_err(map_db_error)?;
991            if rows_affected == 0 {
992                return Err(CommerceError::VersionConflict {
993                    entity: "product".to_string(),
994                    id: id.to_string(),
995                    expected_version: current_version,
996                });
997            }
998
999            let product = tx
1000                .query_row(
1001                    "SELECT * FROM products WHERE id = ?",
1002                    [id.to_string()],
1003                    Self::row_to_product,
1004                )
1005                .map_err(map_db_error)?;
1006
1007            results.push(product);
1008        }
1009
1010        tx.commit().map_err(map_db_error)?;
1011        Ok(results)
1012    }
1013
1014    fn delete_batch(&self, ids: Vec<ProductId>) -> Result<BatchResult<ProductId>> {
1015        validate_batch_size(&ids)?;
1016        let mut result = BatchResult::with_capacity(ids.len());
1017
1018        for (index, id) in ids.into_iter().enumerate() {
1019            match self.delete(id) {
1020                Ok(()) => result.record_success(id),
1021                Err(e) => result.record_failure(index, Some(id.to_string()), &e),
1022            }
1023        }
1024
1025        Ok(result)
1026    }
1027
1028    fn delete_batch_atomic(&self, ids: Vec<ProductId>) -> Result<()> {
1029        validate_batch_size(&ids)?;
1030        if ids.is_empty() {
1031            return Ok(());
1032        }
1033
1034        let mut conn = self.conn()?;
1035        let tx = super::begin_immediate(&mut conn).map_err(map_db_error)?;
1036
1037        let placeholders = build_in_clause(ids.len());
1038
1039        // Archive products (soft delete) with IN clause
1040        let sql = format!(
1041            "UPDATE products SET status = 'archived', updated_at = ? WHERE id IN ({placeholders})"
1042        );
1043
1044        // Build params with timestamp first, then IDs
1045        let now = Utc::now().to_rfc3339();
1046        let mut all_params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(now)];
1047        for id in &ids {
1048            all_params.push(Box::new(id.to_string()));
1049        }
1050        let all_params_refs: Vec<&dyn rusqlite::ToSql> =
1051            all_params.iter().map(std::convert::AsRef::as_ref).collect();
1052
1053        tx.execute(&sql, all_params_refs.as_slice()).map_err(map_db_error)?;
1054
1055        tx.commit().map_err(map_db_error)?;
1056        Ok(())
1057    }
1058
1059    fn get_batch(&self, ids: Vec<ProductId>) -> Result<Vec<Product>> {
1060        validate_batch_size(&ids)?;
1061        if ids.is_empty() {
1062            return Ok(vec![]);
1063        }
1064
1065        let conn = self.conn()?;
1066        let placeholders = build_in_clause(ids.len());
1067        let sql = format!("SELECT * FROM products WHERE id IN ({placeholders})");
1068
1069        let uuid_ids: Vec<Uuid> = ids.iter().map(|id| id.into_uuid()).collect();
1070        let params = uuid_params(&uuid_ids);
1071        let params_refs = params_refs(&params);
1072
1073        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
1074        let products = stmt
1075            .query_map(params_refs.as_slice(), Self::row_to_product)
1076            .map_err(map_db_error)?
1077            .collect::<rusqlite::Result<Vec<_>>>()
1078            .map_err(map_db_error)?;
1079
1080        Ok(products)
1081    }
1082}
1083
1084#[cfg(test)]
1085mod tests {
1086    use super::*;
1087    use crate::SqliteDatabase;
1088    use rust_decimal_macros::dec;
1089    use stateset_core::{
1090        CreateProduct, CreateProductVariant, ProductFilter, ProductRepository, ProductStatus,
1091        UpdateProduct,
1092    };
1093
1094    fn fresh_repo() -> SqliteProductRepository {
1095        SqliteDatabase::in_memory().expect("in-memory").products()
1096    }
1097
1098    fn make_product(repo: &SqliteProductRepository, name: &str, slug: &str) -> Product {
1099        repo.create(CreateProduct {
1100            name: name.into(),
1101            slug: Some(slug.into()),
1102            description: Some(format!("Description for {name}")),
1103            product_type: None,
1104            attributes: None,
1105            seo: None,
1106            variants: Some(vec![CreateProductVariant {
1107                sku: format!("SKU-{slug}"),
1108                name: Some("Default".into()),
1109                price: dec!(19.99),
1110                is_default: Some(true),
1111                ..Default::default()
1112            }]),
1113        })
1114        .expect("create product")
1115    }
1116
1117    #[test]
1118    fn create_product_round_trips_with_default_variant() {
1119        let repo = fresh_repo();
1120        let p = make_product(&repo, "Widget", "widget");
1121        assert_eq!(p.name, "Widget");
1122        assert_eq!(p.slug, "widget");
1123
1124        let by_id = repo.get(p.id).expect("ok").expect("found");
1125        assert_eq!(by_id.id, p.id);
1126        let by_slug = repo.get_by_slug("widget").expect("ok").expect("found");
1127        assert_eq!(by_slug.id, p.id);
1128        assert!(repo.get_by_slug("missing-slug").expect("ok").is_none());
1129
1130        let variants = repo.get_variants(p.id).expect("variants");
1131        assert_eq!(variants.len(), 1);
1132    }
1133
1134    #[test]
1135    fn update_product_changes_name_and_status() {
1136        let repo = fresh_repo();
1137        let p = make_product(&repo, "Original", "original");
1138        let updated = repo
1139            .update(
1140                p.id,
1141                UpdateProduct {
1142                    name: Some("Renamed".into()),
1143                    status: Some(ProductStatus::Archived),
1144                    ..Default::default()
1145                },
1146            )
1147            .expect("update");
1148        assert_eq!(updated.name, "Renamed");
1149        assert_eq!(updated.status, ProductStatus::Archived);
1150    }
1151
1152    #[test]
1153    fn list_filters_by_status() {
1154        let repo = fresh_repo();
1155        // Newly-created products default to ProductStatus::Draft.
1156        let draft = make_product(&repo, "Draft", "draft-prod");
1157        let to_archive = make_product(&repo, "ToArchive", "to-archive");
1158        repo.update(
1159            to_archive.id,
1160            UpdateProduct { status: Some(ProductStatus::Archived), ..Default::default() },
1161        )
1162        .expect("archive");
1163
1164        let drafts = repo
1165            .list(ProductFilter { status: Some(ProductStatus::Draft), ..Default::default() })
1166            .expect("draft");
1167        let archived = repo
1168            .list(ProductFilter { status: Some(ProductStatus::Archived), ..Default::default() })
1169            .expect("archived");
1170        assert!(drafts.iter().any(|p| p.id == draft.id));
1171        assert!(archived.iter().any(|p| p.id == to_archive.id));
1172    }
1173
1174    #[test]
1175    fn delete_removes_product() {
1176        let repo = fresh_repo();
1177        let p = make_product(&repo, "DelMe", "del-me");
1178        repo.delete(p.id).expect("delete");
1179        if let Some(found) = repo.get(p.id).expect("ok") {
1180            assert_ne!(found.status, ProductStatus::Active, "deleted product should not be Active");
1181        }
1182    }
1183
1184    #[test]
1185    fn get_variant_by_sku_round_trips() {
1186        let repo = fresh_repo();
1187        let p = make_product(&repo, "VarTest", "var-test");
1188        let variants = repo.get_variants(p.id).expect("ok");
1189        let v = variants.first().expect("default variant exists");
1190        let by_sku = repo.get_variant_by_sku(&v.sku).expect("ok").expect("found");
1191        assert_eq!(by_sku.id, v.id);
1192        assert!(repo.get_variant_by_sku("missing-sku").expect("ok").is_none());
1193    }
1194
1195    #[test]
1196    fn create_batch_returns_per_input_results() {
1197        let repo = fresh_repo();
1198        let mk = |name: &str, slug: &str| CreateProduct {
1199            name: name.into(),
1200            slug: Some(slug.into()),
1201            description: None,
1202            product_type: None,
1203            attributes: None,
1204            seo: None,
1205            variants: Some(vec![CreateProductVariant {
1206                sku: format!("SKU-B-{slug}"),
1207                price: dec!(1),
1208                is_default: Some(true),
1209                ..Default::default()
1210            }]),
1211        };
1212        let result = repo
1213            .create_batch(vec![mk("A", "a-batch"), mk("B", "b-batch"), mk("C", "c-batch")])
1214            .expect("batch");
1215        assert_eq!(result.success_count, 3);
1216        assert_eq!(result.failure_count, 0);
1217    }
1218
1219    #[test]
1220    fn get_unknown_returns_none() {
1221        let repo = fresh_repo();
1222        assert!(repo.get(stateset_core::ProductId::new()).expect("ok").is_none());
1223    }
1224}
stateset_db/sqlite/products.rs

stateset_db/sqlite/
products.rs
