autofill/sync/
engine.rs

1/* This Source Code Form is subject to the terms of the Mozilla Public
2* License, v. 2.0. If a copy of the MPL was not distributed with this
3* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
4
5use super::{plan_incoming, ProcessIncomingRecordImpl, ProcessOutgoingRecordImpl, SyncRecord};
6use crate::error::*;
7use crate::Store;
8use error_support::warn;
9use rusqlite::{
10    types::{FromSql, ToSql},
11    Connection, Transaction,
12};
13use std::sync::Arc;
14use sync15::bso::{IncomingBso, OutgoingBso};
15use sync15::engine::{CollSyncIds, CollectionRequest, EngineSyncAssociation, SyncEngine};
16use sync15::{telemetry, CollectionName, ServerTimestamp};
17use sync_guid::Guid;
18
19// We have 2 engines in this crate and they are identical except for stuff
20// abstracted here!
21pub struct EngineConfig {
22    pub(crate) namespace: String,          // prefix for meta keys, etc.
23    pub(crate) collection: CollectionName, // collection name on the server.
24}
25
26// meta keys, will be prefixed by the "namespace"
27pub const LAST_SYNC_META_KEY: &str = "last_sync_time";
28pub const GLOBAL_SYNCID_META_KEY: &str = "global_sync_id";
29pub const COLLECTION_SYNCID_META_KEY: &str = "sync_id";
30
31// A trait to abstract the broader sync processes.
32// Send + Sync is required to use a `ConfigSyncEngine` as a `BridgedEngine`.
33pub trait SyncEngineStorageImpl<T>: Send + Sync {
34    fn get_incoming_impl(
35        &self,
36        enc_key: &Option<String>,
37    ) -> Result<Box<dyn ProcessIncomingRecordImpl<Record = T>>>;
38    fn reset_storage(&self, conn: &Transaction<'_>) -> Result<()>;
39    fn get_outgoing_impl(
40        &self,
41        enc_key: &Option<String>,
42    ) -> Result<Box<dyn ProcessOutgoingRecordImpl<Record = T>>>;
43}
44
45// A sync engine that gets functionality from an EngineConfig.
46pub struct ConfigSyncEngine<T> {
47    pub(crate) config: EngineConfig,
48    pub(crate) store: Arc<Store>,
49    pub(crate) storage_impl: Box<dyn SyncEngineStorageImpl<T>>,
50    local_enc_key: Option<String>,
51}
52
53impl<T> ConfigSyncEngine<T> {
54    pub fn new(
55        config: EngineConfig,
56        store: Arc<Store>,
57        storage_impl: Box<dyn SyncEngineStorageImpl<T>>,
58    ) -> Self {
59        Self {
60            config,
61            store,
62            storage_impl,
63            local_enc_key: None,
64        }
65    }
66    fn put_meta(&self, conn: &Connection, tail: &str, value: &dyn ToSql) -> Result<()> {
67        let key = format!("{}.{}", self.config.namespace, tail);
68        crate::db::store::put_meta(conn, &key, value)
69    }
70    fn get_meta<V: FromSql>(&self, conn: &Connection, tail: &str) -> Result<Option<V>> {
71        let key = format!("{}.{}", self.config.namespace, tail);
72        crate::db::store::get_meta(conn, &key)
73    }
74    fn delete_meta(&self, conn: &Connection, tail: &str) -> Result<()> {
75        let key = format!("{}.{}", self.config.namespace, tail);
76        crate::db::store::delete_meta(conn, &key)
77    }
78
79    // Reset the local sync data so the next server request fetches all records.
80    pub fn reset_local_sync_data(&self) -> Result<()> {
81        let db = self.store.lock_db()?;
82        let tx = db.unchecked_transaction()?;
83        self.storage_impl.reset_storage(&tx)?;
84        self.put_meta(&tx, LAST_SYNC_META_KEY, &0)?;
85        tx.commit()?;
86        Ok(())
87    }
88
89    // We cannot call `reset_local_sync_data` from  the store's `scrub_undecryptable_credit_card_data_for_remote_replacement`
90    // function, so we've created this function specifically for it's use. It should not be used anywhere else.
91    pub fn reset_local_sync_data_for_verification(&self, conn: &Connection) -> Result<()> {
92        let tx = conn.unchecked_transaction()?;
93        self.storage_impl.reset_storage(&tx)?;
94        self.put_meta(&tx, LAST_SYNC_META_KEY, &0)?;
95        tx.commit()?;
96        Ok(())
97    }
98}
99
100impl<T: SyncRecord + std::fmt::Debug> SyncEngine for ConfigSyncEngine<T> {
101    fn collection_name(&self) -> CollectionName {
102        self.config.collection.clone()
103    }
104
105    fn set_local_encryption_key(&mut self, key: &str) -> anyhow::Result<()> {
106        self.local_enc_key = Some(key.to_string());
107        Ok(())
108    }
109
110    fn sync_started(&self) -> anyhow::Result<()> {
111        let db = self.store.lock_db()?;
112        let signal = db.begin_interrupt_scope()?;
113        crate::db::schema::create_empty_sync_temp_tables(&db.writer)?;
114        signal.err_if_interrupted()?;
115        Ok(())
116    }
117
118    fn stage_incoming(
119        &self,
120        inbound: Vec<IncomingBso>,
121        telem: &mut telemetry::Engine,
122    ) -> anyhow::Result<()> {
123        let db = self.store.lock_db()?;
124        let signal = db.begin_interrupt_scope()?;
125
126        // Stage all incoming items.
127        let mut incoming_telemetry = telemetry::EngineIncoming::new();
128        incoming_telemetry.applied(inbound.len() as u32);
129        telem.incoming(incoming_telemetry);
130        let tx = db.writer.unchecked_transaction()?;
131        let incoming_impl = self.storage_impl.get_incoming_impl(&self.local_enc_key)?;
132
133        incoming_impl.stage_incoming(&tx, inbound, &signal)?;
134        tx.commit()?;
135        Ok(())
136    }
137
138    fn apply(
139        &self,
140        timestamp: ServerTimestamp,
141        _telem: &mut telemetry::Engine,
142    ) -> anyhow::Result<Vec<OutgoingBso>> {
143        let db = self.store.lock_db()?;
144        let signal = db.begin_interrupt_scope()?;
145        let tx = db.writer.unchecked_transaction()?;
146        let incoming_impl = self.storage_impl.get_incoming_impl(&self.local_enc_key)?;
147        let outgoing_impl = self.storage_impl.get_outgoing_impl(&self.local_enc_key)?;
148
149        // Get "states" for each record...
150        for state in incoming_impl.fetch_incoming_states(&tx)? {
151            signal.err_if_interrupted()?;
152            // Finally get a "plan" and apply it.
153            let action = plan_incoming(&*incoming_impl, &tx, state)?;
154            super::apply_incoming_action(&*incoming_impl, &tx, action)?;
155        }
156
157        // The timestamp value is handled differently in desktop v mobile. Record a
158        // timestamp if we are given one now, so if we are interrupted merging or
159        // creating outgoing changesets we don't need to re-download the same
160        // records.
161        if timestamp != ServerTimestamp(0) {
162            self.put_meta(&tx, LAST_SYNC_META_KEY, &timestamp.as_millis())?;
163        }
164
165        incoming_impl.finish_incoming(&tx)?;
166
167        // Finally, stage outgoing items.
168        let outgoing = outgoing_impl.fetch_outgoing_records(&tx)?;
169        // we're committing now because it may take a long time to actually perform the upload
170        // and we've already staged everything we need to complete the sync in a way that
171        // doesn't require the transaction to stay alive, so we commit now and start a new
172        // transaction once complete
173        tx.commit()?;
174        Ok(outgoing)
175    }
176
177    fn set_uploaded(&self, new_timestamp: ServerTimestamp, ids: Vec<Guid>) -> anyhow::Result<()> {
178        let db = self.store.lock_db()?;
179        self.put_meta(&db.writer, LAST_SYNC_META_KEY, &new_timestamp.as_millis())?;
180        let tx = db.writer.unchecked_transaction()?;
181        let outgoing_impl = self.storage_impl.get_outgoing_impl(&self.local_enc_key)?;
182        outgoing_impl.finish_synced_items(&tx, ids)?;
183        tx.commit()?;
184        Ok(())
185    }
186
187    fn get_collection_request(
188        &self,
189        server_timestamp: ServerTimestamp,
190    ) -> anyhow::Result<Option<CollectionRequest>> {
191        let db = self.store.lock_db()?;
192        let since = ServerTimestamp(
193            self.get_meta::<i64>(&db.writer, LAST_SYNC_META_KEY)?
194                .unwrap_or_default(),
195        );
196        Ok(if since == server_timestamp {
197            None
198        } else {
199            Some(
200                CollectionRequest::new(self.collection_name())
201                    .full()
202                    .newer_than(since),
203            )
204        })
205    }
206
207    fn get_sync_assoc(&self) -> anyhow::Result<EngineSyncAssociation> {
208        let db = self.store.lock_db()?;
209        let global = self.get_meta(&db.writer, GLOBAL_SYNCID_META_KEY)?;
210        let coll = self.get_meta(&db.writer, COLLECTION_SYNCID_META_KEY)?;
211        Ok(if let (Some(global), Some(coll)) = (global, coll) {
212            EngineSyncAssociation::Connected(CollSyncIds { global, coll })
213        } else {
214            EngineSyncAssociation::Disconnected
215        })
216    }
217
218    fn reset(&self, assoc: &EngineSyncAssociation) -> anyhow::Result<()> {
219        let db = self.store.lock_db()?;
220        let tx = db.unchecked_transaction()?;
221        self.storage_impl.reset_storage(&tx)?;
222        // Reset the last sync time, so that the next sync fetches fresh records
223        // from the server.
224        self.put_meta(&tx, LAST_SYNC_META_KEY, &0)?;
225
226        // Clear the sync ID if we're signing out, or set it to whatever the
227        // server gave us if we're signing in.
228        match assoc {
229            EngineSyncAssociation::Disconnected => {
230                self.delete_meta(&tx, GLOBAL_SYNCID_META_KEY)?;
231                self.delete_meta(&tx, COLLECTION_SYNCID_META_KEY)?;
232            }
233            EngineSyncAssociation::Connected(ids) => {
234                self.put_meta(&tx, GLOBAL_SYNCID_META_KEY, &ids.global)?;
235                self.put_meta(&tx, COLLECTION_SYNCID_META_KEY, &ids.coll)?;
236            }
237        }
238
239        tx.commit()?;
240        Ok(())
241    }
242
243    fn wipe(&self) -> anyhow::Result<()> {
244        warn!("not implemented as there isn't a valid use case for it");
245        Ok(())
246    }
247
248    fn last_sync(&self) -> anyhow::Result<Option<ServerTimestamp>> {
249        let db = self.store.lock_db()?;
250        Ok(self
251            .get_meta::<i64>(&db.writer, LAST_SYNC_META_KEY)?
252            .map(ServerTimestamp::from_millis))
253    }
254
255    fn reset_last_sync(&self) -> anyhow::Result<()> {
256        let db = self.store.lock_db()?;
257        self.delete_meta(&db.writer, LAST_SYNC_META_KEY)?;
258        Ok(())
259    }
260}
261
262#[cfg(test)]
263mod tests {
264    use super::*;
265    use crate::db::credit_cards::add_internal_credit_card;
266    use crate::db::credit_cards::tests::{
267        get_all, insert_tombstone_record, test_insert_mirror_record,
268    };
269    use crate::db::models::credit_card::InternalCreditCard;
270    use crate::db::schema::create_empty_sync_temp_tables;
271    use crate::encryption::EncryptorDecryptor;
272    use crate::sync::{IncomingBso, UnknownFields};
273    use nss_as::ensure_initialized;
274    use sql_support::ConnExt;
275
276    impl InternalCreditCard {
277        pub fn into_test_incoming_bso(
278            self,
279            encdec: &EncryptorDecryptor,
280            unknown_fields: UnknownFields,
281        ) -> IncomingBso {
282            let mut payload = self.into_payload(encdec).expect("is json");
283            payload.entry.unknown_fields = unknown_fields;
284            IncomingBso::from_test_content(payload)
285        }
286    }
287
288    // We use the credit-card engine here.
289    fn create_engine() -> ConfigSyncEngine<InternalCreditCard> {
290        let store = crate::db::store::Store::new_memory();
291        crate::sync::credit_card::create_engine(Arc::new(store))
292    }
293
294    pub fn clear_cc_tables(conn: &Connection) -> rusqlite::Result<(), rusqlite::Error> {
295        conn.execute_all(&[
296            "DELETE FROM credit_cards_data;",
297            "DELETE FROM credit_cards_mirror;",
298            "DELETE FROM credit_cards_tombstones;",
299            "DELETE FROM moz_meta;",
300        ])
301    }
302
303    #[test]
304    fn test_credit_card_engine_apply_timestamp() -> Result<()> {
305        ensure_initialized();
306        let mut credit_card_engine = create_engine();
307        let test_key = crate::encryption::create_autofill_key().unwrap();
308        credit_card_engine
309            .set_local_encryption_key(&test_key)
310            .unwrap();
311        {
312            let db = credit_card_engine.store.lock_db()?;
313            create_empty_sync_temp_tables(&db.writer)?;
314        }
315
316        let mut telem = telemetry::Engine::new("whatever");
317        let last_sync = 24;
318        let result = credit_card_engine.apply(ServerTimestamp::from_millis(last_sync), &mut telem);
319        assert!(result.is_ok());
320
321        // check that last sync metadata was set
322        let db = credit_card_engine.store.lock_db()?;
323        let conn = &db.writer;
324
325        assert_eq!(
326            credit_card_engine.get_meta::<i64>(conn, LAST_SYNC_META_KEY)?,
327            Some(last_sync)
328        );
329
330        Ok(())
331    }
332
333    #[test]
334    fn test_credit_card_engine_get_sync_assoc() -> Result<()> {
335        ensure_initialized();
336        let credit_card_engine = create_engine();
337
338        let result = credit_card_engine.get_sync_assoc();
339        assert!(result.is_ok());
340
341        // check that we disconnect if sync IDs not found
342        assert_eq!(result.unwrap(), EngineSyncAssociation::Disconnected);
343
344        // create sync metadata
345        let global_guid = Guid::new("AAAA");
346        let coll_guid = Guid::new("AAAA");
347        let ids = CollSyncIds {
348            global: global_guid,
349            coll: coll_guid,
350        };
351        {
352            let db = credit_card_engine.store.lock_db()?;
353            let conn = &db.writer;
354            credit_card_engine.put_meta(conn, GLOBAL_SYNCID_META_KEY, &ids.global)?;
355            credit_card_engine.put_meta(conn, COLLECTION_SYNCID_META_KEY, &ids.coll)?;
356        }
357
358        let result = credit_card_engine.get_sync_assoc();
359        assert!(result.is_ok());
360
361        // check that we return the metadata
362        assert_eq!(result.unwrap(), EngineSyncAssociation::Connected(ids));
363        Ok(())
364    }
365
366    #[test]
367    fn test_engine_sync_reset() -> Result<()> {
368        ensure_initialized();
369        let engine = create_engine();
370        let encdec = EncryptorDecryptor::new_with_random_key().unwrap();
371
372        let cc = InternalCreditCard {
373            guid: Guid::random(),
374            cc_name: "Ms Jane Doe".to_string(),
375            cc_number_enc: encdec.encrypt("12341232412341234")?,
376            cc_number_last_4: "1234".to_string(),
377            cc_exp_month: 12,
378            cc_exp_year: 2021,
379            cc_type: "visa".to_string(),
380            ..Default::default()
381        };
382
383        {
384            // temp scope for the mutex lock.
385            let db = engine.store.lock_db()?;
386            let tx = db.writer.unchecked_transaction()?;
387            // create a normal record, a mirror record and a tombstone.
388            add_internal_credit_card(&tx, &cc)?;
389            test_insert_mirror_record(
390                &tx,
391                cc.clone()
392                    .into_test_incoming_bso(&encdec, Default::default()),
393            );
394            insert_tombstone_record(&tx, Guid::random().to_string())?;
395            tx.commit()?;
396        }
397
398        // create sync metadata
399        let global_guid = Guid::new("AAAA");
400        let coll_guid = Guid::new("AAAA");
401        let ids = CollSyncIds {
402            global: global_guid.clone(),
403            coll: coll_guid.clone(),
404        };
405        {
406            let db = engine.store.lock_db()?;
407            let conn = &db.writer;
408            engine.put_meta(conn, GLOBAL_SYNCID_META_KEY, &ids.global)?;
409            engine.put_meta(conn, COLLECTION_SYNCID_META_KEY, &ids.coll)?;
410        }
411
412        // call reset for sign out
413        engine
414            .reset(&EngineSyncAssociation::Disconnected)
415            .expect("should work");
416
417        {
418            let db = engine.store.lock_db()?;
419            let conn = &db.writer;
420
421            // check that the mirror and tombstone tables have no records
422            assert!(get_all(conn, "credit_cards_mirror".to_string())?.is_empty());
423            assert!(get_all(conn, "credit_cards_tombstones".to_string())?.is_empty());
424
425            // check that the last sync time was reset to 0
426            let expected_sync_time = 0;
427            assert_eq!(
428                engine
429                    .get_meta::<i64>(conn, LAST_SYNC_META_KEY)?
430                    .unwrap_or(1),
431                expected_sync_time
432            );
433
434            // check that the meta records were deleted
435            assert!(engine
436                .get_meta::<String>(conn, GLOBAL_SYNCID_META_KEY)?
437                .is_none());
438            assert!(engine
439                .get_meta::<String>(conn, COLLECTION_SYNCID_META_KEY)?
440                .is_none());
441
442            clear_cc_tables(conn)?;
443
444            // re-populating the tables
445            let tx = conn.unchecked_transaction()?;
446            add_internal_credit_card(&tx, &cc)?;
447            test_insert_mirror_record(&tx, cc.into_test_incoming_bso(&encdec, Default::default()));
448            insert_tombstone_record(&tx, Guid::random().to_string())?;
449            tx.commit()?;
450        }
451
452        // call reset for sign in
453        engine
454            .reset(&EngineSyncAssociation::Connected(ids))
455            .expect("should work");
456
457        let db = engine.store.lock_db()?;
458        let conn = &db.writer;
459        // check that the meta records were set
460        let retrieved_global_sync_id = engine.get_meta::<String>(conn, GLOBAL_SYNCID_META_KEY)?;
461        assert_eq!(
462            retrieved_global_sync_id.unwrap_or_default(),
463            global_guid.to_string()
464        );
465
466        let retrieved_coll_sync_id = engine.get_meta::<String>(conn, COLLECTION_SYNCID_META_KEY)?;
467        assert_eq!(
468            retrieved_coll_sync_id.unwrap_or_default(),
469            coll_guid.to_string()
470        );
471        Ok(())
472    }
473}
autofill/sync/engine.rs

autofill/sync/
engine.rs
