Add ThreadMetadata store

presage-store-sled/src/lib.rs

@@ -5,6 +5,7 @@ use std::{
     time::{SystemTime, UNIX_EPOCH},
 };
 
+use crate::protobuf::ContentProto;
 use async_trait::async_trait;
 use log::{debug, error, trace, warn};
 use presage::libsignal_service::zkgroup::GroupMasterKeyBytes;
@@ -26,14 +27,14 @@ use presage::libsignal_service::{
     Profile, ServiceAddress,
 };
 use presage::store::{ContentExt, ContentsStore, StateStore, Store, Thread};
+use presage::ThreadMetadata;
 use presage::{manager::RegistrationData, proto::verified};
+use presage_store_cipher::StoreCipher;
 use prost::Message;
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use sled::{Batch, IVec};
 
-use crate::protobuf::ContentProto;
-
 mod error;
 mod protobuf;
 
@@ -49,6 +50,7 @@ const SLED_TREE_SIGNED_PRE_KEYS: &str = "signed_pre_keys";
 const SLED_TREE_KYBER_PRE_KEYS: &str = "kyber_pre_keys";
 const SLED_TREE_STATE: &str = "state";
 const SLED_TREE_THREADS_PREFIX: &str = "threads";
+const SLED_TREE_THREADS_METADATA: &str = "threads_metadata";
 const SLED_TREE_PROFILES: &str = "profiles";
 const SLED_TREE_PROFILE_KEYS: &str = "profile_keys";
 
@@ -305,6 +307,13 @@ impl SledStore {
         hasher.update(key.collect::<Vec<_>>());
         format!("{:x}", hasher.finalize())
     }
+
+    fn thread_metadata_key(&self, thread: Thread) -> Vec<u8> {
+        match thread {
+            Thread::Contact(contact) => contact.to_string().into_bytes(),
+            Thread::Group(group) => group.to_vec(),
+        }
+    }
 }
 
 fn migrate(
@@ -385,6 +394,8 @@ impl ProtocolStore for SledStore {}
 impl StateStore for SledStore {
     type StateStoreError = SledStoreError;
 
+    /// State
+
     fn load_registration_data(&self) -> Result<Option<RegistrationData>, SledStoreError> {
         self.get(SLED_TREE_STATE, SLED_KEY_REGISTRATION)
     }
@@ -424,6 +435,7 @@ impl ContentsStore for SledStore {
     type ContactsIter = SledContactsIter;
     type GroupsIter = SledGroupsIter;
     type MessagesIter = SledMessagesIter;
+    type ThreadMetadataIter = SledThreadMetadataIter;
 
     fn clear_contacts(&mut self) -> Result<(), SledStoreError> {
         self.write().drop_tree(SLED_TREE_CONTACTS)?;
@@ -603,6 +615,33 @@ impl ContentsStore for SledStore {
         let key = self.profile_key_for_uuid(uuid, key);
         self.get(SLED_TREE_PROFILES, key)
     }
+    /// Thread metadata
+
+    fn save_thread_metadata(
+        &mut self,
+        metadata: ThreadMetadata,
+    ) -> Result<(), Self::ContentsStoreError> {
+        let key = self.thread_metadata_key(metadata.thread.clone());
+        self.insert(SLED_TREE_THREADS_METADATA, key, metadata)?;
+        Ok(())
+    }
+
+    fn thread_metadata(
+        &self,
+        thread: Thread,
+    ) -> Result<Option<ThreadMetadata>, Self::ContentsStoreError> {
+        let key = self.thread_metadata_key(thread);
+        self.get(SLED_TREE_THREADS_METADATA, key)
+    }
+
+    fn thread_metadatas(&self) -> Result<Self::ThreadMetadataIter, SledStoreError> {
+        let tree = self.read().open_tree(SLED_TREE_THREADS_METADATA)?;
+        let iter = tree.iter();
+        Ok(SledThreadMetadataIter {
+            cipher: self.cipher.clone(),
+            iter,
+        })
+    }
 }
 
 impl PreKeysStore for SledStore {
@@ -1140,6 +1179,28 @@ impl DoubleEndedIterator for SledMessagesIter {
     }
 }
 
+pub struct SledThreadMetadataIter {
+    cipher: Option<Arc<StoreCipher>>,
+    iter: sled::Iter,
+}
+
+impl Iterator for SledThreadMetadataIter {
+    type Item = Result<ThreadMetadata, SledStoreError>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.iter
+            .next()?
+            .map_err(SledStoreError::from)
+            .and_then(|(_key, value)| {
+                self.cipher.as_ref().map_or_else(
+                    || serde_json::from_slice(&value).map_err(SledStoreError::from),
+                    |c| c.decrypt_value(&value).map_err(SledStoreError::from),
+                )
+            })
+            .into()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use core::fmt;

presage/Cargo.toml

@@ -13,7 +13,7 @@ base64 = "0.21"
 futures = "0.3"
 log = "0.4.8"
 rand = "0.8"
-serde = "1.0"
+serde =  { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 thiserror = "1.0"
 url = "2.2"

presage/src/errors.rs

@@ -56,6 +56,8 @@ pub enum Error<S: std::error::Error> {
     AttachmentCipherError(#[from] libsignal_service::attachment_cipher::AttachmentCipherError),
     #[error("unknown group")]
     UnknownGroup,
+    #[error("unknown contact")]
+    UnknownContact,
     #[error("unknown recipient")]
     UnknownRecipient,
     #[error("timeout: {0}")]

presage/src/lib.rs

@@ -1,14 +1,36 @@
 mod cache;
 mod errors;
+mod serializers;
+use serde::{Deserialize, Serialize};
 pub mod manager;
-mod serde;
 pub mod store;
 
+pub use crate::libsignal_service::prelude;
+pub use errors::Error;
 pub use libsignal_service;
 /// Protobufs used in Signal protocol and service communication
 pub use libsignal_service::proto;
-
-pub use errors::Error;
 pub use manager::Manager;
+pub use store::Thread;
 
 const USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "-rs-", env!("CARGO_PKG_VERSION"));
+
+// TODO: open a PR in libsignal and make sure the bytes can be read from `GroupMasterKey` instead of using this type
+pub type GroupMasterKeyBytes = [u8; 32];
+
+#[derive(Deserialize, Serialize, Debug, Clone)]
+pub struct ThreadMetadata {
+    pub thread: Thread,
+    pub last_message: Option<ThreadMetadataMessageContent>,
+    pub unread_messages_count: usize,
+    pub title: Option<String>,
+    pub archived: bool,
+    pub muted: bool,
+}
+
+#[derive(Deserialize, Serialize, Debug, Clone)]
+pub struct ThreadMetadataMessageContent {
+    pub sender: prelude::Uuid,
+    pub timestamp: u64,
+    pub message: Option<String>,
+}

presage/src/manager/linking.rs

@@ -30,8 +30,11 @@ impl<S: Store> Manager<S, Linking> {
     ///
     /// #[tokio::main]
     /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    ///     let store =
-    ///         SledStore::open("/tmp/presage-example", MigrationConflictStrategy::Drop, OnNewIdentity::Trust)?;
+    ///     let store = SledStore::open(
+    ///         "/tmp/presage-example",
+    ///         MigrationConflictStrategy::Drop,
+    ///         OnNewIdentity::Trust,
+    ///     )?;
     ///
     ///     let (mut tx, mut rx) = oneshot::channel();
     ///     let (manager, err) = future::join(

presage/src/manager/registered.rs

@@ -45,7 +45,7 @@ use sha2::Digest;
 use tokio::sync::Mutex;
 
 use crate::cache::CacheCell;
-use crate::serde::serde_profile_key;
+use crate::serializers::serde_profile_key;
 use crate::store::{Store, Thread};
 use crate::{Error, Manager};
 

presage/src/manager/registration.rs

@@ -43,8 +43,11 @@ impl<S: Store> Manager<S, Registration> {
     ///
     /// #[tokio::main]
     /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    ///     let store =
-    ///         SledStore::open("/tmp/presage-example", MigrationConflictStrategy::Drop, OnNewIdentity::Trust)?;
+    ///     let store = SledStore::open(
+    ///         "/tmp/presage-example",
+    ///         MigrationConflictStrategy::Drop,
+    ///         OnNewIdentity::Trust,
+    ///     )?;
     ///
     ///     let manager = Manager::register(
     ///         store,

presage/src/store.rs

@@ -2,6 +2,7 @@
 
 use std::{fmt, ops::RangeBounds, time::SystemTime};
 
+use crate::{manager::RegistrationData, GroupMasterKeyBytes, ThreadMetadata};
 use libsignal_service::{
     content::{ContentBody, Metadata},
     groups_v2::{Group, Timer},
@@ -14,17 +15,23 @@ use libsignal_service::{
     },
     protocol::{IdentityKey, ProtocolAddress, ProtocolStore, SenderKeyStore},
     session_store::SessionStoreExt,
-    zkgroup::GroupMasterKeyBytes,
     Profile,
 };
 use log::error;
 use serde::{Deserialize, Serialize};
 
-use crate::manager::RegistrationData;
-
 /// An error trait implemented by store error types
 pub trait StoreError: std::error::Error + Sync + Send + 'static {}
 
+// pub trait Store: ProtocolStore + SenderKeyStore + SessionStoreExt + Sync + Clone {
+// type Error: StoreError;
+//
+// type ContactsIter: Iterator<Item = Result<Contact, Self::Error>>;
+// type GroupsIter: Iterator<Item = Result<(GroupMasterKeyBytes, Group), Self::Error>>;
+// type MessagesIter: Iterator<Item = Result<Content, Self::Error>>;
+// type ThreadMetadataIter: Iterator<Item = Result<ThreadMetadata, Self::Error>>;
+//
+/// State
 /// Stores the registered state of the manager
 pub trait StateStore {
     type StateStoreError: StoreError;
@@ -57,6 +64,11 @@ pub trait ContentsStore {
     /// Each items is a tuple consisting of the group master key and its corresponding data.
     type GroupsIter: Iterator<Item = Result<(GroupMasterKeyBytes, Group), Self::ContentsStoreError>>;
 
+    /// Iterator over all stored thread metadata
+    ///
+    /// Each item is a tuple consisting of the thread and its corresponding metadata.
+    type ThreadMetadataIter: Iterator<Item = Result<ThreadMetadata, Self::ContentsStoreError>>;
+
     /// Iterator over all stored messages
     type MessagesIter: Iterator<Item = Result<Content, Self::ContentsStoreError>>;
 
@@ -233,7 +245,22 @@ pub trait ContentsStore {
         profile: Profile,
     ) -> Result<(), Self::ContentsStoreError>;
 
-    /// Retrieve a profile by [Uuid] and [ProfileKey].
+    /// Retrieve ThereadMetadata for all threads.
+    fn thread_metadatas(&self) -> Result<Self::ThreadMetadataIter, Self::ContentsStoreError>;
+
+    /// Retrieve ThereadMetadata for a single thread.
+    fn thread_metadata(
+        &self,
+        thread: Thread,
+    ) -> Result<Option<ThreadMetadata>, Self::ContentsStoreError>;
+
+    /// Save ThereadMetadata for a single thread.
+    /// This will overwrite any existing metadata for the thread.
+    fn save_thread_metadata(
+        &mut self,
+        metadata: ThreadMetadata,
+    ) -> Result<(), Self::ContentsStoreError>;
+
     fn profile(
         &self,
         uuid: Uuid,