webext_storage/sync/
bridge.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 anyhow::Result;
6use rusqlite::Transaction;
7use std::sync::{Arc, Weak};
8use sync15::bso::{IncomingBso, OutgoingBso};
9use sync15::engine::{CollSyncIds, CollectionRequest, EngineSyncAssociation, SyncEngine};
10use sync15::{telemetry, CollectionName, ServerTimestamp};
11use sync_guid::Guid as SyncGuid;
12
13// The collection name Desktop's Sync framework uses for `storage.sync`. Only
14// used for telemetry labelling here (Desktop builds the collection URL itself).
15const COLLECTION_NAME: &str = "extension-storage";
16
17use crate::db::{delete_meta, get_meta, put_meta, ThreadSafeStorageDb};
18use crate::schema;
19use crate::sync::incoming::{apply_actions, get_incoming, plan_incoming, stage_incoming};
20use crate::sync::outgoing::{get_outgoing, record_uploaded, stage_outgoing};
21use crate::WebExtStorageStore;
22
23const LAST_SYNC_META_KEY: &str = "last_sync_time";
24const SYNC_ID_META_KEY: &str = "sync_id";
25
26impl WebExtStorageStore {
27    // Returns a bridged sync engine for this store.
28    pub fn bridged_engine(self: Arc<Self>) -> Arc<WebExtStorageBridgedEngine> {
29        let engine = Box::new(WebExtSyncEngine::new(&self.db));
30        Arc::new(WebExtStorageBridgedEngine::new(engine))
31    }
32}
33
34pub struct WebExtSyncEngine {
35    db: Weak<ThreadSafeStorageDb>,
36}
37
38impl WebExtSyncEngine {
39    /// Creates a bridged engine for syncing.
40    pub fn new(db: &Arc<ThreadSafeStorageDb>) -> Self {
41        WebExtSyncEngine {
42            db: Arc::downgrade(db),
43        }
44    }
45
46    fn do_reset(&self, tx: &Transaction<'_>) -> Result<()> {
47        tx.execute_batch(
48            "DELETE FROM storage_sync_mirror;
49             UPDATE storage_sync_data SET sync_change_counter = 1;",
50        )?;
51        delete_meta(tx, LAST_SYNC_META_KEY)?;
52        Ok(())
53    }
54
55    fn thread_safe_storage_db(&self) -> Result<Arc<ThreadSafeStorageDb>> {
56        self.db
57            .upgrade()
58            .ok_or_else(|| crate::error::Error::DatabaseConnectionClosed.into())
59    }
60}
61
62impl SyncEngine for WebExtSyncEngine {
63    fn collection_name(&self) -> CollectionName {
64        COLLECTION_NAME.into()
65    }
66
67    // Read-only view of the engine-owned last-sync time, for the Desktop bridge.
68    // It's written only internally, in `apply`/`set_uploaded`.
69    fn last_sync(&self) -> Result<Option<ServerTimestamp>> {
70        let shared_db = self.thread_safe_storage_db()?;
71        let db = shared_db.lock();
72        let conn = db.get_connection()?;
73        Ok(get_meta::<i64>(conn, LAST_SYNC_META_KEY)?.map(ServerTimestamp))
74    }
75
76    fn reset_last_sync(&self) -> Result<()> {
77        let shared_db = self.thread_safe_storage_db()?;
78        let db = shared_db.lock();
79        let conn = db.get_connection()?;
80        let tx = conn.unchecked_transaction()?;
81        delete_meta(&tx, LAST_SYNC_META_KEY)?;
82        tx.commit()?;
83        Ok(())
84    }
85
86    fn get_sync_assoc(&self) -> Result<EngineSyncAssociation> {
87        let shared_db = self.thread_safe_storage_db()?;
88        let db = shared_db.lock();
89        let conn = db.get_connection()?;
90        // Bridged engines never maintain the "global" guid - that's all managed
91        // by the consumer (Desktop); they only care about the per-collection one.
92        Ok(match get_meta::<String>(conn, SYNC_ID_META_KEY)? {
93            Some(coll) => EngineSyncAssociation::Connected(CollSyncIds {
94                global: SyncGuid::empty(),
95                coll: coll.into(),
96            }),
97            None => EngineSyncAssociation::Disconnected,
98        })
99    }
100
101    fn sync_started(&self) -> Result<()> {
102        let shared_db = self.thread_safe_storage_db()?;
103        let db = shared_db.lock();
104        let conn = db.get_connection()?;
105        schema::create_empty_sync_temp_tables(conn)?;
106        Ok(())
107    }
108
109    fn stage_incoming(
110        &self,
111        incoming_bsos: Vec<IncomingBso>,
112        _telem: &mut telemetry::Engine,
113    ) -> Result<()> {
114        let shared_db = self.thread_safe_storage_db()?;
115        let db = shared_db.lock();
116        let signal = db.begin_interrupt_scope()?;
117        let conn = db.get_connection()?;
118        let tx = conn.unchecked_transaction()?;
119        let incoming_content: Vec<_> = incoming_bsos
120            .into_iter()
121            .map(IncomingBso::into_content::<super::WebextRecord>)
122            .collect();
123        stage_incoming(&tx, &incoming_content, &signal)?;
124        tx.commit()?;
125        Ok(())
126    }
127
128    fn apply(
129        &self,
130        timestamp: ServerTimestamp,
131        _telem: &mut telemetry::Engine,
132    ) -> Result<Vec<OutgoingBso>> {
133        let shared_db = self.thread_safe_storage_db()?;
134        let db = shared_db.lock();
135        let signal = db.begin_interrupt_scope()?;
136        let conn = db.get_connection()?;
137        let tx = conn.unchecked_transaction()?;
138        let incoming = get_incoming(&tx)?;
139        let actions = incoming
140            .into_iter()
141            .map(|(item, state)| (item, plan_incoming(state)))
142            .collect();
143        apply_actions(&tx, actions, &signal)?;
144        stage_outgoing(&tx)?;
145        // The engine owns its last-sync time: record the collection timestamp we
146        // just synced to, so it advances without any external `set_last_sync`.
147        // (Timestamp is zero only in an upload-only path, which must not move it.)
148        if timestamp != ServerTimestamp(0) {
149            put_meta(&tx, LAST_SYNC_META_KEY, &timestamp.as_millis())?;
150        }
151        tx.commit()?;
152
153        Ok(get_outgoing(conn, &signal)?)
154    }
155
156    fn set_uploaded(&self, new_timestamp: ServerTimestamp, ids: Vec<SyncGuid>) -> Result<()> {
157        let shared_db = self.thread_safe_storage_db()?;
158        let db = shared_db.lock();
159        let conn = db.get_connection()?;
160        let signal = db.begin_interrupt_scope()?;
161        let tx = conn.unchecked_transaction()?;
162        record_uploaded(&tx, &ids, &signal)?;
163        // Advance the engine-owned last-sync time to the post-upload timestamp.
164        if new_timestamp != ServerTimestamp(0) {
165            put_meta(&tx, LAST_SYNC_META_KEY, &new_timestamp.as_millis())?;
166        }
167        tx.commit()?;
168
169        Ok(())
170    }
171
172    fn sync_finished(&self) -> Result<()> {
173        let shared_db = self.thread_safe_storage_db()?;
174        let db = shared_db.lock();
175        let conn = db.get_connection()?;
176        schema::create_empty_sync_temp_tables(conn)?;
177        Ok(())
178    }
179
180    fn get_collection_request(
181        &self,
182        server_timestamp: ServerTimestamp,
183    ) -> Result<Option<CollectionRequest>> {
184        let shared_db = self.thread_safe_storage_db()?;
185        let db = shared_db.lock();
186        let conn = db.get_connection()?;
187        let since = ServerTimestamp(get_meta::<i64>(conn, LAST_SYNC_META_KEY)?.unwrap_or(0));
188        Ok(if since == server_timestamp {
189            None
190        } else {
191            Some(
192                CollectionRequest::new(COLLECTION_NAME.into())
193                    .full()
194                    .newer_than(since),
195            )
196        })
197    }
198
199    fn reset(&self, assoc: &EngineSyncAssociation) -> Result<()> {
200        let shared_db = self.thread_safe_storage_db()?;
201        let db = shared_db.lock();
202        let conn = db.get_connection()?;
203        let tx = conn.unchecked_transaction()?;
204        self.do_reset(&tx)?;
205        // A `Disconnected` reset clears the sync ID; a `Connected` one adopts the
206        // (per-collection) ID. `do_reset` already cleared the last sync time.
207        match assoc {
208            EngineSyncAssociation::Disconnected => {
209                delete_meta(&tx, SYNC_ID_META_KEY)?;
210            }
211            EngineSyncAssociation::Connected(ids) => {
212                put_meta(&tx, SYNC_ID_META_KEY, &ids.coll.to_string())?;
213            }
214        }
215        tx.commit()?;
216        Ok(())
217    }
218
219    fn wipe(&self) -> Result<()> {
220        let shared_db = self.thread_safe_storage_db()?;
221        let db = shared_db.lock();
222        let conn = db.get_connection()?;
223        let tx = conn.unchecked_transaction()?;
224        // We assume the meta table is only used by sync.
225        tx.execute_batch(
226            "DELETE FROM storage_sync_data; DELETE FROM storage_sync_mirror; DELETE FROM meta;",
227        )?;
228        tx.commit()?;
229        Ok(())
230    }
231}
232
233// The UniFFI-exposed `WebExtStorageBridgedEngine` (a thin newtype around
234// `sync15::engine::BridgedEngineWrapper`) is generated by this macro, which
235// removes the facade + BSO marshalling boilerplate. The wrapper drives the
236// `SyncEngine` impl on the `BridgedEngine` defined above (webext-storage is
237// Desktop-only, but implements the one unified `SyncEngine` trait like everyone
238// else).
239sync15::uniffi_bridged_engine!(WebExtStorageBridgedEngine);
240
241impl From<anyhow::Error> for crate::error::Error {
242    fn from(value: anyhow::Error) -> Self {
243        crate::error::Error::SyncError(value.to_string())
244    }
245}
246
247#[cfg(test)]
248mod tests {
249    use super::*;
250    use crate::db::test::new_mem_thread_safe_storage_db;
251    use crate::db::StorageDb;
252    use sync15::engine::BridgedEngineWrapper;
253
254    // The sync-ID and reset semantics that used to live on the old
255    // `BridgedEngine` trait now live on `BridgedEngineWrapper` (which drives our
256    // `SyncEngine`), so we exercise them the same way Desktop does - through the
257    // wrapper. Each engine holds a `Weak` to the shared db, so callers keep the
258    // strong `Arc` alive and inspect DB state through it directly.
259    fn wrapper(db: &Arc<ThreadSafeStorageDb>) -> BridgedEngineWrapper {
260        BridgedEngineWrapper::new(Box::new(WebExtSyncEngine::new(db)))
261    }
262
263    fn query_count(db: &StorageDb, table: &str) -> u32 {
264        let conn = db.get_connection().expect("should retrieve connection");
265        conn.query_row_and_then(&format!("SELECT COUNT(*) FROM {};", table), [], |row| {
266            row.get::<_, u32>(0)
267        })
268        .expect("should work")
269    }
270
271    // Sets up mock data for the tests here.
272    fn setup_mock_data(db: &Arc<ThreadSafeStorageDb>) -> Result<()> {
273        {
274            let shared = db.lock();
275            let conn = shared.get_connection().expect("should retrieve connection");
276            conn.execute(
277                "INSERT INTO storage_sync_data (ext_id, data, sync_change_counter)
278                    VALUES ('ext-a', 'invalid-json', 2)",
279                [],
280            )?;
281            conn.execute(
282                "INSERT INTO storage_sync_mirror (guid, ext_id, data)
283                    VALUES ('guid', 'ext-a', '3')",
284                [],
285            )?;
286        }
287        // Seed a last-sync time directly - there's no public setter for it.
288        {
289            let shared = db.lock();
290            let conn = shared.get_connection().expect("should retrieve connection");
291            put_meta(conn, LAST_SYNC_META_KEY, &1i64)?;
292        }
293
294        let shared = db.lock();
295        // and assert we wrote what we think we did.
296        assert_eq!(query_count(&shared, "storage_sync_data"), 1);
297        assert_eq!(query_count(&shared, "storage_sync_mirror"), 1);
298        assert_eq!(query_count(&shared, "meta"), 1);
299        Ok(())
300    }
301
302    // Assuming a DB setup with setup_mock_data, assert it was correctly reset.
303    fn assert_reset(db: &Arc<ThreadSafeStorageDb>) -> Result<()> {
304        // A reset never wipes data...
305        let shared = db.lock();
306        let conn = shared.get_connection().expect("should retrieve connection");
307        assert_eq!(query_count(&shared, "storage_sync_data"), 1);
308
309        // But did reset the change counter.
310        let cc = conn.query_row_and_then(
311            "SELECT sync_change_counter FROM storage_sync_data WHERE ext_id = 'ext-a';",
312            [],
313            |row| row.get::<_, u32>(0),
314        )?;
315        assert_eq!(cc, 1);
316        // But did wipe the mirror...
317        assert_eq!(query_count(&shared, "storage_sync_mirror"), 0);
318        // And the last_sync should have been wiped.
319        assert!(get_meta::<i64>(conn, LAST_SYNC_META_KEY)?.is_none());
320        Ok(())
321    }
322
323    // Assuming a DB setup with setup_mock_data, assert it has not been reset.
324    fn assert_not_reset(db: &Arc<ThreadSafeStorageDb>) -> Result<()> {
325        let shared = db.lock();
326        let conn = shared.get_connection().expect("should retrieve connection");
327        assert_eq!(query_count(&shared, "storage_sync_data"), 1);
328        let cc = conn.query_row_and_then(
329            "SELECT sync_change_counter FROM storage_sync_data WHERE ext_id = 'ext-a';",
330            [],
331            |row| row.get::<_, u32>(0),
332        )?;
333        assert_eq!(cc, 2);
334        assert_eq!(query_count(&shared, "storage_sync_mirror"), 1);
335        // And the last_sync should remain.
336        assert!(get_meta::<i64>(conn, LAST_SYNC_META_KEY)?.is_some());
337        Ok(())
338    }
339
340    #[test]
341    fn test_wipe() -> Result<()> {
342        let strong = new_mem_thread_safe_storage_db();
343        setup_mock_data(&strong)?;
344
345        wrapper(&strong).wipe()?;
346
347        let db = strong.lock();
348        assert_eq!(query_count(&db, "storage_sync_data"), 0);
349        assert_eq!(query_count(&db, "storage_sync_mirror"), 0);
350        assert_eq!(query_count(&db, "meta"), 0);
351        Ok(())
352    }
353
354    #[test]
355    fn test_reset() -> Result<()> {
356        let strong = new_mem_thread_safe_storage_db();
357        setup_mock_data(&strong)?;
358        {
359            let db = strong.lock();
360            let conn = db.get_connection()?;
361            put_meta(conn, SYNC_ID_META_KEY, &"sync-id".to_string())?;
362        }
363
364        wrapper(&strong).reset()?;
365        assert_reset(&strong)?;
366
367        {
368            let db = strong.lock();
369            let conn = db.get_connection()?;
370            // Only an explicit reset kills the sync-id, so check that here.
371            assert_eq!(get_meta::<String>(conn, SYNC_ID_META_KEY)?, None);
372        }
373
374        Ok(())
375    }
376
377    #[test]
378    fn test_ensure_missing_sync_id() -> Result<()> {
379        let strong = new_mem_thread_safe_storage_db();
380        setup_mock_data(&strong)?;
381
382        assert_eq!(wrapper(&strong).sync_id()?, None);
383        // We don't have a sync ID - so setting one should reset.
384        wrapper(&strong).ensure_current_sync_id("new-id")?;
385        // should have cause a reset.
386        assert_reset(&strong)?;
387        Ok(())
388    }
389
390    #[test]
391    fn test_ensure_new_sync_id() -> Result<()> {
392        let strong = new_mem_thread_safe_storage_db();
393        setup_mock_data(&strong)?;
394
395        {
396            let db = strong.lock();
397            let conn = db.get_connection()?;
398            put_meta(conn, SYNC_ID_META_KEY, &"old-id".to_string())?;
399        }
400
401        assert_not_reset(&strong)?;
402        assert_eq!(wrapper(&strong).sync_id()?, Some("old-id".to_string()));
403
404        wrapper(&strong).ensure_current_sync_id("new-id")?;
405        // should have cause a reset.
406        assert_reset(&strong)?;
407        // should have the new id.
408        assert_eq!(wrapper(&strong).sync_id()?, Some("new-id".to_string()));
409        Ok(())
410    }
411
412    #[test]
413    fn test_ensure_same_sync_id() -> Result<()> {
414        let strong = new_mem_thread_safe_storage_db();
415        setup_mock_data(&strong)?;
416        assert_not_reset(&strong)?;
417
418        {
419            let db = strong.lock();
420            let conn = db.get_connection()?;
421            put_meta(conn, SYNC_ID_META_KEY, &"sync-id".to_string())?;
422        }
423
424        wrapper(&strong).ensure_current_sync_id("sync-id")?;
425        // should not have reset.
426        assert_not_reset(&strong)?;
427        Ok(())
428    }
429
430    #[test]
431    fn test_reset_sync_id() -> Result<()> {
432        let strong = new_mem_thread_safe_storage_db();
433        setup_mock_data(&strong)?;
434
435        {
436            let db = strong.lock();
437            let conn = db.get_connection()?;
438            put_meta(conn, SYNC_ID_META_KEY, &"sync-id".to_string())?;
439        }
440
441        assert_eq!(wrapper(&strong).sync_id()?, Some("sync-id".to_string()));
442        let new_id = wrapper(&strong).reset_sync_id()?;
443        // should have cause a reset.
444        assert_reset(&strong)?;
445        assert_eq!(wrapper(&strong).sync_id()?, Some(new_id));
446        Ok(())
447    }
448}
webext_storage/sync/bridge.rs

webext_storage/sync/
bridge.rs
