Only broadcast consent changes (#1485)

* return changed records

* only broadcast changes

* lint

* one more assertion

xmtp_mls/src/client.rs

@@ -444,13 +444,11 @@ where
             }
         }
 
-        conn.insert_or_replace_consent_records(records)?;
-        conn.insert_or_replace_consent_records(&new_records)?;
+        new_records.extend_from_slice(records);
+        let changed_records = conn.insert_or_replace_consent_records(&new_records)?;
 
-        if self.history_sync_url.is_some() {
-            let mut records = records.to_vec();
-            records.append(&mut new_records);
-            let records = records
+        if self.history_sync_url.is_some() && !changed_records.is_empty() {
+            let records = changed_records
                 .into_iter()
                 .map(UserPreferenceUpdate::ConsentUpdate)
                 .collect();

xmtp_mls/src/storage/encrypted_store/consent_record.rs

@@ -57,12 +57,34 @@ impl DbConnection {
         })?)
     }
 
-    /// Insert consent_records, and replace existing entries
+    /// Insert consent_records, and replace existing entries, returns records that are new or changed
     pub fn insert_or_replace_consent_records(
         &self,
         records: &[StoredConsentRecord],
-    ) -> Result<(), StorageError> {
-        self.raw_query(|conn| -> diesel::QueryResult<_> {
+    ) -> Result<Vec<StoredConsentRecord>, StorageError> {
+        let mut query = consent_records::table
+            .into_boxed()
+            .filter(false.into_sql::<diesel::sql_types::Bool>());
+        let primary_keys: Vec<_> = records
+            .iter()
+            .map(|r| (&r.entity, &r.entity_type))
+            .collect();
+        for (entity, entity_type) in primary_keys {
+            query = query.or_filter(
+                consent_records::entity_type
+                    .eq(entity_type)
+                    .and(consent_records::entity.eq(entity)),
+            );
+        }
+
+        let changed = self.raw_query(|conn| -> diesel::QueryResult<_> {
+            let existing: Vec<StoredConsentRecord> = query.load(conn)?;
+            let changed: Vec<_> = records
+                .iter()
+                .filter(|r| !existing.contains(r))
+                .cloned()
+                .collect();
+
             conn.transaction::<_, diesel::result::Error, _>(|conn| {
                 for record in records.iter() {
                     diesel::insert_into(dsl::consent_records)
@@ -73,10 +95,12 @@ impl DbConnection {
                         .execute(conn)?;
                 }
                 Ok(())
-            })
+            })?;
+
+            Ok(changed)
         })?;
 
-        Ok(())
+        Ok(changed)
     }
 
     pub fn maybe_insert_consent_record_return_existing(
@@ -205,13 +229,34 @@ mod tests {
             let inbox_id = "inbox_1";
             let consent_record = generate_consent_record(
                 ConsentType::InboxId,
-                ConsentState::Denied,
+                ConsentState::Allowed,
                 inbox_id.to_string(),
             );
             let consent_record_entity = consent_record.entity.clone();
 
-            conn.insert_or_replace_consent_records(&[consent_record])
+            // Insert the record
+            let result = conn
+                .insert_or_replace_consent_records(&[consent_record.clone()])
+                .expect("should store without error");
+            // One record was inserted
+            assert_eq!(result.len(), 1);
+
+            // Insert it again
+            let result = conn
+                .insert_or_replace_consent_records(&[consent_record.clone()])
+                .expect("should store without error");
+            // Nothing should change
+            assert_eq!(result.len(), 0);
+
+            // Insert it again, this time with a Denied state
+            let result = conn
+                .insert_or_replace_consent_records(&[StoredConsentRecord {
+                    state: ConsentState::Denied,
+                    ..consent_record
+                }])
                 .expect("should store without error");
+            // Should change
+            assert_eq!(result.len(), 1);
 
             let consent_record = conn
                 .get_consent_record(inbox_id.to_owned(), ConsentType::InboxId)