1use super::{map_db_error, parse_datetime_row, parse_uuid_row, with_immediate_transaction};
4use chrono::Utc;
5use r2d2::Pool;
6use r2d2_sqlite::SqliteConnectionManager;
7use stateset_core::{
8 CommerceError, CreateIntegrationMapping, IntegrationMapping, IntegrationMappingFilter,
9 IntegrationMappingId, IntegrationMappingRepository, MappingLookup, Result,
10 UpdateIntegrationMapping,
11};
12
13#[derive(Debug)]
14pub struct SqliteIntegrationMappingRepository {
15 pool: Pool<SqliteConnectionManager>,
16}
17
18impl SqliteIntegrationMappingRepository {
19 #[must_use]
20 pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
21 Self { pool }
22 }
23
24 fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
25 self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
26 }
27
28 fn row_to_mapping(row: &rusqlite::Row<'_>) -> rusqlite::Result<IntegrationMapping> {
29 Ok(IntegrationMapping {
30 id: parse_uuid_row(&row.get::<_, String>("id")?, "integration_mapping", "id")?.into(),
31 integration: row.get("integration")?,
32 mapping_group: row.get("mapping_group")?,
33 field_name: row.get("field_name")?,
34 external_value: row.get("external_value")?,
35 internal_value: row.get("internal_value")?,
36 is_active: row.get::<_, i32>("is_active")? != 0,
37 created_at: parse_datetime_row(
38 &row.get::<_, String>("created_at")?,
39 "integration_mapping",
40 "created_at",
41 )?,
42 updated_at: parse_datetime_row(
43 &row.get::<_, String>("updated_at")?,
44 "integration_mapping",
45 "updated_at",
46 )?,
47 })
48 }
49}
50
51impl IntegrationMappingRepository for SqliteIntegrationMappingRepository {
52 fn create(&self, input: CreateIntegrationMapping) -> Result<IntegrationMapping> {
53 let id = IntegrationMappingId::new();
54 let id_str = id.to_string();
55 let now_str = Utc::now().to_rfc3339();
56 with_immediate_transaction(&self.pool, |tx| {
57 tx.execute(
58 "INSERT INTO integration_mappings (id, integration, mapping_group, field_name, external_value, internal_value, is_active, created_at, updated_at)
59 VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?)",
60 rusqlite::params![
61 &id_str,
62 &input.integration,
63 &input.mapping_group,
64 &input.field_name,
65 &input.external_value,
66 &input.internal_value,
67 &now_str,
68 &now_str,
69 ],
70 )?;
71 tx.query_row(
72 "SELECT * FROM integration_mappings WHERE id = ?",
73 [&id_str],
74 Self::row_to_mapping,
75 )
76 })
77 }
78
79 fn get(&self, id: IntegrationMappingId) -> Result<Option<IntegrationMapping>> {
80 let conn = self.conn()?;
81 match conn.query_row(
82 "SELECT * FROM integration_mappings WHERE id = ?",
83 [id.to_string()],
84 Self::row_to_mapping,
85 ) {
86 Ok(m) => Ok(Some(m)),
87 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
88 Err(e) => Err(map_db_error(e)),
89 }
90 }
91
92 fn update(
93 &self,
94 id: IntegrationMappingId,
95 input: UpdateIntegrationMapping,
96 ) -> Result<IntegrationMapping> {
97 let id_str = id.to_string();
98 let now_str = Utc::now().to_rfc3339();
99 with_immediate_transaction(&self.pool, |tx| {
100 let mut sets = vec!["updated_at = ?".to_string()];
101 let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(now_str.clone())];
102 if let Some(ref internal_value) = input.internal_value {
103 sets.push("internal_value = ?".into());
104 params.push(Box::new(internal_value.clone()));
105 }
106 if let Some(is_active) = input.is_active {
107 sets.push("is_active = ?".into());
108 params.push(Box::new(is_active as i32));
109 }
110 let sql = format!("UPDATE integration_mappings SET {} WHERE id = ?", sets.join(", "));
111 params.push(Box::new(id_str.clone()));
112 let param_refs: Vec<&dyn rusqlite::types::ToSql> =
113 params.iter().map(|p| p.as_ref()).collect();
114 tx.execute(&sql, param_refs.as_slice())?;
115 tx.query_row(
116 "SELECT * FROM integration_mappings WHERE id = ?",
117 [&id_str],
118 Self::row_to_mapping,
119 )
120 })
121 }
122
123 fn list(&self, filter: IntegrationMappingFilter) -> Result<Vec<IntegrationMapping>> {
124 let conn = self.conn()?;
125 let mut sql = "SELECT * FROM integration_mappings WHERE 1=1".to_string();
126 let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![];
127 if let Some(ref integration) = filter.integration {
128 sql.push_str(" AND integration = ?");
129 params.push(Box::new(integration.clone()));
130 }
131 if let Some(ref mapping_group) = filter.mapping_group {
132 sql.push_str(" AND mapping_group = ?");
133 params.push(Box::new(mapping_group.clone()));
134 }
135 if let Some(ref field_name) = filter.field_name {
136 sql.push_str(" AND field_name = ?");
137 params.push(Box::new(field_name.clone()));
138 }
139 if let Some(is_active) = filter.is_active {
140 sql.push_str(" AND is_active = ?");
141 params.push(Box::new(is_active as i32));
142 }
143 sql.push_str(" ORDER BY mapping_group, field_name, external_value");
144 crate::sqlite::append_limit_offset(&mut sql, filter.limit, filter.offset);
145 let param_refs: Vec<&dyn rusqlite::types::ToSql> =
146 params.iter().map(|p| p.as_ref()).collect();
147 let mut stmt = conn.prepare(&sql).map_err(map_db_error)?;
148 let rows = stmt
149 .query_map(param_refs.as_slice(), Self::row_to_mapping)
150 .map_err(map_db_error)?
151 .collect::<std::result::Result<Vec<_>, _>>()
152 .map_err(map_db_error)?;
153 Ok(rows)
154 }
155
156 fn delete(&self, id: IntegrationMappingId) -> Result<()> {
157 let conn = self.conn()?;
158 conn.execute("DELETE FROM integration_mappings WHERE id = ?", [id.to_string()])
159 .map_err(map_db_error)?;
160 Ok(())
161 }
162
163 fn bulk_upsert(&self, items: Vec<CreateIntegrationMapping>) -> Result<u64> {
164 let now_str = Utc::now().to_rfc3339();
165 with_immediate_transaction(&self.pool, |tx| {
166 let mut affected: u64 = 0;
167 for item in &items {
168 let id_str = IntegrationMappingId::new().to_string();
169 affected += tx.execute(
170 "INSERT INTO integration_mappings (id, integration, mapping_group, field_name, external_value, internal_value, is_active, created_at, updated_at)
171 VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?)
172 ON CONFLICT(integration, mapping_group, field_name, external_value) DO UPDATE SET
173 internal_value = excluded.internal_value,
174 is_active = 1,
175 updated_at = excluded.updated_at",
176 rusqlite::params![
177 &id_str,
178 &item.integration,
179 &item.mapping_group,
180 &item.field_name,
181 &item.external_value,
182 &item.internal_value,
183 &now_str,
184 &now_str,
185 ],
186 )? as u64;
187 }
188 Ok(affected)
189 })
190 }
191
192 fn resolve(&self, lookup: &MappingLookup) -> Result<Option<String>> {
193 let conn = self.conn()?;
194 let value: Option<String> = conn
195 .query_row(
196 "SELECT internal_value FROM integration_mappings
197 WHERE integration = ? AND mapping_group = ? AND field_name = ? AND external_value = ? AND is_active = 1",
198 rusqlite::params![
199 &lookup.integration,
200 &lookup.mapping_group,
201 &lookup.field_name,
202 &lookup.external_value,
203 ],
204 |r| r.get(0),
205 )
206 .ok();
207 Ok(value)
208 }
209}
210
211#[cfg(test)]
212mod tests {
213 use super::*;
214 use crate::DatabaseConfig;
215 use crate::sqlite::SqliteDatabase;
216
217 fn test_repo() -> SqliteIntegrationMappingRepository {
218 let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
219 SqliteIntegrationMappingRepository::new(db.pool().clone())
220 }
221
222 fn carrier(external: &str, internal: &str) -> CreateIntegrationMapping {
223 CreateIntegrationMapping {
224 integration: "shopify".into(),
225 mapping_group: "carrier".into(),
226 field_name: "carrier_code".into(),
227 external_value: external.into(),
228 internal_value: internal.into(),
229 }
230 }
231
232 fn lookup(external: &str) -> MappingLookup {
233 MappingLookup {
234 integration: "shopify".into(),
235 mapping_group: "carrier".into(),
236 field_name: "carrier_code".into(),
237 external_value: external.into(),
238 }
239 }
240
241 #[test]
242 fn create_and_resolve() {
243 let repo = test_repo();
244 repo.create(carrier("USPS Ground", "usps")).expect("create");
245 assert_eq!(
246 repo.resolve(&lookup("USPS Ground")).expect("resolve"),
247 Some("usps".to_string())
248 );
249 assert_eq!(repo.resolve(&lookup("Unknown")).expect("resolve"), None);
250 }
251
252 #[test]
253 fn inactive_mapping_does_not_resolve() {
254 let repo = test_repo();
255 let m = repo.create(carrier("FedEx Home", "fedex")).expect("create");
256 repo.update(
257 m.id,
258 UpdateIntegrationMapping { is_active: Some(false), ..Default::default() },
259 )
260 .expect("deactivate");
261 assert_eq!(repo.resolve(&lookup("FedEx Home")).expect("resolve"), None);
262 }
263
264 #[test]
265 fn bulk_upsert_inserts_then_updates() {
266 let repo = test_repo();
267 let n = repo.bulk_upsert(vec![carrier("A", "a"), carrier("B", "b")]).expect("bulk");
268 assert_eq!(n, 2);
269 repo.bulk_upsert(vec![carrier("A", "a2")]).expect("bulk2");
271 assert_eq!(repo.resolve(&lookup("A")).expect("resolve"), Some("a2".to_string()));
272 assert_eq!(repo.list(IntegrationMappingFilter::default()).expect("list").len(), 2);
273 }
274
275 #[test]
276 fn list_filters_by_group() {
277 let repo = test_repo();
278 repo.create(carrier("A", "a")).expect("create");
279 repo.create(CreateIntegrationMapping {
280 integration: "shopify".into(),
281 mapping_group: "order_status".into(),
282 field_name: "status".into(),
283 external_value: "fulfilled".into(),
284 internal_value: "shipped".into(),
285 })
286 .expect("create");
287 let carriers = repo
288 .list(IntegrationMappingFilter {
289 mapping_group: Some("carrier".into()),
290 ..Default::default()
291 })
292 .expect("list");
293 assert_eq!(carriers.len(), 1);
294 }
295}