needs tests

xmtp/src/storage/encrypted_store/mod.rs

@@ -9,7 +9,6 @@
 //! if there are any outstanding database migrations and perform them as needed. When updating the
 //! table definitions `schema.rs` must also be updated. To generate the correct schemas you can run
 //! `diesel print-schema` or use `cargo run update-schema` which will update the files for you.
-//!
 
 pub mod models;
 pub mod schema;

xmtp_mls/src/storage/encrypted_store/group.rs

@@ -11,8 +11,7 @@ use diesel::{
 };
 
 use super::schema::groups;
-use crate::impl_fetch;
-use crate::impl_store;
+use crate::{impl_fetch, impl_store};
 
 /// The Group ID type.
 pub type ID = Vec<u8>;
@@ -90,23 +89,21 @@ mod tests {
 
     #[test]
     fn it_stores_group() {
-        with_store(|store| {
-            let mut conn = store.conn().unwrap();
+        with_store(|_, conn| {
             let test_group = StoredGroup::new(vec![0x0], 100, GroupMembershipState::Allowed);
 
-            test_group.store(&mut conn).unwrap();
-            assert_eq!(groups.first::<StoredGroup>(&mut conn).unwrap(), test_group);
+            test_group.store(conn).unwrap();
+            assert_eq!(groups.first::<StoredGroup>(conn).unwrap(), test_group);
         })
     }
 
     #[test]
     fn it_fetches_group() {
-        with_store(|store| {
-            let mut conn = store.conn().unwrap();
+        with_store(|_, conn| {
             let test_group = StoredGroup::new(vec![0x0], 100, GroupMembershipState::Allowed);
             diesel::insert_into(groups)
                 .values(test_group.clone())
-                .execute(&mut conn)
+                .execute(conn)
                 .unwrap();
             let fetched_group = conn.fetch(vec![0x0]).ok().flatten().unwrap();
             assert_eq!(test_group, fetched_group);
@@ -115,14 +112,13 @@ mod tests {
 
     #[test]
     fn it_updates_group_membership_state() {
-        with_store(|store| {
+        with_store(|store, conn| {
             let id = vec![0x0];
-            let mut conn = store.conn().unwrap();
             let test_group = StoredGroup::new(id.clone(), 100, GroupMembershipState::Pending);
 
-            test_group.store(&mut conn).unwrap();
+            test_group.store(conn).unwrap();
             let updated_group = store
-                .update_group_membership(&mut conn, id, GroupMembershipState::Rejected)
+                .update_group_membership(conn, id, GroupMembershipState::Rejected)
                 .unwrap();
             assert_eq!(
                 updated_group,

xmtp_mls/src/storage/encrypted_store/group_message.rs

@@ -1,10 +1,17 @@
-use diesel::prelude::*;
+use diesel::{
+    backend::Backend,
+    deserialize::{self, FromSql, FromSqlRow},
+    expression::AsExpression,
+    prelude::*,
+    serialize::{self, IsNull, Output, ToSql},
+    sql_types::Integer,
+    sqlite::Sqlite,
+};
 
-use super::schema::group_messages;
-use crate::impl_fetch;
-use crate::impl_store;
+use super::{schema::group_messages, DbConnection, EncryptedMessageStore};
+use crate::{impl_fetch, impl_store, StorageError};
 
-#[derive(Insertable, Identifiable, Queryable, Debug, Clone)]
+#[derive(Insertable, Identifiable, Queryable, Debug, Clone, PartialEq, Eq)]
 #[diesel(table_name = group_messages)]
 #[diesel(primary_key(id))]
 /// Successfully processed messages to be returned to the User.
@@ -18,12 +25,138 @@ pub struct StoredGroupMessage {
     /// Time in nanoseconds the message was sent.
     pub sent_at_ns: i64,
     /// Group Message Kind Enum
-    pub kind: i32,
+    pub kind: GroupMessageKind,
     /// The ID of the App Installation this message was sent from.
     pub sender_installation_id: Vec<u8>,
     /// Network wallet address of the Sender
     pub sender_wallet_address: String,
 }
 
+#[repr(i32)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, AsExpression, FromSqlRow)]
+#[diesel(sql_type = Integer)]
+pub enum GroupMessageKind {
+    Application = 1,
+    MemberAdded = 2,
+    MemberRemoved = 3
+}
+
+impl ToSql<Integer, Sqlite> for GroupMessageKind 
+where
+    i32: ToSql<Integer, Sqlite>,
+{
+    fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Sqlite>) -> serialize::Result {
+        out.set_value(*self as i32);
+        Ok(IsNull::No)
+    }
+}
+
+impl FromSql<Integer, Sqlite> for GroupMessageKind 
+where
+    i32: FromSql<Integer, Sqlite>,
+{
+    fn from_sql(bytes: <Sqlite as Backend>::RawValue<'_>) -> deserialize::Result<Self> {
+        match i32::from_sql(bytes)? {
+            1 => Ok(GroupMessageKind::Application),
+            2 => Ok(GroupMessageKind::MemberAdded),
+            3 => Ok(GroupMessageKind::MemberRemoved),
+            x => Err(format!("Unrecognized variant {}", x).into()),
+        }
+    }
+}
+
 impl_fetch!(StoredGroupMessage, group_messages, Vec<u8>);
 impl_store!(StoredGroupMessage, group_messages);
+
+
+impl EncryptedMessageStore {
+
+    /// Query for group messages
+    pub fn get_group_messages(
+        &self,
+        conn: &mut DbConnection,
+        group_id: &[u8],
+        sent_after: Option<i64>,
+        sent_before: Option<i64>,
+        kind: Option<i32>,
+    ) -> Result<Vec<StoredGroupMessage>, StorageError> {
+        use super::schema::group_messages::dsl;
+
+        let mut query = dsl::group_messages
+            .filter(dsl::group_id.eq(group_id))
+            .into_boxed();
+
+        if let Some(sent_after) = sent_after {
+            query = query.filter(dsl::sent_at_ns.gt(sent_after));
+        }
+
+        if let Some(sent_before) = sent_before {
+            query = query.filter(dsl::sent_at_ns.lt(sent_before));
+        }
+
+        if let Some(kind) = kind {
+            query = query.filter(dsl::kind.eq(kind));
+        }
+        Ok(query.load::<StoredGroupMessage>(conn)?)
+    }
+
+    /// Get a particular group message
+    pub fn get_group_message(
+        &self,
+        id: &[u8],
+        conn: &mut DbConnection,
+    ) -> Result<Option<StoredGroupMessage>, StorageError> {
+        use super::schema::group_messages::dsl;
+        Ok(dsl::group_messages.filter(dsl::id.eq(id)).first(conn).optional()?)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rand::Rng;
+    use crate::{
+        storage::encrypted_store::{schema::groups::dsl::groups, tests::with_store},
+        Fetch, Store,
+    };
+
+    fn rand_bytes(length: usize) -> Vec<u8> {
+        (0..length).map(|_| { rand::random::<u8>() }).collect()
+    } 
+
+    fn generate_message(kind: Option<GroupMessageKind>) -> StoredGroupMessage {
+        let mut rng = rand::thread_rng();
+
+        StoredGroupMessage {
+            id: rand_bytes(32),
+            group_id: rand_bytes(32),
+            decrypted_message_bytes: rand_bytes(600),
+            sent_at_ns: rng.gen(),
+            sender_installation_id: rand_bytes(64),
+            sender_wallet_address: "0x0".to_string(),
+            kind: kind.unwrap_or(GroupMessageKind::Application),
+        }
+    }
+
+    #[test]
+    fn no_error_on_empty_messages() {
+        with_store(|store, conn| {
+            let id = vec![0x0];
+            let mut conn = store.conn().unwrap();
+            // TODO: could replace w/ something like an assert_ok macro in tokio-test. not sure
+            // it's worth pulling the whole library for that
+            assert!(matches!(store.get_group_message(&id, &mut conn), Ok(_)));
+        })
+    }
+
+    #[test]
+    fn it_gets_messages() {
+        with_store(|store, conn| {
+            let message = generate_message(None);
+            let id = message.id.clone();
+            message.store(conn).unwrap();
+            let stored_message = store.get_group_message(&id, conn).ok().flatten().unwrap();
+            assert_eq!(message, stored_message);
+        })
+    }
+}

xmtp_mls/src/storage/encrypted_store/mod.rs

@@ -9,7 +9,6 @@
 //! if there are any outstanding database migrations and perform them as needed. When updating the
 //! table definitions `schema.rs` must also be updated. To generate the correct schemas you can run
 //! `diesel print-schema` or use `cargo run update-schema` which will update the files for you.
-//!
 
 pub mod group;
 pub mod group_intent;
@@ -29,6 +28,7 @@ use diesel::{
 };
 use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
 use group::{GroupMembershipState, StoredGroup};
+use group_message::StoredGroupMessage;
 use log::warn;
 use rand::RngCore;
 use xmtp_cryptography::utils as crypto_utils;
@@ -231,6 +231,7 @@ macro_rules! impl_store {
     };
 }
 
+
 #[cfg(test)]
 mod tests {
     use std::{boxed::Box, fs};
@@ -254,15 +255,16 @@ mod tests {
     /// Test harness that loads an Ephemeral store.
     pub fn with_store<F, R>(fun: F) -> R
     where
-        F: FnOnce(EncryptedMessageStore) -> R,
+        F: FnOnce(EncryptedMessageStore, &mut super::DbConnection) -> R,
     {
         crate::tests::setup();
         let store = EncryptedMessageStore::new(
             StorageOption::Ephemeral,
             EncryptedMessageStore::generate_enc_key(),
         )
         .unwrap();
-        fun(store)
+        let mut conn = store.conn().expect("acquiring a Connection failed");
+        fun(store, &mut conn)
     }
 
     #[test]

xmtp_mls/src/storage/encrypted_store/topic_refresh_state.rs

@@ -1,8 +1,7 @@
 use diesel::prelude::*;
 
 use super::schema::topic_refresh_state;
-use crate::impl_fetch;
-use crate::impl_store;
+use crate::{impl_fetch, impl_store};
 
 #[derive(Insertable, Identifiable, Queryable, Debug, Clone)]
 #[diesel(table_name = topic_refresh_state)]