stateset_db/sqlite/
channels.rs

1//! SQLite implementation of the channel repository
2
3use super::{
4    map_db_error, parse_datetime_row, parse_enum_row, parse_json_row, parse_uuid_opt_row,
5    parse_uuid_row, with_immediate_transaction,
6};
7use chrono::Utc;
8use r2d2::Pool;
9use r2d2_sqlite::SqliteConnectionManager;
10use rusqlite::OptionalExtension;
11use stateset_core::{
12    Channel, ChannelFilter, ChannelId, ChannelProductMapping, ChannelProductSyncItem,
13    ChannelRepository, ChannelStatus, ChannelType, CommerceError, CreateChannel, Result,
14    UpdateChannel,
15};
16
17#[derive(Debug)]
18pub struct SqliteChannelRepository {
19    pool: Pool<SqliteConnectionManager>,
20}
21
22impl SqliteChannelRepository {
23    #[must_use]
24    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
25        Self { pool }
26    }
27
28    fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
29        self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
30    }
31
32    fn row_to_channel(row: &rusqlite::Row<'_>) -> rusqlite::Result<Channel> {
33        let tags_json: String = row.get("tags")?;
34        let metadata_json: String = row.get("metadata")?;
35        Ok(Channel {
36            id: parse_uuid_row(&row.get::<_, String>("id")?, "channel", "id")?.into(),
37            name: row.get("name")?,
38            channel_type: parse_enum_row::<ChannelType>(
39                &row.get::<_, String>("channel_type")?,
40                "channel",
41                "channel_type",
42            )?,
43            integration: row.get("integration")?,
44            status: parse_enum_row::<ChannelStatus>(
45                &row.get::<_, String>("status")?,
46                "channel",
47                "status",
48            )?,
49            api_locked: row.get::<_, i32>("api_locked")? != 0,
50            default_warehouse_id: parse_uuid_opt_row(
51                row.get::<_, Option<String>>("default_warehouse_id")?,
52                "channel",
53                "default_warehouse_id",
54            )?
55            .map(Into::into),
56            tags: parse_json_row(&tags_json, "channel", "tags")?,
57            metadata: parse_json_row(&metadata_json, "channel", "metadata")?,
58            created_at: parse_datetime_row(
59                &row.get::<_, String>("created_at")?,
60                "channel",
61                "created_at",
62            )?,
63            updated_at: parse_datetime_row(
64                &row.get::<_, String>("updated_at")?,
65                "channel",
66                "updated_at",
67            )?,
68        })
69    }
70
71    fn row_to_mapping(row: &rusqlite::Row<'_>) -> rusqlite::Result<ChannelProductMapping> {
72        Ok(ChannelProductMapping {
73            channel_id: parse_uuid_row(
74                &row.get::<_, String>("channel_id")?,
75                "mapping",
76                "channel_id",
77            )?
78            .into(),
79            channel_sku: row.get("channel_sku")?,
80            product_id: parse_uuid_row(
81                &row.get::<_, String>("product_id")?,
82                "mapping",
83                "product_id",
84            )?
85            .into(),
86            internal_sku: row.get("internal_sku")?,
87            created_at: parse_datetime_row(
88                &row.get::<_, String>("created_at")?,
89                "mapping",
90                "created_at",
91            )?,
92            updated_at: parse_datetime_row(
93                &row.get::<_, String>("updated_at")?,
94                "mapping",
95                "updated_at",
96            )?,
97        })
98    }
99
100    fn json_err(e: serde_json::Error) -> rusqlite::Error {
101        rusqlite::Error::ToSqlConversionFailure(Box::new(CommerceError::DatabaseError(
102            e.to_string(),
103        )))
104    }
105}
106
107impl ChannelRepository for SqliteChannelRepository {
108    fn create(&self, input: CreateChannel) -> Result<Channel> {
109        let id = ChannelId::new();
110        let id_str = id.to_string();
111        let now_str = Utc::now().to_rfc3339();
112        let tags_json = serde_json::to_string(&input.tags)
113            .map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
114        let metadata_json = serde_json::to_string(&input.metadata)
115            .map_err(|e| CommerceError::DatabaseError(e.to_string()))?;
116
117        with_immediate_transaction(&self.pool, |tx| {
118            tx.execute(
119                "INSERT INTO channels (id, name, channel_type, integration, status, api_locked, default_warehouse_id, tags, metadata, created_at, updated_at)
120                 VALUES (?, ?, ?, ?, 'active', 0, ?, ?, ?, ?, ?)",
121                rusqlite::params![
122                    &id_str,
123                    &input.name,
124                    input.channel_type.to_string(),
125                    &input.integration,
126                    input.default_warehouse_id.map(|w| w.to_string()),
127                    &tags_json,
128                    &metadata_json,
129                    &now_str,
130                    &now_str,
131                ],
132            )?;
133            tx.query_row("SELECT * FROM channels WHERE id = ?", [&id_str], Self::row_to_channel)
134        })
135    }
136
137    fn get(&self, id: ChannelId) -> Result<Option<Channel>> {
138        let conn = self.conn()?;
139        match conn.query_row(
140            "SELECT * FROM channels WHERE id = ?",
141            [id.to_string()],
142            Self::row_to_channel,
143        ) {
144            Ok(c) => Ok(Some(c)),
145            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
146            Err(e) => Err(map_db_error(e)),
147        }
148    }
149
150    fn update(&self, id: ChannelId, input: UpdateChannel) -> Result<Channel> {
151        let id_str = id.to_string();
152        let now_str = Utc::now().to_rfc3339();
153
154        with_immediate_transaction(&self.pool, |tx| {
155            // Reject mutations on locked channels (reads stay allowed elsewhere).
156            let locked: i32 = tx
157                .query_row("SELECT api_locked FROM channels WHERE id = ?", [&id_str], |r| r.get(0))
158                .optional()?
159                .ok_or(rusqlite::Error::QueryReturnedNoRows)?;
160            if locked != 0 {
161                return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
162                    CommerceError::Conflict("channel is API-locked".into()),
163                )));
164            }
165
166            let mut sets = vec!["updated_at = ?".to_string()];
167            let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(now_str.clone())];
168
169            if let Some(ref name) = input.name {
170                sets.push("name = ?".into());
171                params.push(Box::new(name.clone()));
172            }
173            if let Some(ref integration) = input.integration {
174                sets.push("integration = ?".into());
175                params.push(Box::new(integration.clone()));
176            }
177            if let Some(status) = input.status {
178                sets.push("status = ?".into());
179                params.push(Box::new(status.to_string()));
180            }
181            if let Some(wh) = input.default_warehouse_id {
182                sets.push("default_warehouse_id = ?".into());
183                params.push(Box::new(wh.to_string()));
184            }
185            if let Some(ref tags) = input.tags {
186                sets.push("tags = ?".into());
187                params.push(Box::new(serde_json::to_string(tags).map_err(Self::json_err)?));
188            }
189            if let Some(ref metadata) = input.metadata {
190                sets.push("metadata = ?".into());
191                params.push(Box::new(serde_json::to_string(metadata).map_err(Self::json_err)?));
192            }
193
194            let sql = format!("UPDATE channels SET {} WHERE id = ?", sets.join(", "));
195            params.push(Box::new(id_str.clone()));
196            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
197                params.iter().map(|p| p.as_ref()).collect();
198            tx.execute(&sql, param_refs.as_slice())?;
199
200            tx.query_row("SELECT * FROM channels WHERE id = ?", [&id_str], Self::row_to_channel)
201        })
202    }
203
204    fn list(&self, filter: ChannelFilter) -> Result<Vec<Channel>> {
205        let conn = self.conn()?;
206        let mut sql = "SELECT * FROM channels WHERE status != 'deleted'".to_string();
207        let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
208
209        if let Some(t) = filter.channel_type {
210            sql.push_str(" AND channel_type = ?");
211            params.push(Box::new(t.to_string()));
212        }
213        if let Some(s) = filter.status {
214            sql.push_str(" AND status = ?");
215            params.push(Box::new(s.to_string()));
216        }
217        if let Some(ref integration) = filter.integration {
218            sql.push_str(" AND integration = ?");
219            params.push(Box::new(integration.clone()));
220        }
221        if let Some(locked) = filter.api_locked {
222            sql.push_str(" AND api_locked = ?");
223            params.push(Box::new(locked as i32));
224        }
225        sql.push_str(" ORDER BY created_at DESC");
226        crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
227
228        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
229            params.iter().map(|p| p.as_ref()).collect();
230        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
231        let rows = stmt
232            .query_map(param_refs.as_slice(), Self::row_to_channel)
233            .map_err(map_db_error)?
234            .collect::<std::result::Result<Vec<_>, _>>()
235            .map_err(map_db_error)?;
236        Ok(rows)
237    }
238
239    fn delete(&self, id: ChannelId) -> Result<()> {
240        let id_str = id.to_string();
241        with_immediate_transaction(&self.pool, |tx| {
242            let locked: i32 = tx
243                .query_row("SELECT api_locked FROM channels WHERE id = ?", [&id_str], |r| r.get(0))
244                .optional()?
245                .ok_or(rusqlite::Error::QueryReturnedNoRows)?;
246            if locked != 0 {
247                return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
248                    CommerceError::Conflict("channel is API-locked".into()),
249                )));
250            }
251            tx.execute("UPDATE channels SET status = 'deleted' WHERE id = ?", [&id_str])?;
252            Ok(())
253        })
254    }
255
256    fn set_lock(&self, id: ChannelId, locked: bool) -> Result<Channel> {
257        let id_str = id.to_string();
258        let now_str = Utc::now().to_rfc3339();
259        with_immediate_transaction(&self.pool, |tx| {
260            tx.execute(
261                "UPDATE channels SET api_locked = ?, updated_at = ? WHERE id = ?",
262                rusqlite::params![locked as i32, &now_str, &id_str],
263            )?;
264            tx.query_row("SELECT * FROM channels WHERE id = ?", [&id_str], Self::row_to_channel)
265        })
266    }
267
268    fn sync_products(&self, id: ChannelId, items: Vec<ChannelProductSyncItem>) -> Result<u64> {
269        let id_str = id.to_string();
270        let now_str = Utc::now().to_rfc3339();
271        with_immediate_transaction(&self.pool, |tx| {
272            let mut affected: u64 = 0;
273            for item in &items {
274                if item.delete {
275                    affected += tx.execute(
276                        "DELETE FROM channel_product_mappings WHERE channel_id = ? AND channel_sku = ?",
277                        rusqlite::params![&id_str, &item.channel_sku],
278                    )? as u64;
279                } else {
280                    let product_id = item.product_id.ok_or_else(|| {
281                        rusqlite::Error::ToSqlConversionFailure(Box::new(
282                            CommerceError::ValidationError(
283                                "product_id is required when delete is false".into(),
284                            ),
285                        ))
286                    })?;
287                    let internal_sku = item.internal_sku.clone().unwrap_or_default();
288                    affected += tx.execute(
289                        "INSERT INTO channel_product_mappings (channel_id, channel_sku, product_id, internal_sku, created_at, updated_at)
290                         VALUES (?, ?, ?, ?, ?, ?)
291                         ON CONFLICT(channel_id, channel_sku) DO UPDATE SET
292                            product_id = excluded.product_id,
293                            internal_sku = excluded.internal_sku,
294                            updated_at = excluded.updated_at",
295                        rusqlite::params![
296                            &id_str,
297                            &item.channel_sku,
298                            product_id.to_string(),
299                            &internal_sku,
300                            &now_str,
301                            &now_str,
302                        ],
303                    )? as u64;
304                }
305            }
306            Ok(affected)
307        })
308    }
309
310    fn list_product_mappings(&self, id: ChannelId) -> Result<Vec<ChannelProductMapping>> {
311        let conn = self.conn()?;
312        let mut stmt = conn
313            .prepare(
314                "SELECT * FROM channel_product_mappings WHERE channel_id = ? ORDER BY channel_sku",
315            )
316            .map_err(map_db_error)?;
317        let rows = stmt
318            .query_map([id.to_string()], Self::row_to_mapping)
319            .map_err(map_db_error)?
320            .collect::<std::result::Result<Vec<_>, _>>()
321            .map_err(map_db_error)?;
322        Ok(rows)
323    }
324}
325
326#[cfg(test)]
327mod tests {
328    use super::*;
329    use crate::DatabaseConfig;
330    use crate::sqlite::SqliteDatabase;
331    use stateset_core::ProductId;
332
333    fn test_repo() -> SqliteChannelRepository {
334        let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
335        SqliteChannelRepository::new(db.pool().clone())
336    }
337
338    fn new_channel(repo: &SqliteChannelRepository, name: &str, t: ChannelType) -> Channel {
339        repo.create(CreateChannel {
340            name: name.into(),
341            channel_type: t,
342            integration: Some("shopify".into()),
343            default_warehouse_id: None,
344            tags: vec!["retail".into()],
345            metadata: serde_json::json!({"k":"v"}),
346        })
347        .expect("create channel")
348    }
349
350    #[test]
351    fn create_and_get() {
352        let repo = test_repo();
353        let c = new_channel(&repo, "Shopify US", ChannelType::SalesChannel);
354        assert_eq!(c.name, "Shopify US");
355        assert!(!c.api_locked);
356        let fetched = repo.get(c.id).expect("get").expect("found");
357        assert_eq!(fetched.id, c.id);
358        assert_eq!(fetched.tags, vec!["retail".to_string()]);
359    }
360
361    #[test]
362    fn lock_blocks_update_and_delete() {
363        let repo = test_repo();
364        let c = new_channel(&repo, "Locked", ChannelType::SalesChannel);
365        repo.set_lock(c.id, true).expect("lock");
366        let upd = repo.update(c.id, UpdateChannel { name: Some("x".into()), ..Default::default() });
367        assert!(upd.is_err(), "update on locked channel must fail");
368        assert!(repo.delete(c.id).is_err(), "delete on locked channel must fail");
369        repo.set_lock(c.id, false).expect("unlock");
370        assert!(
371            repo.update(c.id, UpdateChannel { name: Some("ok".into()), ..Default::default() })
372                .is_ok()
373        );
374    }
375
376    #[test]
377    fn list_filters_and_excludes_deleted() {
378        let repo = test_repo();
379        new_channel(&repo, "A", ChannelType::SalesChannel);
380        let b = new_channel(&repo, "B", ChannelType::FulfillmentChannel);
381        let all = repo.list(ChannelFilter::default()).expect("list");
382        assert_eq!(all.len(), 2);
383        let only_fc = repo
384            .list(ChannelFilter {
385                channel_type: Some(ChannelType::FulfillmentChannel),
386                ..Default::default()
387            })
388            .expect("list fc");
389        assert_eq!(only_fc.len(), 1);
390        repo.delete(b.id).expect("delete");
391        assert_eq!(repo.list(ChannelFilter::default()).expect("list").len(), 1);
392    }
393
394    #[test]
395    fn sync_products_upsert_and_delete() {
396        let repo = test_repo();
397        let c = new_channel(&repo, "Sync", ChannelType::SalesChannel);
398        let pid = ProductId::new();
399        let n = repo
400            .sync_products(
401                c.id,
402                vec![ChannelProductSyncItem {
403                    channel_sku: "EXT-1".into(),
404                    product_id: Some(pid),
405                    internal_sku: Some("SKU-1".into()),
406                    delete: false,
407                }],
408            )
409            .expect("sync");
410        assert_eq!(n, 1);
411        let mappings = repo.list_product_mappings(c.id).expect("mappings");
412        assert_eq!(mappings.len(), 1);
413        assert_eq!(mappings[0].internal_sku, "SKU-1");
414
415        // delete it
416        repo.sync_products(
417            c.id,
418            vec![ChannelProductSyncItem {
419                channel_sku: "EXT-1".into(),
420                product_id: None,
421                internal_sku: None,
422                delete: true,
423            }],
424        )
425        .expect("sync delete");
426        assert_eq!(repo.list_product_mappings(c.id).expect("mappings").len(), 0);
427    }
428
429    #[test]
430    fn sync_requires_product_when_not_deleting() {
431        let repo = test_repo();
432        let c = new_channel(&repo, "Sync2", ChannelType::SalesChannel);
433        let res = repo.sync_products(
434            c.id,
435            vec![ChannelProductSyncItem {
436                channel_sku: "EXT-9".into(),
437                product_id: None,
438                internal_sku: None,
439                delete: false,
440            }],
441        );
442        assert!(res.is_err());
443    }
444}
stateset_db/sqlite/channels.rs

stateset_db/sqlite/
channels.rs
