Add WakeupService

core/conversations/src/conversation.rs

@@ -26,6 +26,8 @@ pub(crate) trait Convo<S: ExternalServices> {
         cx: &mut ServiceContext<S>,
         enc: EncryptedPayload,
     ) -> Result<ConvoOutcome, ChatError>;
+
+    fn wakeup(&mut self, service_ctx: &mut ServiceContext<S>) -> Result<(), ChatError>;
 }
 
 /// Group-only operations.

core/conversations/src/conversation/group_v1.rs

@@ -276,6 +276,10 @@ impl<S: ExternalServices> Convo<S> for GroupV1Convo {
             content,
         })
     }
+
+    fn wakeup(&mut self, _: &mut ServiceContext<S>) -> Result<(), ChatError> {
+        Ok(())
+    }
 }
 
 impl<S: ExternalServices> GroupConvo<S> for GroupV1Convo {

core/conversations/src/conversation/privatev1.rs

@@ -275,6 +275,10 @@ impl<S: ExternalServices> Convo<S> for PrivateV1Convo {
             content,
         })
     }
+
+    fn wakeup(&mut self, _: &mut ServiceContext<S>) -> Result<(), ChatError> {
+        Ok(())
+    }
 }
 
 impl Debug for PrivateV1Convo {

core/conversations/src/core.rs

@@ -1,6 +1,6 @@
 use crate::causal_history::{CausalHistoryStore, MissingMessage};
 use crate::service_context::{ExternalServices, ServiceContext};
-use crate::{DeliveryService, IdentityProvider, RegistrationService};
+use crate::{DeliveryService, IdentityProvider, RegistrationService, WakeupService};
 use crate::{
     conversation::{Convo, GroupConvo, GroupV1Convo, PrivateV1Convo},
     errors::ChatError,
@@ -31,11 +31,12 @@ pub struct Core<S: ExternalServices> {
 
 // Constructors live on the `(DS, RS, CS)` form: `S` can't be inferred backwards
 // through `S::DS`, so the bundle is built from the three args here.
-impl<IP, DS, RS, CS> Core<(IP, DS, RS, CS)>
+impl<IP, DS, RS, WS, CS> Core<(IP, DS, RS, WS, CS)>
 where
     IP: IdentityProvider + 'static,
     DS: DeliveryService + 'static,
     RS: RegistrationService + 'static,
+    WS: WakeupService + 'static,
     CS: ChatStore + 'static,
 {
     /// Opens or creates a `Core` with the given storage configuration.
@@ -46,6 +47,7 @@ where
         ident: IP,
         delivery: DS,
         registration: RS,
+        wakeup_service: WS,
         mut store: CS,
     ) -> Result<Self, ChatError> {
         let identity = if let Some(identity) = store.load_identity()? {
@@ -56,7 +58,14 @@ where
             identity
         };
 
-        Self::assemble(ident, identity, delivery, registration, store)
+        Self::assemble(
+            ident,
+            identity,
+            delivery,
+            registration,
+            wakeup_service,
+            store,
+        )
     }
 
     /// Creates a new in-memory `Core` (for testing).
@@ -66,10 +75,18 @@ where
         ident: IP,
         delivery: DS,
         registration: RS,
+        wakeup_service: WS,
         store: CS,
     ) -> Result<Self, ChatError> {
         let identity = Identity::new(ident.id().as_str().to_string());
-        let mut core = Self::assemble(ident, identity, delivery, registration, store)?;
+        let mut core = Self::assemble(
+            ident,
+            identity,
+            delivery,
+            registration,
+            wakeup_service,
+            store,
+        )?;
 
         core.register_keypackage()?;
         core.register_account_bundle()?;
@@ -83,6 +100,7 @@ where
         identity: Identity,
         mut delivery: DS,
         registration: RS,
+        wakeup_service: WS,
         store: CS,
     ) -> Result<Self, ChatError> {
         let inbox = Inbox::new(&identity);
@@ -109,6 +127,7 @@ where
                 mls_provider,
                 causal,
                 identity,
+                wakeup_service,
             },
             inbox,
             pq_inbox,
@@ -263,6 +282,28 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
         let enc_payload = EncryptedPayload::decode(enc_payload_bytes)?;
         let mut convo = self.load_convo(convo_id)?;
         convo.handle_frame(&mut self.services, enc_payload)
+    pub fn wakeup(&mut self, convo_id: ConversationIdRef) -> Result<(), ChatError> {
+        info!(convos = ?self.cached_convos.keys().collect::<Vec<_>>(), id = ?self.services.mls_identity.id(), "Cached Convos");
+
+        match convo_id {
+            c if c == self.pq_inbox.id() => todo!(),
+            c if self.cached_convos.contains_key(c) => self.wakeup_convo(c),
+            _ => Ok(()),
+        }
+    }
+
+    // Dispatch encrypted payload to its corresponding conversation
+    fn wakeup_convo(&mut self, convo_id: ConversationIdRef) -> Result<(), ChatError> {
+        let Some(convo) = self.cached_convos.get_mut(convo_id) else {
+            return Err(ChatError::generic("No Convo Found"));
+        };
+        let convo = match convo {
+            ConvoTypeOwned::Group(c) => c.as_mut(),
+        };
+
+        convo.wakeup(&mut self.services)
+    }
+
     }
 
     /// Rebuilds a conversation from storage — the one site that branches on

core/conversations/src/lib.rs

@@ -27,8 +27,8 @@ pub use outcomes::{
     Content, ConversationClass, ConvoOutcome, InboxOutcome, NewConversation, PayloadOutcome,
 };
 pub use service_context::ExternalServices;
-pub use service_traits::{DeliveryService, RegistrationService};
 pub use shared_traits::IdentityProvider;
+pub use service_traits::{DeliveryService, RegistrationService, WakeupService};
 pub use storage::ConversationKind;
 pub use types::AddressedEnvelope;
 pub use utils::hex_trunc;

core/conversations/src/service_context.rs

@@ -6,6 +6,7 @@ use storage::ChatStore;
 use crate::IdentityProvider;
 use crate::causal_history::CausalHistoryStore;
 use crate::inbox_v2::{MlsEphemeralPqProvider, MlsIdentityProvider};
+use crate::service_traits::WakeupService;
 use crate::{DeliveryService, RegistrationService};
 
 /// Bundles the external service types (`DS`, `RS`, `CS`) behind one `S`. The
@@ -14,19 +15,22 @@ pub trait ExternalServices {
     type IP: IdentityProvider;
     type DS: DeliveryService;
     type RS: RegistrationService;
+    type WS: WakeupService;
     type CS: ChatStore;
 }
 
-impl<IP, DS, RS, CS> ExternalServices for (IP, DS, RS, CS)
+impl<IP, DS, RS, WS, CS> ExternalServices for (IP, DS, RS, WS, CS)
 where
     IP: IdentityProvider,
     DS: DeliveryService,
     RS: RegistrationService,
+    WS: WakeupService,
     CS: ChatStore,
 {
     type IP = IP;
     type DS = DS;
     type RS = RS;
+    type WS = WS;
     type CS = CS;
 }
 
@@ -39,6 +43,7 @@ pub(crate) struct ServiceContext<S: ExternalServices> {
     pub(crate) mls_provider: MlsEphemeralPqProvider,
     pub(crate) causal: CausalHistoryStore,
     pub(crate) identity: Identity,
+    pub(crate) wakeup_service: S::WS,
 }
 
 #[cfg(test)]
@@ -106,7 +111,16 @@ mod test_support {
         }
     }
 
-    impl<IP: IdentityProvider, CS: ChatStore> ServiceContext<(IP, NoopDelivery, NoopRegistration, CS)> {
+    #[derive(Debug)]
+    pub(crate) struct NoopWakeups;
+
+    impl WakeupService for NoopWakeups {
+        fn wakeup_in(&mut self, duration: std::time::Duration, convo_id: crate::ConversationId) {}
+    }
+
+    impl<IP: IdentityProvider, CS: ChatStore>
+        ServiceContext<(IP, NoopDelivery, NoopRegistration, WS, CS)>
+    {
         /// Builds a context around a real store, stubbing other services.
         pub(crate) fn for_test(ident: IP, store: CS) -> Result<Self, ChatError> {
             let name = ident.id().as_str().to_string();
@@ -118,6 +132,7 @@ mod test_support {
                 mls_provider: MlsEphemeralPqProvider::new().map_err(ChatError::generic)?,
                 causal: CausalHistoryStore::new(),
                 identity: Identity::new(name),
+                wakeup_service: NoopWakeup {},
             })
         }
     }

core/conversations/src/service_traits.rs

@@ -2,9 +2,12 @@
 /// platform clients. Platforms can alter the behaviour of the chat core by supplying
 /// different implementations.
 use shared_traits::IdentityProvider;
-use std::{fmt::Debug, fmt::Display};
+use std::{
+    fmt::{Debug, Display},
+    time::Duration,
+};
 
-use crate::{AccountDirectory, types::AddressedEnvelope};
+use crate::{AccountDirectory, ConversationId, types::AddressedEnvelope};
 
 /// A Delivery service is responsible for payload transport.
 /// This interface allows Conversations to send payloads on the wire as well as
@@ -62,3 +65,7 @@ impl<T: RegistrationService> KeyPackageProvider for T {
         RegistrationService::retrieve(self, device_id)
     }
 }
+
+pub trait WakeupService: Debug {
+    fn wakeup_in(&mut self, duration: Duration, convo_id: ConversationId);
+}

core/integration_tests_core/tests/causal_history.rs

@@ -7,13 +7,21 @@
 use std::ops::{Deref, DerefMut};
 
 use components::{EphemeralRegistry, LocalBroadcaster, MemStore};
-use libchat::{Core, MissingMessage};
+use libchat::{Core, MissingMessage, WakeupService};
 use logos_account::TestLogosAccount;
+
+#[derive(Debug)]
+struct NoopWakeupService {}
+impl WakeupService for NoopWakeupService {
+    fn wakeup_in(&mut self, _: std::time::Duration, _: libchat::ConversationId) {}
+}
+
 struct Client {
     inner: Core<(
         TestLogosAccount,
         LocalBroadcaster,
         EphemeralRegistry,
+        NoopWakeupService,
         MemStore,
     )>,
 }
@@ -24,6 +32,7 @@ impl Client {
             TestLogosAccount,
             LocalBroadcaster,
             EphemeralRegistry,
+            NoopWakeupService,
             MemStore,
         )>,
     ) -> Self {
@@ -54,6 +63,7 @@ impl Deref for Client {
         TestLogosAccount,
         LocalBroadcaster,
         EphemeralRegistry,
+        NoopWakeupService,
         MemStore,
     )>;
     fn deref(&self) -> &Self::Target {
@@ -73,12 +83,24 @@ fn missing_group_message_is_detected() {
     let rs = EphemeralRegistry::new();
 
     let saro_account = TestLogosAccount::new("saro");
-    let saro_ctx =
-        Core::new_with_name(saro_account, ds.new_consumer(), rs.clone(), MemStore::new()).unwrap();
+    let saro_ctx = Core::new_with_name(
+        saro_account,
+        ds.new_consumer(),
+        rs.clone(),
+        NoopWakeupService {},
+        MemStore::new(),
+    )
+    .unwrap();
 
     let raya_account = TestLogosAccount::new("raya");
-    let raya_ctx =
-        Core::new_with_name(raya_account, ds.clone(), rs.clone(), MemStore::new()).unwrap();
+    let raya_ctx = Core::new_with_name(
+        raya_account,
+        ds.clone(),
+        rs.clone(),
+        NoopWakeupService {},
+        MemStore::new(),
+    )
+    .unwrap();
 
     let mut saro = Client::init(saro_ctx);
     let mut raya = Client::init(raya_ctx);

core/integration_tests_core/tests/private_integration.rs

@@ -1,15 +1,22 @@
 use chat_sqlite::{ChatStorage, StorageConfig};
-use libchat::{ConversationClass, Core, Introduction, PayloadOutcome};
+use libchat::{ConversationClass, Core, Introduction, PayloadOutcome, WakeupService};
 use logos_account::TestLogosAccount;
 use storage::{ConversationStore, IdentityStore};
 use tempfile::tempdir;
 
 use components::{EphemeralRegistry, LocalBroadcaster};
 
+#[derive(Debug)]
+struct NoopWakeupService {}
+impl WakeupService for NoopWakeupService {
+    fn wakeup_in(&mut self, _: std::time::Duration, _: libchat::ConversationId) {}
+}
+
 type PrivateCore = Core<(
     TestLogosAccount,
     LocalBroadcaster,
     EphemeralRegistry,
+    NoopWakeupService,
     ChatStorage,
 )>;
 
@@ -60,11 +67,19 @@ fn ctx_integration() {
         saro_account,
         ds.clone(),
         rs.clone(),
+        NoopWakeupService {},
         ChatStorage::in_memory(),
     )
     .unwrap();
     let raya_account = TestLogosAccount::new("raya");
-    let mut raya = Core::new_with_name(raya_account, ds, rs, ChatStorage::in_memory()).unwrap();
+    let mut raya = Core::new_with_name(
+        raya_account,
+        ds,
+        rs,
+        NoopWakeupService {},
+        ChatStorage::in_memory(),
+    )
+    .unwrap();
 
     // Raya creates intro bundle and sends to Saro
     let bundle = raya.create_intro_bundle().unwrap();
@@ -107,7 +122,7 @@ fn identity_persistence() {
     let rs = EphemeralRegistry::new();
     let store1 = ChatStorage::new(StorageConfig::InMemory).unwrap();
     let alice_account = TestLogosAccount::new("alice");
-    let ctx1 = Core::new_with_name(alice_account, ds, rs, store1).unwrap();
+    let ctx1 = Core::new_with_name(alice_account, ds, rs, NoopWakeupService {}, store1).unwrap();
     let pubkey1 = ctx1.identity().public_key();
     let name1 = ctx1.installation_name().to_string();
 
@@ -127,7 +142,7 @@ fn open_persists_new_identity() {
     let rs = EphemeralRegistry::new();
     let store = ChatStorage::new(StorageConfig::File(db_path.clone())).unwrap();
     let alice_account = TestLogosAccount::new("alice");
-    let core = Core::new_from_store(alice_account, ds, rs, store).unwrap();
+    let core = Core::new_from_store(alice_account, ds, rs, NoopWakeupService {}, store).unwrap();
     let pubkey = core.identity().public_key();
     drop(core);
 
@@ -147,11 +162,19 @@ fn conversation_metadata_persistence() {
         alice_account,
         ds.clone(),
         rs.clone(),
+        NoopWakeupService {},
         ChatStorage::in_memory(),
     )
     .unwrap();
     let bob_account = TestLogosAccount::new("bob");
-    let mut bob = Core::new_with_name(bob_account, ds, rs, ChatStorage::in_memory()).unwrap();
+    let mut bob = Core::new_with_name(
+        bob_account,
+        ds,
+        rs,
+        NoopWakeupService {},
+        ChatStorage::in_memory(),
+    )
+    .unwrap();
 
     let bundle = alice.create_intro_bundle().unwrap();
     let intro = Introduction::try_from(bundle.as_slice()).unwrap();
@@ -180,11 +203,19 @@ fn conversation_full_flow() {
         alice_account,
         ds.clone(),
         rs.clone(),
+        NoopWakeupService {},
         ChatStorage::in_memory(),
     )
     .unwrap();
     let bob_account = TestLogosAccount::new("bob");
-    let mut bob = Core::new_with_name(bob_account, ds, rs, ChatStorage::in_memory()).unwrap();
+    let mut bob = Core::new_with_name(
+        bob_account,
+        ds,
+        rs,
+        NoopWakeupService {},
+        ChatStorage::in_memory(),
+    )
+    .unwrap();
 
     let bundle = alice.create_intro_bundle().unwrap();
     let intro = Introduction::try_from(bundle.as_slice()).unwrap();