Live Consent Sync (#1234)

* update proto * process streamed updates * first stream update location * use local events to dispatch update * optional buffer on the local events * cleanup * rename * cleanup * add warn * lint * logs * wip * cleanup * remove comment * syncable groups fix * revert this file * invert the approach * add down * wip * wip tests * ignore non sync msgs * cleanup, test fix * lint * wasm * wasm * cleanup * map an error * lint * lint * small fix * test * revert * do not drop channel * lint * alternate remedy * wasm fix * revert gen_protos * fix potential race condition in test
xmtp · Nov 18, 2024 · 0b0bcde · 0b0bcde
1 parent e9489ae
commit 0b0bcde
Show file tree

Hide file tree

Showing 12 changed files with 1,984 additions and 1,350 deletions.
diff --git a/xmtp_id/src/associations/serialization.rs b/xmtp_id/src/associations/serialization.rs
@@ -70,6 +70,8 @@ pub enum DeserializationError {
     InvalidAccountId,
     #[error("Invalid hash (needs to be 32 bytes)")]
     InvalidHash,
+    #[error("A required field is unspecified: {0}")]
+    Unspecified(&'static str),
     #[error("Error creating public key from proto bytes")]
     Ed25519(#[from] ed25519_dalek::ed25519::Error),
 }

diff --git a/xmtp_mls/src/client.rs b/xmtp_mls/src/client.rs
@@ -56,7 +56,7 @@ use crate::{
         refresh_state::EntityKind,
         sql_key_store, EncryptedMessageStore, StorageError,
     },
-    subscriptions::LocalEvents,
+    subscriptions::{LocalEventError, LocalEvents},
     verified_key_package_v2::{KeyPackageVerificationError, VerifiedKeyPackageV2},
     xmtp_openmls_provider::XmtpOpenMlsProvider,
     Fetch, Store, XmtpApi,
@@ -106,6 +106,8 @@ pub enum ClientError {
     // the box is to prevent infinite cycle between client and group errors
     #[error(transparent)]
     Group(Box<GroupError>),
+    #[error(transparent)]
+    LocalEvent(#[from] LocalEventError),
     #[error("generic:{0}")]
     Generic(String),
 }
@@ -188,6 +190,10 @@ pub enum MessageProcessingError {
     Generic(String),
     #[error("intent is missing staged_commit field")]
     IntentMissingStagedCommit,
+    #[error(transparent)]
+    Deserialization(#[from] xmtp_id::associations::DeserializationError),
+    #[error(transparent)]
+    LocalEvent(#[from] LocalEventError),
 }
 
 impl crate::retry::RetryableError for MessageProcessingError {
@@ -501,6 +507,14 @@ where
         conn.insert_or_replace_consent_records(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);
+            self.local_events
+                .send(LocalEvents::ConsentUpdate(records))
+                .map_err(|e| ClientError::Generic(e.to_string()))?;
+        }
+
         Ok(())
     }