mls-storage: add MessageState

Signed-off-by: Yuki Kishimoto <[email protected]>

Author: erskingardner

crates/nostr-mls-memory-storage/src/lib.rs

@@ -173,7 +173,7 @@ mod tests {
     use nostr::{EventId, Kind, PublicKey, RelayUrl, Tags, Timestamp, UnsignedEvent};
     use nostr_mls_storage::groups::types::{Group, GroupExporterSecret, GroupState, GroupType};
     use nostr_mls_storage::groups::GroupStorage;
-    use nostr_mls_storage::messages::types::{Message, ProcessedMessageState};
+    use nostr_mls_storage::messages::types::{Message, MessageState, ProcessedMessageState};
     use nostr_mls_storage::messages::MessageStorage;
     use nostr_mls_storage::welcomes::types::{ProcessedWelcomeState, Welcome, WelcomeState};
     use nostr_mls_storage::welcomes::WelcomeStorage;
@@ -533,6 +533,7 @@ mod tests {
                 "Hello, world!".to_string(),
             ),
             wrapper_event_id: wrapper_id,
+            state: MessageState::Created,
         };
 
         // Save the message

crates/nostr-mls-sqlite-storage/migrations/V100__initial.sql

@@ -51,6 +51,7 @@ CREATE TABLE IF NOT EXISTS messages (
     tags JSONB NOT NULL,
     event JSONB NOT NULL,
     wrapper_event_id BLOB NOT NULL, -- Wrapper event ID as byte array
+    state TEXT NOT NULL,
     FOREIGN KEY (mls_group_id) REFERENCES groups(mls_group_id) ON DELETE CASCADE
 );
 
@@ -60,6 +61,7 @@ CREATE INDEX IF NOT EXISTS idx_messages_wrapper_event_id ON messages(wrapper_eve
 CREATE INDEX IF NOT EXISTS idx_messages_created_at ON messages(created_at);
 CREATE INDEX IF NOT EXISTS idx_messages_pubkey ON messages(pubkey);
 CREATE INDEX IF NOT EXISTS idx_messages_kind ON messages(kind);
+CREATE INDEX IF NOT EXISTS idx_messages_state ON messages(state);
 
 -- Processed Messages table
 CREATE TABLE IF NOT EXISTS processed_messages (

crates/nostr-mls-sqlite-storage/src/db.rs

@@ -8,7 +8,9 @@ use nostr::{EventId, JsonUtil, Kind, PublicKey, RelayUrl, Tags, Timestamp, Unsig
 use nostr_mls_storage::groups::types::{
     Group, GroupExporterSecret, GroupRelay, GroupState, GroupType,
 };
-use nostr_mls_storage::messages::types::{Message, ProcessedMessage, ProcessedMessageState};
+use nostr_mls_storage::messages::types::{
+    Message, MessageState, ProcessedMessage, ProcessedMessageState,
+};
 use nostr_mls_storage::welcomes::types::{
     ProcessedWelcome, ProcessedWelcomeState, Welcome, WelcomeState,
 };
@@ -123,6 +125,7 @@ pub fn row_to_message(row: &Row) -> SqliteResult<Message> {
     let tags_json: &str = row.get_ref("tags")?.as_str()?;
     let event_json: &str = row.get_ref("event")?.as_str()?;
     let wrapper_event_id_blob: &[u8] = row.get_ref("wrapper_event_id")?.as_blob()?;
+    let state_str: &str = row.get_ref("state")?.as_str()?;
 
     // Parse values
     let id: EventId =
@@ -142,6 +145,9 @@ pub fn row_to_message(row: &Row) -> SqliteResult<Message> {
     let wrapper_event_id: EventId = EventId::from_slice(wrapper_event_id_blob)
         .map_err(|_| map_invalid_blob_data("Invalid wrapper event ID"))?;
 
+    let state: MessageState =
+        MessageState::from_str(state_str).map_err(|_| map_invalid_text_data("Invalid state"))?;
+
     Ok(Message {
         id,
         pubkey,
@@ -152,6 +158,7 @@ pub fn row_to_message(row: &Row) -> SqliteResult<Message> {
         tags,
         event,
         wrapper_event_id,
+        state,
     })
 }
 

crates/nostr-mls-sqlite-storage/src/messages.rs

@@ -27,8 +27,8 @@ impl MessageStorage for NostrMlsSqliteStorage {
         conn_guard
             .execute(
                 "INSERT OR REPLACE INTO messages
-             (id, pubkey, kind, mls_group_id, created_at, content, tags, event, wrapper_event_id)
-             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+             (id, pubkey, kind, mls_group_id, created_at, content, tags, event, wrapper_event_id, state)
+             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                 params![
                     message.id.as_bytes(),
                     message.pubkey.as_bytes(),
@@ -39,6 +39,7 @@ impl MessageStorage for NostrMlsSqliteStorage {
                     tags_json,
                     message.event.as_json(),
                     message.wrapper_event_id.as_bytes(),
+                    message.state.as_str(),
                 ],
             )
             .map_err(into_message_err)?;
@@ -114,7 +115,7 @@ mod tests {
     use nostr::{EventId, Kind, PublicKey, Tags, Timestamp, UnsignedEvent};
     use nostr_mls_storage::groups::types::{Group, GroupState, GroupType};
     use nostr_mls_storage::groups::GroupStorage;
-    use nostr_mls_storage::messages::types::ProcessedMessageState;
+    use nostr_mls_storage::messages::types::{MessageState, ProcessedMessageState};
 
     use super::*;
 
@@ -168,6 +169,7 @@ mod tests {
                 "content".to_string(),
             ),
             wrapper_event_id,
+            state: MessageState::Created,
         };
 
         // Save the message

crates/nostr-mls-storage/src/messages/types.rs

@@ -46,11 +46,80 @@ pub struct Message {
     pub event: UnsignedEvent,
     /// The event id of the 1059 event that contained the message
     pub wrapper_event_id: EventId,
+    /// The state of the message
+    pub state: MessageState,
+}
+
+/// The state of the message
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum MessageState {
+    /// The message was created successfully and stored but we don't yet know if it was published to relays.
+    Created,
+    /// The message was successfully processed and stored in the database
+    Processed,
+    /// The message was deleted by the original sender - via a delete event
+    Deleted,
+}
+
+impl fmt::Display for MessageState {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.as_str())
+    }
+}
+
+impl MessageState {
+    /// Get as `&str`
+    pub fn as_str(&self) -> &str {
+        match self {
+            Self::Created => "created",
+            Self::Processed => "processed",
+            Self::Deleted => "deleted",
+        }
+    }
+}
+
+impl FromStr for MessageState {
+    type Err = MessageError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "created" => Ok(Self::Created),
+            "processed" => Ok(Self::Processed),
+            "deleted" => Ok(Self::Deleted),
+            _ => Err(MessageError::InvalidParameters(format!(
+                "Invalid message state: {}",
+                s
+            ))),
+        }
+    }
+}
+
+impl Serialize for MessageState {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(self.as_str())
+    }
+}
+
+impl<'de> Deserialize<'de> for MessageState {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let s: String = String::deserialize(deserializer)?;
+        Self::from_str(&s).map_err(serde::de::Error::custom)
+    }
 }
 
 /// The Processing State of the message,
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub enum ProcessedMessageState {
+    /// The processed message (and message) was created successfully and stored but we don't yet know if it was published to relays.
+    /// This state only happens when you are sending a message. Since we can't decrypt messages from ourselves in MLS groups,
+    /// once we see this message we mark it as processed but skip the rest of the processing.
+    Created,
     /// The message was successfully processed and stored in the database
     Processed,
     /// The message failed to be processed and stored in the database
@@ -67,6 +136,7 @@ impl ProcessedMessageState {
     /// Get as `&str`
     pub fn as_str(&self) -> &str {
         match self {
+            Self::Created => "created",
             Self::Processed => "processed",
             Self::Failed => "failed",
         }
@@ -78,6 +148,7 @@ impl FromStr for ProcessedMessageState {
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s {
+            "created" => Ok(Self::Created),
             "processed" => Ok(Self::Processed),
             "failed" => Ok(Self::Failed),
             _ => Err(MessageError::InvalidParameters(format!(
@@ -113,8 +184,104 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_message_state_from_str() {
+        assert_eq!(
+            MessageState::from_str("created").unwrap(),
+            MessageState::Created
+        );
+        assert_eq!(
+            MessageState::from_str("processed").unwrap(),
+            MessageState::Processed
+        );
+        assert_eq!(
+            MessageState::from_str("deleted").unwrap(),
+            MessageState::Deleted
+        );
+
+        let err = MessageState::from_str("invalid").unwrap_err();
+        match err {
+            MessageError::InvalidParameters(msg) => {
+                assert!(msg.contains("Invalid message state: invalid"));
+            }
+            _ => panic!("Expected InvalidParameters error"),
+        }
+    }
+
+    #[test]
+    fn test_message_state_to_string() {
+        assert_eq!(MessageState::Created.to_string(), "created");
+        assert_eq!(MessageState::Processed.to_string(), "processed");
+        assert_eq!(MessageState::Deleted.to_string(), "deleted");
+    }
+
+    #[test]
+    fn test_message_state_serialization() {
+        let created = MessageState::Created;
+        let serialized = serde_json::to_string(&created).unwrap();
+        assert_eq!(serialized, r#""created""#);
+
+        let processed = MessageState::Processed;
+        let serialized = serde_json::to_string(&processed).unwrap();
+        assert_eq!(serialized, r#""processed""#);
+
+        let deleted = MessageState::Deleted;
+        let serialized = serde_json::to_string(&deleted).unwrap();
+        assert_eq!(serialized, r#""deleted""#);
+    }
+
+    #[test]
+    fn test_message_state_deserialization() {
+        let created: MessageState = serde_json::from_str(r#""created""#).unwrap();
+        assert_eq!(created, MessageState::Created);
+
+        let processed: MessageState = serde_json::from_str(r#""processed""#).unwrap();
+        assert_eq!(processed, MessageState::Processed);
+
+        let deleted: MessageState = serde_json::from_str(r#""deleted""#).unwrap();
+        assert_eq!(deleted, MessageState::Deleted);
+
+        // Test invalid state
+        let result = serde_json::from_str::<MessageState>(r#""invalid""#);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_message_serialization() {
+        // Create a message to test serialization
+        let pubkey =
+            PublicKey::from_hex("8a9de562cbbed225b6ea0118dd3997a02df92c0bffd2224f71081a7450c3e549")
+                .unwrap();
+        let message = Message {
+            id: EventId::all_zeros(),
+            pubkey: pubkey.clone(),
+            kind: Kind::MlsGroupMessage,
+            mls_group_id: vec![1, 2, 3, 4],
+            created_at: Timestamp::now(),
+            content: "Test message".to_string(),
+            tags: Tags::new(),
+            event: UnsignedEvent::new(
+                pubkey,
+                Timestamp::now(),
+                Kind::MlsGroupMessage,
+                Tags::new(),
+                "Test message".to_string(),
+            ),
+            wrapper_event_id: EventId::all_zeros(),
+            state: MessageState::Created,
+        };
+
+        let serialized = serde_json::to_value(&message).unwrap();
+        assert_eq!(serialized["state"], json!("created"));
+        assert_eq!(serialized["content"], json!("Test message"));
+    }
+
     #[test]
     fn test_processed_message_state_from_str() {
+        assert_eq!(
+            ProcessedMessageState::from_str("created").unwrap(),
+            ProcessedMessageState::Created
+        );
         assert_eq!(
             ProcessedMessageState::from_str("processed").unwrap(),
             ProcessedMessageState::Processed
@@ -135,12 +302,17 @@ mod tests {
 
     #[test]
     fn test_processed_message_state_to_string() {
+        assert_eq!(ProcessedMessageState::Created.to_string(), "created");
         assert_eq!(ProcessedMessageState::Processed.to_string(), "processed");
         assert_eq!(ProcessedMessageState::Failed.to_string(), "failed");
     }
 
     #[test]
     fn test_processed_message_state_serialization() {
+        let created = ProcessedMessageState::Created;
+        let serialized = serde_json::to_string(&created).unwrap();
+        assert_eq!(serialized, r#""created""#);
+
         let processed = ProcessedMessageState::Processed;
         let serialized = serde_json::to_string(&processed).unwrap();
         assert_eq!(serialized, r#""processed""#);
@@ -152,11 +324,18 @@ mod tests {
 
     #[test]
     fn test_processed_message_state_deserialization() {
+        let created: ProcessedMessageState = serde_json::from_str(r#""created""#).unwrap();
+        assert_eq!(created, ProcessedMessageState::Created);
+
         let processed: ProcessedMessageState = serde_json::from_str(r#""processed""#).unwrap();
         assert_eq!(processed, ProcessedMessageState::Processed);
 
         let failed: ProcessedMessageState = serde_json::from_str(r#""failed""#).unwrap();
         assert_eq!(failed, ProcessedMessageState::Failed);
+
+        // Test invalid state
+        let result = serde_json::from_str::<ProcessedMessageState>(r#""invalid""#);
+        assert!(result.is_err());
     }
 
     #[test]
@@ -172,6 +351,50 @@ mod tests {
 
         let serialized = serde_json::to_value(&processed_message).unwrap();
         assert_eq!(serialized["state"], json!("processed"));
-        assert_eq!(serialized["failure_reason"], json!(""));
+        assert_eq!(serialized["failure_reason"], json!(null));
+
+        // Create a failed message with a reason
+        let failed_message = ProcessedMessage {
+            wrapper_event_id: EventId::all_zeros(),
+            message_event_id: Some(EventId::all_zeros()),
+            processed_at: Timestamp::now(),
+            state: ProcessedMessageState::Failed,
+            failure_reason: Some("Decryption failed".to_string()),
+        };
+
+        let serialized = serde_json::to_value(&failed_message).unwrap();
+        assert_eq!(serialized["state"], json!("failed"));
+        assert_eq!(serialized["failure_reason"], json!("Decryption failed"));
+        assert!(serialized["message_event_id"].is_string());
+    }
+
+    #[test]
+    fn test_processed_message_deserialization() {
+        let json_str = r#"{
+            "wrapper_event_id": "0000000000000000000000000000000000000000000000000000000000000000",
+            "message_event_id": null,
+            "processed_at": 1677721600,
+            "state": "processed",
+            "failure_reason": null
+        }"#;
+
+        let processed_message: ProcessedMessage = serde_json::from_str(json_str).unwrap();
+        assert_eq!(processed_message.state, ProcessedMessageState::Processed);
+        assert_eq!(processed_message.failure_reason, None);
+
+        let json_str = r#"{
+            "wrapper_event_id": "0000000000000000000000000000000000000000000000000000000000000000",
+            "message_event_id": "0000000000000000000000000000000000000000000000000000000000000000",
+            "processed_at": 1677721600,
+            "state": "failed",
+            "failure_reason": "Decryption failed"
+        }"#;
+
+        let failed_message: ProcessedMessage = serde_json::from_str(json_str).unwrap();
+        assert_eq!(failed_message.state, ProcessedMessageState::Failed);
+        assert_eq!(
+            failed_message.failure_reason,
+            Some("Decryption failed".to_string())
+        );
     }
 }