stateset_db/sqlite/
integration_field_mappings.rs

1//! SQLite implementation of the integration field-mapping repository
2
3use super::{
4    map_db_error, parse_datetime_row, parse_enum_row, parse_uuid_row, with_immediate_transaction,
5};
6use chrono::Utc;
7use r2d2::Pool;
8use r2d2_sqlite::SqliteConnectionManager;
9use stateset_core::{
10    CommerceError, CreateIntegrationFieldMapping, FieldTransform, IntegrationFieldMapping,
11    IntegrationFieldMappingFilter, IntegrationFieldMappingId, IntegrationFieldMappingRepository,
12    Result, UpdateIntegrationFieldMapping,
13};
14
15#[derive(Debug)]
16pub struct SqliteIntegrationFieldMappingRepository {
17    pool: Pool<SqliteConnectionManager>,
18}
19
20impl SqliteIntegrationFieldMappingRepository {
21    #[must_use]
22    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
23        Self { pool }
24    }
25
26    fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
27        self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
28    }
29
30    fn row_to_mapping(row: &rusqlite::Row<'_>) -> rusqlite::Result<IntegrationFieldMapping> {
31        Ok(IntegrationFieldMapping {
32            id: parse_uuid_row(&row.get::<_, String>("id")?, "ifm", "id")?.into(),
33            integration_account: row.get("integration_account")?,
34            mapping_group: row.get("mapping_group")?,
35            source_field: row.get("source_field")?,
36            destination_field: row.get("destination_field")?,
37            template: row.get("template")?,
38            transform: parse_enum_row::<FieldTransform>(
39                &row.get::<_, String>("transform")?,
40                "ifm",
41                "transform",
42            )?,
43            fallback: row.get("fallback")?,
44            is_active: row.get::<_, i32>("is_active")? != 0,
45            created_at: parse_datetime_row(
46                &row.get::<_, String>("created_at")?,
47                "ifm",
48                "created_at",
49            )?,
50            updated_at: parse_datetime_row(
51                &row.get::<_, String>("updated_at")?,
52                "ifm",
53                "updated_at",
54            )?,
55        })
56    }
57
58    fn insert(
59        tx: &rusqlite::Connection,
60        input: &CreateIntegrationFieldMapping,
61        now: &str,
62    ) -> rusqlite::Result<String> {
63        let id_str = IntegrationFieldMappingId::new().to_string();
64        tx.execute(
65            "INSERT INTO integration_field_mappings (id, integration_account, mapping_group, source_field, destination_field, template, transform, fallback, is_active, created_at, updated_at)
66             VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)",
67            rusqlite::params![
68                &id_str,
69                &input.integration_account,
70                &input.mapping_group,
71                &input.source_field,
72                &input.destination_field,
73                &input.template,
74                input.transform.to_string(),
75                &input.fallback,
76                now,
77                now,
78            ],
79        )?;
80        Ok(id_str)
81    }
82}
83
84impl IntegrationFieldMappingRepository for SqliteIntegrationFieldMappingRepository {
85    fn create(&self, input: CreateIntegrationFieldMapping) -> Result<IntegrationFieldMapping> {
86        let now = Utc::now().to_rfc3339();
87        with_immediate_transaction(&self.pool, |tx| {
88            let id = Self::insert(tx, &input, &now)?;
89            tx.query_row(
90                "SELECT * FROM integration_field_mappings WHERE id = ?",
91                [&id],
92                Self::row_to_mapping,
93            )
94        })
95    }
96
97    fn get(&self, id: IntegrationFieldMappingId) -> Result<Option<IntegrationFieldMapping>> {
98        let conn = self.conn()?;
99        match conn.query_row(
100            "SELECT * FROM integration_field_mappings WHERE id = ?",
101            [id.to_string()],
102            Self::row_to_mapping,
103        ) {
104            Ok(m) => Ok(Some(m)),
105            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
106            Err(e) => Err(map_db_error(e)),
107        }
108    }
109
110    fn update(
111        &self,
112        id: IntegrationFieldMappingId,
113        input: UpdateIntegrationFieldMapping,
114    ) -> Result<IntegrationFieldMapping> {
115        let id_str = id.to_string();
116        let now = Utc::now().to_rfc3339();
117        with_immediate_transaction(&self.pool, |tx| {
118            let mut sets = vec!["updated_at = ?".to_string()];
119            let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(now.clone())];
120            if let Some(ref dest) = input.destination_field {
121                sets.push("destination_field = ?".into());
122                params.push(Box::new(dest.clone()));
123            }
124            if let Some(ref template) = input.template {
125                sets.push("template = ?".into());
126                params.push(Box::new(template.clone()));
127            }
128            if let Some(transform) = input.transform {
129                sets.push("transform = ?".into());
130                params.push(Box::new(transform.to_string()));
131            }
132            if let Some(ref fallback) = input.fallback {
133                sets.push("fallback = ?".into());
134                params.push(Box::new(fallback.clone()));
135            }
136            if let Some(is_active) = input.is_active {
137                sets.push("is_active = ?".into());
138                params.push(Box::new(is_active as i32));
139            }
140            let sql =
141                format!("UPDATE integration_field_mappings SET {} WHERE id = ?", sets.join(", "));
142            params.push(Box::new(id_str.clone()));
143            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
144                params.iter().map(|p| p.as_ref()).collect();
145            tx.execute(&sql, param_refs.as_slice())?;
146            tx.query_row(
147                "SELECT * FROM integration_field_mappings WHERE id = ?",
148                [&id_str],
149                Self::row_to_mapping,
150            )
151        })
152    }
153
154    fn list(&self, filter: IntegrationFieldMappingFilter) -> Result<Vec<IntegrationFieldMapping>> {
155        let conn = self.conn()?;
156        let mut sql = "SELECT * FROM integration_field_mappings WHERE 1=1".to_string();
157        let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
158        if let Some(ref account) = filter.integration_account {
159            sql.push_str(" AND integration_account = ?");
160            params.push(Box::new(account.clone()));
161        }
162        if let Some(ref group) = filter.mapping_group {
163            sql.push_str(" AND mapping_group = ?");
164            params.push(Box::new(group.clone()));
165        }
166        if let Some(ref source) = filter.source_field {
167            sql.push_str(" AND source_field = ?");
168            params.push(Box::new(source.clone()));
169        }
170        if let Some(is_active) = filter.is_active {
171            sql.push_str(" AND is_active = ?");
172            params.push(Box::new(is_active as i32));
173        }
174        sql.push_str(" ORDER BY mapping_group, source_field");
175        crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
176        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
177            params.iter().map(|p| p.as_ref()).collect();
178        let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
179        let rows = stmt
180            .query_map(param_refs.as_slice(), Self::row_to_mapping)
181            .map_err(map_db_error)?
182            .collect::<std::result::Result<Vec<_>, _>>()
183            .map_err(map_db_error)?;
184        Ok(rows)
185    }
186
187    fn delete(&self, id: IntegrationFieldMappingId) -> Result<()> {
188        let conn = self.conn()?;
189        conn.execute("DELETE FROM integration_field_mappings WHERE id = ?", [id.to_string()])
190            .map_err(map_db_error)?;
191        Ok(())
192    }
193
194    fn bulk_create(&self, items: Vec<CreateIntegrationFieldMapping>) -> Result<u64> {
195        let now = Utc::now().to_rfc3339();
196        with_immediate_transaction(&self.pool, |tx| {
197            let mut count = 0u64;
198            for item in &items {
199                Self::insert(tx, item, &now)?;
200                count += 1;
201            }
202            Ok(count)
203        })
204    }
205
206    fn bulk_delete(&self, ids: Vec<IntegrationFieldMappingId>) -> Result<u64> {
207        with_immediate_transaction(&self.pool, |tx| {
208            let mut count = 0u64;
209            for id in &ids {
210                count += tx.execute(
211                    "DELETE FROM integration_field_mappings WHERE id = ?",
212                    [id.to_string()],
213                )? as u64;
214            }
215            Ok(count)
216        })
217    }
218
219    fn distinct_groups(&self, integration_account: &str) -> Result<Vec<String>> {
220        let conn = self.conn()?;
221        let mut stmt = conn
222            .prepare("SELECT DISTINCT mapping_group FROM integration_field_mappings WHERE integration_account = ? ORDER BY mapping_group")
223            .map_err(map_db_error)?;
224        let rows = stmt
225            .query_map([integration_account], |r| r.get::<_, String>(0))
226            .map_err(map_db_error)?
227            .collect::<std::result::Result<Vec<_>, _>>()
228            .map_err(map_db_error)?;
229        Ok(rows)
230    }
231}
232
233#[cfg(test)]
234mod tests {
235    use super::*;
236    use crate::DatabaseConfig;
237    use crate::sqlite::SqliteDatabase;
238
239    fn test_repo() -> SqliteIntegrationFieldMappingRepository {
240        let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
241        SqliteIntegrationFieldMappingRepository::new(db.pool().clone())
242    }
243
244    fn mapping(group: &str, source: &str) -> CreateIntegrationFieldMapping {
245        CreateIntegrationFieldMapping {
246            integration_account: "acct-1".into(),
247            mapping_group: group.into(),
248            source_field: source.into(),
249            destination_field: "dest".into(),
250            template: None,
251            transform: FieldTransform::Uppercase,
252            fallback: Some("NA".into()),
253        }
254    }
255
256    #[test]
257    fn create_get_update() {
258        let repo = test_repo();
259        let m = repo.create(mapping("order", "order.email")).expect("create");
260        assert_eq!(m.transform, FieldTransform::Uppercase);
261        let fetched = repo.get(m.id).expect("get").expect("found");
262        assert_eq!(fetched.source_field, "order.email");
263
264        let updated = repo
265            .update(
266                m.id,
267                UpdateIntegrationFieldMapping { is_active: Some(false), ..Default::default() },
268            )
269            .expect("update");
270        assert!(!updated.is_active);
271    }
272
273    #[test]
274    fn bulk_create_and_delete() {
275        let repo = test_repo();
276        let n = repo
277            .bulk_create(vec![
278                mapping("order", "a"),
279                mapping("order", "b"),
280                mapping("shipment", "c"),
281            ])
282            .expect("bulk");
283        assert_eq!(n, 3);
284        let all = repo.list(IntegrationFieldMappingFilter::default()).expect("list");
285        assert_eq!(all.len(), 3);
286        let ids: Vec<_> = all.iter().take(2).map(|m| m.id).collect();
287        let deleted = repo.bulk_delete(ids).expect("bulk delete");
288        assert_eq!(deleted, 2);
289        assert_eq!(repo.list(IntegrationFieldMappingFilter::default()).expect("list").len(), 1);
290    }
291
292    #[test]
293    fn distinct_groups_dedups() {
294        let repo = test_repo();
295        repo.bulk_create(vec![
296            mapping("order", "a"),
297            mapping("order", "b"),
298            mapping("shipment", "c"),
299        ])
300        .expect("bulk");
301        let groups = repo.distinct_groups("acct-1").expect("groups");
302        assert_eq!(groups, vec!["order".to_string(), "shipment".to_string()]);
303    }
304
305    #[test]
306    fn list_filters_by_group() {
307        let repo = test_repo();
308        repo.bulk_create(vec![mapping("order", "a"), mapping("shipment", "c")]).expect("bulk");
309        let order = repo
310            .list(IntegrationFieldMappingFilter {
311                mapping_group: Some("order".into()),
312                ..Default::default()
313            })
314            .expect("list");
315        assert_eq!(order.len(), 1);
316    }
317}
stateset_db/sqlite/integration_field_mappings.rs

stateset_db/sqlite/
integration_field_mappings.rs
