From f3aa5d5cf01d6dcc8cba46b5cdb3c9ba9ffeab9f Mon Sep 17 00:00:00 2001 From: kaichaosun Date: Fri, 27 Feb 2026 12:53:13 +0800 Subject: [PATCH] feat: storage for conversations --- Cargo.lock | 2 + conversations/Cargo.toml | 4 + conversations/src/chat.rs | 615 ++++++++++++++++++++ conversations/src/conversation/privatev1.rs | 48 +- conversations/src/errors.rs | 2 + conversations/src/lib.rs | 2 + conversations/src/storage/db.rs | 260 +++++++++ conversations/src/storage/mod.rs | 14 + conversations/src/storage/types.rs | 62 ++ double-ratchets/src/storage/db.rs | 6 + double-ratchets/src/storage/session.rs | 198 +++---- storage/src/sqlite.rs | 16 +- 12 files changed, 1112 insertions(+), 117 deletions(-) create mode 100644 conversations/src/chat.rs create mode 100644 conversations/src/storage/db.rs create mode 100644 conversations/src/storage/mod.rs create mode 100644 conversations/src/storage/types.rs diff --git a/Cargo.lock b/Cargo.lock index bcdb99d..bffaee7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -501,6 +501,8 @@ dependencies = [ "prost", "rand_core", "safer-ffi", + "storage", + "tempfile", "thiserror", "x25519-dalek", ] diff --git a/conversations/Cargo.toml b/conversations/Cargo.toml index cdb02a9..b77d9f6 100644 --- a/conversations/Cargo.toml +++ b/conversations/Cargo.toml @@ -18,3 +18,7 @@ rand_core = { version = "0.6" } safer-ffi = "0.1.13" thiserror = "2.0.17" x25519-dalek = { version = "2.0.1", features = ["static_secrets", "reusable_secrets", "getrandom"] } +storage = { path = "../storage" } + +[dev-dependencies] +tempfile = "3" diff --git a/conversations/src/chat.rs b/conversations/src/chat.rs new file mode 100644 index 0000000..19b9be1 --- /dev/null +++ b/conversations/src/chat.rs @@ -0,0 +1,615 @@ +//! ChatManager with integrated SQLite persistence. +//! +//! This is the main entry point for the conversations API. It handles all +//! storage operations internally - users don't need to interact with storage directly. + +use std::rc::Rc; + +use double_ratchets::storage::RatchetStorage; +use prost::Message; + +use crate::{ + conversation::PrivateV1Convo, + conversation::{Convo, Id}, + errors::ChatError, + identity::Identity, + inbox::{Inbox, Introduction}, + proto, + storage::{ChatRecord, ChatStorage, StorageError}, + types::{AddressedEnvelope, ContentData}, +}; + +// Re-export StorageConfig from storage crate for convenience +pub use storage::StorageConfig; + +/// Error type for ChatManager operations. +#[derive(Debug, thiserror::Error)] +pub enum ChatManagerError { + #[error("chat error: {0}")] + Chat(#[from] ChatError), + + #[error("storage error: {0}")] + Storage(#[from] StorageError), + + #[error("chat not found: {0}")] + ChatNotFound(String), +} + +/// ChatManager is the main entry point for the chat API. +/// +/// It manages identity, inbox, and chats with all state persisted to SQLite. +/// Chats are loaded from storage on each operation - no in-memory caching. +/// Uses a single shared database for both chat metadata and ratchet state. +/// +/// # Example +/// +/// ```ignore +/// // Create a new chat manager with encrypted storage +/// let mut chat = ChatManager::open(StorageConfig::Encrypted { +/// path: "chat.db".into(), +/// key: "my_secret_key".into(), +/// })?; +/// +/// // Get your address to share with others +/// println!("My address: {}", chat.local_address()); +/// +/// // Create an intro bundle to share +/// let intro = chat.create_intro_bundle()?; +/// +/// // Start a chat with someone +/// let (chat_id, envelopes) = chat.start_private_chat(&their_intro, "Hello!")?; +/// // Send envelopes over the network... +/// +/// // Send more messages +/// let envelopes = chat.send_message(&chat_id, b"How are you?")?; +/// ``` +pub struct ChatManager { + identity: Rc, + inbox: Inbox, + /// Storage for chat metadata (identity, inbox keys, chat records). + storage: ChatStorage, + /// Storage config for creating ratchet storage instances. + /// For file/encrypted databases, SQLite handles connection efficiently. + /// For in-memory testing, use SharedInMemory to share data. + storage_config: StorageConfig, +} + +impl ChatManager { + /// Opens or creates a ChatManager with the given storage configuration. + /// + /// If an identity exists in storage, it will be restored. + /// Otherwise, a new identity will be created and saved. + pub fn open(config: StorageConfig) -> Result { + let mut storage = ChatStorage::new(config.clone())?; + + // Load or create identity + let identity = if let Some(identity) = storage.load_identity()? { + identity + } else { + let identity = Identity::new("default"); + storage.save_identity(&identity)?; + identity + }; + + let identity = Rc::new(identity); + let inbox = Inbox::new(Rc::clone(&identity)); + + Ok(Self { + identity, + inbox, + storage, + storage_config: config, + }) + } + + /// Creates a new in-memory ChatManager (for testing). + /// + /// Uses a shared in-memory SQLite database so that multiple storage + /// instances within the same ChatManager share data. + /// + /// The `db_name` should be unique per ChatManager instance to avoid + /// sharing data between different users. + pub fn in_memory(db_name: &str) -> Result { + Self::open(StorageConfig::SharedInMemory(db_name.to_string())) + } + + /// Creates a new RatchetStorage instance using the stored config. + fn create_ratchet_storage(&self) -> Result { + Ok(RatchetStorage::with_config(self.storage_config.clone())?) + } + + /// Load a chat from storage. + fn load_chat(&self, chat_id: &str) -> Result { + let ratchet_storage = self.create_ratchet_storage()?; + if ratchet_storage.exists(chat_id)? { + let base_conv_id = chat_id.parse()?; + Ok(PrivateV1Convo::open(ratchet_storage, base_conv_id)?) + } else if self.storage.chat_exists(chat_id)? { + // Chat metadata exists but no ratchet state - data inconsistency + Err(ChatManagerError::ChatNotFound(format!( + "{} (corrupted: missing ratchet state)", + chat_id + ))) + } else { + Err(ChatManagerError::ChatNotFound(chat_id.to_string())) + } + } + + /// Get the local identity's public address. + /// + /// This address can be shared with others so they can identify you. + pub fn local_address(&self) -> String { + hex::encode(self.identity.public_key().as_bytes()) + } + + /// Create an introduction bundle that can be shared with others. + /// + /// Others can use this bundle to initiate a chat with you. + /// Share it via QR code, link, or any other out-of-band method. + /// + /// The ephemeral key is automatically persisted to storage. + pub fn create_intro_bundle(&mut self) -> Result { + let (pkb, secret) = self.inbox.create_bundle(); + let intro = Introduction::from(pkb); + + // Persist the ephemeral key + let public_key_hex = hex::encode(intro.ephemeral_key.as_bytes()); + self.storage.save_inbox_key(&public_key_hex, &secret)?; + + Ok(intro) + } + + /// Start a new private conversation with someone using their introduction bundle. + /// + /// Returns the chat ID and envelopes that must be delivered to the remote party. + /// The chat state is automatically persisted (via RatchetSession). + pub fn start_private_chat( + &mut self, + remote_bundle: &Introduction, + initial_message: &str, + ) -> Result<(String, Vec), ChatManagerError> { + // Create new storage for this conversation's RatchetSession + let ratchet_storage = self.create_ratchet_storage()?; + + let (convo, payloads) = self.inbox.invite_to_private_convo( + ratchet_storage, + remote_bundle, + initial_message.to_string(), + )?; + + let chat_id = convo.id().to_string(); + + let envelopes: Vec = payloads + .into_iter() + .map(|p| p.to_envelope(chat_id.clone())) + .collect(); + + // Persist chat metadata + let chat_record = ChatRecord::new_private( + chat_id.clone(), + remote_bundle.installation_key, + payloads_delivery_address(&envelopes), + ); + self.storage.save_chat(&chat_record)?; + + // Ratchet state is automatically persisted by RatchetSession + // convo is dropped here - state already saved + + Ok((chat_id, envelopes)) + } + + /// Send a message to an existing chat. + /// + /// Returns envelopes that must be delivered to chat participants. + pub fn send_message( + &mut self, + chat_id: &str, + content: &[u8], + ) -> Result, ChatManagerError> { + // Load chat from storage + let mut chat = self.load_chat(chat_id)?; + + let payloads = chat.send_message(content)?; + + // Ratchet state is automatically persisted by RatchetSession + + let remote_id = chat.remote_id(); + Ok(payloads + .into_iter() + .map(|p| p.to_envelope(remote_id.clone())) + .collect()) + } + + /// Handle an incoming payload from the network. + /// + /// This processes both inbox handshakes (to establish new chats) and + /// messages for existing chats. + /// + /// Returns the decrypted content if successful. + /// Any new chats or state changes are automatically persisted. + pub fn handle_incoming(&mut self, payload: &[u8]) -> Result { + // Try to decode as an envelope + if let Ok(envelope) = proto::EnvelopeV1::decode(payload) { + let chat_id = &envelope.conversation_hint; + + // Check if we have this chat - if so, route to it for decryption + if !chat_id.is_empty() && self.chat_exists(chat_id)? { + return self.receive_message(chat_id, &envelope.payload); + } + + // We don't have this chat - try to handle as inbox handshake + // Pass the conversation_hint so both parties use the same chat ID + return self.handle_inbox_handshake(chat_id, &envelope.payload); + } + + // Not a valid envelope format + Err(ChatManagerError::Chat(ChatError::Protocol( + "invalid envelope format".to_string(), + ))) + } + + /// Handle an inbox handshake to establish a new chat. + fn handle_inbox_handshake( + &mut self, + conversation_hint: &str, + payload: &[u8], + ) -> Result { + // Extract the ephemeral key hex from the payload + let key_hex = Inbox::extract_ephemeral_key_hex(payload)?; + + // Load the ephemeral key from storage + let ephemeral_key = self + .storage + .load_inbox_key(&key_hex)? + .ok_or_else(|| ChatManagerError::Chat(ChatError::UnknownEphemeralKey()))?; + + let ratchet_storage = self.create_ratchet_storage()?; + let result = + self.inbox + .handle_frame(ratchet_storage, conversation_hint, payload, &ephemeral_key)?; + + let chat_id = result.convo.id().to_string(); + + // Persist the new chat metadata + let chat_record = ChatRecord { + chat_id: chat_id.clone(), + chat_type: "private_v1".to_string(), + remote_public_key: Some(result.remote_public_key), + remote_address: hex::encode(result.remote_public_key), + created_at: crate::utils::timestamp_millis() as i64, + }; + self.storage.save_chat(&chat_record)?; + + // Delete the ephemeral key from storage after successful handshake + self.storage.delete_inbox_key(&key_hex)?; + + // Ratchet state is automatically persisted by RatchetSession + // result.convo is dropped here - state already saved + + Ok(ContentData { + conversation_id: chat_id, + data: result.initial_content.unwrap_or_default(), + }) + } + + /// Receive and decrypt a message for an existing chat. + /// + /// The payload should be the raw encrypted payload bytes. + pub fn receive_message( + &mut self, + chat_id: &str, + payload: &[u8], + ) -> Result { + // Load chat from storage + let mut chat = self.load_chat(chat_id)?; + + // Decode and decrypt the payload + let encrypted_payload = proto::EncryptedPayload::decode(payload).map_err(|e| { + ChatManagerError::Chat(ChatError::Protocol(format!("failed to decode: {}", e))) + })?; + + let frame = chat.decrypt(encrypted_payload)?; + let content = PrivateV1Convo::extract_content(&frame).unwrap_or_default(); + + // Ratchet state is automatically persisted by RatchetSession + + Ok(ContentData { + conversation_id: chat_id.to_string(), + data: content, + }) + } + + /// List all chat IDs from storage. + pub fn list_chats(&self) -> Result, ChatManagerError> { + Ok(self.storage.list_chat_ids()?) + } + + /// Check if a chat exists in storage. + pub fn chat_exists(&self, chat_id: &str) -> Result { + Ok(self.storage.chat_exists(chat_id)?) + } + + /// Delete a chat from storage. + pub fn delete_chat(&mut self, chat_id: &str) -> Result<(), ChatManagerError> { + self.storage.delete_chat(chat_id)?; + // Also delete ratchet state from double-ratchets storage + if let Ok(mut ratchet_storage) = self.create_ratchet_storage() { + let _ = ratchet_storage.delete(chat_id); + } + Ok(()) + } +} + +/// Extract delivery address from envelopes (helper function). +fn payloads_delivery_address(envelopes: &[AddressedEnvelope]) -> String { + envelopes + .first() + .map(|e| e.delivery_address.clone()) + .unwrap_or_else(|| "unknown".to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_create_chat_manager() { + let manager = ChatManager::in_memory("test1").unwrap(); + assert!(!manager.local_address().is_empty()); + } + + #[test] + fn test_identity_persistence() { + let manager = ChatManager::in_memory("test2").unwrap(); + let address = manager.local_address(); + + // Identity should be persisted + let loaded = manager.storage.load_identity().unwrap(); + assert!(loaded.is_some()); + assert_eq!(loaded.unwrap().address(), address); + } + + #[test] + fn test_create_intro_bundle() { + let mut manager = ChatManager::in_memory("test3").unwrap(); + let bundle = manager.create_intro_bundle(); + assert!(bundle.is_ok()); + } + + #[test] + fn test_start_private_chat() { + let mut alice = ChatManager::in_memory("alice1").unwrap(); + let mut bob = ChatManager::in_memory("bob1").unwrap(); + + // Bob creates an intro bundle + let bob_intro = bob.create_intro_bundle().unwrap(); + + // Alice starts a chat with Bob + let result = alice.start_private_chat(&bob_intro, "Hello Bob!"); + assert!(result.is_ok()); + + let (chat_id, envelopes) = result.unwrap(); + assert!(!chat_id.is_empty()); + assert!(!envelopes.is_empty()); + + // Chat should be persisted + let stored = alice.list_chats().unwrap(); + assert!(stored.contains(&chat_id)); + } + + #[test] + fn test_inbox_key_persistence() { + let mut manager = ChatManager::in_memory("test4").unwrap(); + + // Create intro bundle (should persist ephemeral key) + let intro = manager.create_intro_bundle().unwrap(); + let key_hex = hex::encode(intro.ephemeral_key.as_bytes()); + + // Key should be persisted - load it directly + let loaded_key = manager.storage.load_inbox_key(&key_hex).unwrap(); + assert!(loaded_key.is_some()); + } + + #[test] + fn test_chat_exists() { + let mut alice = ChatManager::in_memory("alice2").unwrap(); + let mut bob = ChatManager::in_memory("bob2").unwrap(); + + let bob_intro = bob.create_intro_bundle().unwrap(); + let (chat_id, _) = alice.start_private_chat(&bob_intro, "Hello!").unwrap(); + + // Chat should exist + assert!(alice.chat_exists(&chat_id).unwrap()); + assert!(!alice.chat_exists("nonexistent").unwrap()); + } + + #[test] + fn test_delete_chat() { + let mut alice = ChatManager::in_memory("alice3").unwrap(); + let mut bob = ChatManager::in_memory("bob3").unwrap(); + + let bob_intro = bob.create_intro_bundle().unwrap(); + let (chat_id, _) = alice.start_private_chat(&bob_intro, "Hello!").unwrap(); + + // Delete chat + alice.delete_chat(&chat_id).unwrap(); + + // Chat should no longer exist + assert!(!alice.chat_exists(&chat_id).unwrap()); + assert!(alice.list_chats().unwrap().is_empty()); + } + + #[test] + fn test_ratchet_state_persistence() { + use tempfile::tempdir; + + // Create a temporary directory for the database + let dir = tempdir().unwrap(); + let db_path = dir.path().join("test.db"); + + let mut bob = ChatManager::in_memory("bob4").unwrap(); + let bob_intro = bob.create_intro_bundle().unwrap(); + + let chat_id; + + // Scope 1: Create chat and send messages + { + let mut alice = + ChatManager::open(StorageConfig::File(db_path.to_str().unwrap().to_string())) + .unwrap(); + + let result = alice.start_private_chat(&bob_intro, "Message 1").unwrap(); + chat_id = result.0; + + // Send more messages - this advances the ratchet + alice.send_message(&chat_id, b"Message 2").unwrap(); + alice.send_message(&chat_id, b"Message 3").unwrap(); + + // Chat should be in storage + assert!(alice.chat_exists(&chat_id).unwrap()); + } + // alice is dropped here, simulating app close + + // Scope 2: Reopen and verify chat is restored + { + let mut alice2 = + ChatManager::open(StorageConfig::File(db_path.to_str().unwrap().to_string())) + .unwrap(); + + // Chat should still be in storage + assert!(alice2.list_chats().unwrap().contains(&chat_id)); + + // Send another message - this will load the chat and advance ratchet + let result = alice2.send_message(&chat_id, b"Message 4"); + assert!(result.is_ok(), "Should be able to send after restore"); + } + } + + #[test] + fn test_full_message_roundtrip() { + use tempfile::tempdir; + + // Use temp files instead of in-memory for proper storage sharing + let dir = tempdir().unwrap(); + let alice_db = dir.path().join("alice.db"); + let bob_db = dir.path().join("bob.db"); + + let mut alice = + ChatManager::open(StorageConfig::File(alice_db.to_str().unwrap().to_string())).unwrap(); + let mut bob = + ChatManager::open(StorageConfig::File(bob_db.to_str().unwrap().to_string())).unwrap(); + + // Bob creates an intro bundle and shares it with Alice + let bob_intro = bob.create_intro_bundle().unwrap(); + + // Alice starts a chat with Bob and sends "Hello!" + let (alice_chat_id, envelopes) = + alice.start_private_chat(&bob_intro, "Hello Bob!").unwrap(); + + // Verify Alice has the chat + assert!(alice.chat_exists(&alice_chat_id).unwrap()); + assert_eq!(alice.list_chats().unwrap().len(), 1); + + // Simulate network delivery: Bob receives the envelope + let envelope = envelopes.first().unwrap(); + let content = bob.handle_incoming(&envelope.data).unwrap(); + + // Bob should have received the message + assert_eq!(content.data, b"Hello Bob!"); + + // Bob should now have a chat + assert_eq!(bob.list_chats().unwrap().len(), 1); + let bob_chat_id = bob.list_chats().unwrap().first().unwrap().clone(); + + // Bob replies to Alice + let bob_reply_envelopes = bob.send_message(&bob_chat_id, b"Hi Alice!").unwrap(); + assert!(!bob_reply_envelopes.is_empty()); + + // Alice receives Bob's reply + let bob_reply = bob_reply_envelopes.first().unwrap(); + let alice_received = alice.handle_incoming(&bob_reply.data).unwrap(); + + assert_eq!(alice_received.data, b"Hi Alice!"); + assert_eq!(alice_received.conversation_id, alice_chat_id); + + // Continue the conversation - Alice sends another message + let alice_envelopes = alice.send_message(&alice_chat_id, b"How are you?").unwrap(); + let alice_msg = alice_envelopes.first().unwrap(); + let bob_received = bob.handle_incoming(&alice_msg.data).unwrap(); + + assert_eq!(bob_received.data, b"How are you?"); + + // Bob replies again + let bob_envelopes = bob + .send_message(&bob_chat_id, b"I'm good, thanks!") + .unwrap(); + let bob_msg = bob_envelopes.first().unwrap(); + let alice_received2 = alice.handle_incoming(&bob_msg.data).unwrap(); + + assert_eq!(alice_received2.data, b"I'm good, thanks!"); + } + + #[test] + fn test_message_persistence_across_sessions() { + use tempfile::tempdir; + + let dir = tempdir().unwrap(); + let alice_db = dir.path().join("alice.db"); + let bob_db = dir.path().join("bob.db"); + + let alice_chat_id; + let bob_chat_id; + let bob_intro; + + // Phase 1: Establish chat + { + let mut alice = + ChatManager::open(StorageConfig::File(alice_db.to_str().unwrap().to_string())) + .unwrap(); + let mut bob = + ChatManager::open(StorageConfig::File(bob_db.to_str().unwrap().to_string())) + .unwrap(); + + bob_intro = bob.create_intro_bundle().unwrap(); + let (chat_id, envelopes) = alice.start_private_chat(&bob_intro, "Initial").unwrap(); + alice_chat_id = chat_id; + + // Bob receives + let envelope = envelopes.first().unwrap(); + let content = bob.handle_incoming(&envelope.data).unwrap(); + assert_eq!(content.data, b"Initial"); + bob_chat_id = bob.list_chats().unwrap().first().unwrap().clone(); + } + // Both dropped - simulates app restart + + // Phase 2: Continue conversation after restart + { + let mut alice = + ChatManager::open(StorageConfig::File(alice_db.to_str().unwrap().to_string())) + .unwrap(); + let mut bob = + ChatManager::open(StorageConfig::File(bob_db.to_str().unwrap().to_string())) + .unwrap(); + + // Both should have persisted chats + assert!(alice.list_chats().unwrap().contains(&alice_chat_id)); + assert!(bob.list_chats().unwrap().contains(&bob_chat_id)); + + // Alice sends a message (chat loads from storage) + let envelopes = alice + .send_message(&alice_chat_id, b"After restart") + .unwrap(); + + // Bob receives (chat loads from storage) + let envelope = envelopes.first().unwrap(); + let content = bob.handle_incoming(&envelope.data).unwrap(); + assert_eq!(content.data, b"After restart"); + + // Bob replies + let bob_envelopes = bob.send_message(&bob_chat_id, b"Still works!").unwrap(); + let bob_msg = bob_envelopes.first().unwrap(); + let alice_received = alice.handle_incoming(&bob_msg.data).unwrap(); + assert_eq!(alice_received.data, b"Still works!"); + } + } +} diff --git a/conversations/src/conversation/privatev1.rs b/conversations/src/conversation/privatev1.rs index 0b8042e..b924d3e 100644 --- a/conversations/src/conversation/privatev1.rs +++ b/conversations/src/conversation/privatev1.rs @@ -7,9 +7,12 @@ use chat_proto::logoschat::{ encryption::{Doubleratchet, EncryptedPayload, encrypted_payload::Encryption}, }; use crypto::{PrivateKey, PublicKey, SymmetricKey32}; -use double_ratchets::{Header, InstallationKeyPair, RatchetState}; +use double_ratchets::{Header, InstallationKeyPair, RatchetSession, RatchetState, RatchetStorage}; use prost::{Message, bytes::Bytes}; -use std::fmt::Debug; +use std::{ + fmt::{self, Debug, Display, Formatter}, + str::FromStr, +}; use crate::{ conversation::{ChatError, ConversationId, Convo, Id}, @@ -52,10 +55,34 @@ impl BaseConvoId { } } +impl Display for BaseConvoId { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + write!(f, "{}", hex::encode(self.0)) + } +} + +impl FromStr for BaseConvoId { + type Err = ChatError; + + fn from_str(s: &str) -> Result { + let bytes = hex::decode(s).map_err(|_| ChatError::BadParsing("base conversation ID"))?; + + if bytes.len() != 18 { + return Err(ChatError::BadParsing("base conversation ID")); + } + + let mut arr = [0u8; 18]; + arr.copy_from_slice(&bytes); + + Ok(Self(arr)) + } +} + pub struct PrivateV1Convo { local_convo_id: String, remote_convo_id: String, dr_state: RatchetState, + session: Option, } impl PrivateV1Convo { @@ -74,6 +101,7 @@ impl PrivateV1Convo { local_convo_id, remote_convo_id, dr_state, + session: None, } } @@ -93,9 +121,25 @@ impl PrivateV1Convo { local_convo_id, remote_convo_id, dr_state, + session: None, } } + /// Open an existing conversation from storage. + pub fn open(storage: RatchetStorage, base_convo_id: BaseConvoId) -> Result { + let local_convo_id = base_convo_id.id_for_participant(Role::Responder); + let remote_convo_id = base_convo_id.id_for_participant(Role::Initiator); + + let session = RatchetSession::open(storage, &local_convo_id)?; + + Ok(Self { + local_convo_id, + remote_convo_id, + dr_state: session.state().clone(), + session: Some(session), + }) + } + fn encrypt(&mut self, frame: PrivateV1Frame) -> EncryptedPayload { let encoded_bytes = frame.encode_to_vec(); let (cipher_text, header) = self.dr_state.encrypt_message(&encoded_bytes); diff --git a/conversations/src/errors.rs b/conversations/src/errors.rs index d551960..1df0e68 100644 --- a/conversations/src/errors.rs +++ b/conversations/src/errors.rs @@ -20,6 +20,8 @@ pub enum ChatError { BadParsing(&'static str), #[error("convo with id: {0} was not found")] NoConvo(String), + #[error("session error: {0}")] + Session(#[from] double_ratchets::SessionError), } #[derive(Error, Debug)] diff --git a/conversations/src/lib.rs b/conversations/src/lib.rs index 79d6a5a..b82bb22 100644 --- a/conversations/src/lib.rs +++ b/conversations/src/lib.rs @@ -1,4 +1,5 @@ mod api; +mod chat; mod context; mod conversation; mod crypto; @@ -6,6 +7,7 @@ mod errors; mod identity; mod inbox; mod proto; +mod storage; mod types; mod utils; diff --git a/conversations/src/storage/db.rs b/conversations/src/storage/db.rs new file mode 100644 index 0000000..3bd0085 --- /dev/null +++ b/conversations/src/storage/db.rs @@ -0,0 +1,260 @@ +//! Chat-specific storage implementation. + +use storage::{RusqliteError, SqliteDb, StorageConfig, StorageError, params}; +use x25519_dalek::StaticSecret; + +use super::types::{ChatRecord, IdentityRecord}; +use crate::identity::Identity; + +/// Schema for chat storage tables. +/// Note: Ratchet state is stored by double_ratchets::RatchetStorage separately. +const CHAT_SCHEMA: &str = " + -- Identity table (single row) + CREATE TABLE IF NOT EXISTS identity ( + id INTEGER PRIMARY KEY CHECK (id = 1), + secret_key BLOB NOT NULL + ); + + -- Inbox ephemeral keys for handshakes + CREATE TABLE IF NOT EXISTS inbox_keys ( + public_key_hex TEXT PRIMARY KEY, + secret_key BLOB NOT NULL, + created_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now')) + ); + + -- Chat metadata + CREATE TABLE IF NOT EXISTS chats ( + chat_id TEXT PRIMARY KEY, + chat_type TEXT NOT NULL, + remote_public_key BLOB, + remote_address TEXT NOT NULL, + created_at INTEGER NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_chats_type ON chats(chat_type); +"; + +/// Chat-specific storage operations. +/// +/// This struct wraps a SqliteDb and provides domain-specific +/// storage operations for chat state (identity, inbox keys, chat metadata). +/// +/// Note: Ratchet state persistence is delegated to double_ratchets::RatchetStorage. +pub struct ChatStorage { + db: SqliteDb, +} + +impl ChatStorage { + /// Creates a new ChatStorage with the given configuration. + pub fn new(config: StorageConfig) -> Result { + let db = SqliteDb::new(config)?; + Self::run_migration(db) + } + + /// Creates a new chat storage with the given database. + fn run_migration(db: SqliteDb) -> Result { + db.connection().execute_batch(CHAT_SCHEMA)?; + Ok(Self { db }) + } + + // ==================== Identity Operations ==================== + + /// Saves the identity (secret key). + pub fn save_identity(&mut self, identity: &Identity) -> Result<(), StorageError> { + let record = IdentityRecord::from(identity); + self.db.connection().execute( + "INSERT OR REPLACE INTO identity (id, secret_key) VALUES (1, ?1)", + params![record.secret_key.as_slice()], + )?; + Ok(()) + } + + /// Loads the identity if it exists. + pub fn load_identity(&self) -> Result, StorageError> { + let mut stmt = self + .db + .connection() + .prepare("SELECT secret_key FROM identity WHERE id = 1")?; + + let result = stmt.query_row([], |row| { + let secret_key: Vec = row.get(0)?; + Ok(secret_key) + }); + + match result { + Ok(secret_key) => { + let bytes: [u8; 32] = secret_key + .try_into() + .map_err(|_| StorageError::InvalidData("Invalid secret key length".into()))?; + let record = IdentityRecord { secret_key: bytes }; + Ok(Some(Identity::from(record))) + } + Err(RusqliteError::QueryReturnedNoRows) => Ok(None), + Err(e) => Err(e.into()), + } + } + + // ==================== Inbox Key Operations ==================== + + /// Saves an inbox ephemeral key. + pub fn save_inbox_key( + &mut self, + public_key_hex: &str, + secret: &StaticSecret, + ) -> Result<(), StorageError> { + self.db.connection().execute( + "INSERT OR REPLACE INTO inbox_keys (public_key_hex, secret_key) VALUES (?1, ?2)", + params![public_key_hex, secret.as_bytes().as_slice()], + )?; + Ok(()) + } + + /// Loads a single inbox ephemeral key by public key hex. + pub fn load_inbox_key( + &self, + public_key_hex: &str, + ) -> Result, StorageError> { + let mut stmt = self + .db + .connection() + .prepare("SELECT secret_key FROM inbox_keys WHERE public_key_hex = ?1")?; + + let result = stmt.query_row(params![public_key_hex], |row| { + let secret_key: Vec = row.get(0)?; + Ok(secret_key) + }); + + match result { + Ok(secret_key) => { + let bytes: [u8; 32] = secret_key + .try_into() + .map_err(|_| StorageError::InvalidData("Invalid secret key length".into()))?; + Ok(Some(StaticSecret::from(bytes))) + } + Err(RusqliteError::QueryReturnedNoRows) => Ok(None), + Err(e) => Err(e.into()), + } + } + + /// Deletes an inbox ephemeral key after it has been used. + pub fn delete_inbox_key(&mut self, public_key_hex: &str) -> Result<(), StorageError> { + self.db.connection().execute( + "DELETE FROM inbox_keys WHERE public_key_hex = ?1", + params![public_key_hex], + )?; + Ok(()) + } + + // ==================== Chat Metadata Operations ==================== + + /// Saves a chat record. + pub fn save_chat(&mut self, chat: &ChatRecord) -> Result<(), StorageError> { + self.db.connection().execute( + "INSERT OR REPLACE INTO chats (chat_id, chat_type, remote_public_key, remote_address, created_at) + VALUES (?1, ?2, ?3, ?4, ?5)", + params![ + chat.chat_id, + chat.chat_type, + chat.remote_public_key.as_ref().map(|k| k.as_slice()), + chat.remote_address, + chat.created_at, + ], + )?; + Ok(()) + } + + /// Lists all chat IDs. + pub fn list_chat_ids(&self) -> Result, StorageError> { + let mut stmt = self.db.connection().prepare("SELECT chat_id FROM chats")?; + let rows = stmt.query_map([], |row| row.get(0))?; + + let mut ids = Vec::new(); + for row in rows { + ids.push(row?); + } + + Ok(ids) + } + + /// Checks if a chat exists in storage. + pub fn chat_exists(&self, chat_id: &str) -> Result { + let mut stmt = self + .db + .connection() + .prepare("SELECT 1 FROM chats WHERE chat_id = ?1")?; + + let exists = stmt.exists(params![chat_id])?; + Ok(exists) + } + + /// Finds a chat by remote address. + /// Returns the chat_id if found, None otherwise. + #[allow(dead_code)] + pub fn find_chat_by_remote_address( + &self, + remote_address: &str, + ) -> Result, StorageError> { + let mut stmt = self + .db + .connection() + .prepare("SELECT chat_id FROM chats WHERE remote_address = ?1 LIMIT 1")?; + + let mut rows = stmt.query(params![remote_address])?; + if let Some(row) = rows.next()? { + Ok(Some(row.get(0)?)) + } else { + Ok(None) + } + } + + /// Deletes a chat record. + /// Note: Ratchet state must be deleted separately via RatchetStorage. + pub fn delete_chat(&mut self, chat_id: &str) -> Result<(), StorageError> { + self.db + .connection() + .execute("DELETE FROM chats WHERE chat_id = ?1", params![chat_id])?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_identity_roundtrip() { + let mut storage = ChatStorage::new(StorageConfig::InMemory).unwrap(); + + // Initially no identity + assert!(storage.load_identity().unwrap().is_none()); + + // Save identity + let identity = Identity::new(); + let address = identity.address(); + storage.save_identity(&identity).unwrap(); + + // Load identity + let loaded = storage.load_identity().unwrap().unwrap(); + assert_eq!(loaded.address(), address); + } + + #[test] + fn test_chat_roundtrip() { + let mut storage = ChatStorage::new(StorageConfig::InMemory).unwrap(); + + let secret = x25519_dalek::StaticSecret::random(); + let remote_key = x25519_dalek::PublicKey::from(&secret); + let chat = ChatRecord::new_private( + "chat_123".to_string(), + remote_key, + "delivery_addr".to_string(), + ); + + // Save chat + storage.save_chat(&chat).unwrap(); + + // List chats + let ids = storage.list_chat_ids().unwrap(); + assert_eq!(ids, vec!["chat_123"]); + } +} diff --git a/conversations/src/storage/mod.rs b/conversations/src/storage/mod.rs new file mode 100644 index 0000000..5153dbf --- /dev/null +++ b/conversations/src/storage/mod.rs @@ -0,0 +1,14 @@ +//! Storage module for persisting chat state. +//! +//! This module provides storage implementations for the chat manager state, +//! built on top of the shared `storage` crate. +//! +//! Note: This module is internal. Users should use `ChatManager` which +//! handles all storage operations automatically. + +mod db; +pub(crate) mod types; + +pub(crate) use db::ChatStorage; +pub(crate) use storage::StorageError; +pub(crate) use types::ChatRecord; diff --git a/conversations/src/storage/types.rs b/conversations/src/storage/types.rs new file mode 100644 index 0000000..553ac1b --- /dev/null +++ b/conversations/src/storage/types.rs @@ -0,0 +1,62 @@ +//! Storage record types for serialization/deserialization. +//! +//! Note: Ratchet state types (RatchetStateRecord, SkippedKeyRecord) are in +//! double_ratchets::storage module and handled by RatchetStorage. + +use x25519_dalek::{PublicKey, StaticSecret}; + +use crate::identity::Identity; + +/// Record for storing identity (secret key). +#[derive(Debug)] +pub struct IdentityRecord { + /// The secret key bytes (32 bytes). + pub secret_key: [u8; 32], +} + +impl From<&Identity> for IdentityRecord { + fn from(identity: &Identity) -> Self { + Self { + secret_key: identity.secret().to_bytes(), + } + } +} + +impl From for Identity { + fn from(record: IdentityRecord) -> Self { + let secret = StaticSecret::from(record.secret_key); + Identity::from_secret(secret) + } +} + +/// Record for storing chat metadata. +/// Note: The actual double ratchet state is stored separately by RatchetStorage. +#[derive(Debug, Clone)] +pub struct ChatRecord { + /// Unique chat identifier. + pub chat_id: String, + /// Type of chat (e.g., "private_v1", "group_v1"). + pub chat_type: String, + /// Remote party's public key (for private chats). + pub remote_public_key: Option<[u8; 32]>, + /// Remote party's delivery address. + pub remote_address: String, + /// Creation timestamp (unix millis). + pub created_at: i64, +} + +impl ChatRecord { + pub fn new_private( + chat_id: String, + remote_public_key: PublicKey, + remote_address: String, + ) -> Self { + Self { + chat_id, + chat_type: "private_v1".to_string(), + remote_public_key: Some(remote_public_key.to_bytes()), + remote_address, + created_at: crate::utils::timestamp_millis() as i64, + } + } +} diff --git a/double-ratchets/src/storage/db.rs b/double-ratchets/src/storage/db.rs index 43b3f4b..a41d2bd 100644 --- a/double-ratchets/src/storage/db.rs +++ b/double-ratchets/src/storage/db.rs @@ -47,6 +47,12 @@ pub struct RatchetStorage { } impl RatchetStorage { + /// Creates a new RatchetStorage with the given configuration. + pub fn with_config(config: storage::StorageConfig) -> Result { + let db = SqliteDb::new(config)?; + Self::run_migration(db) + } + /// Opens an existing encrypted database file. pub fn new(path: &str, key: &str) -> Result { let db = SqliteDb::sqlcipher(path.to_string(), key.to_string())?; diff --git a/double-ratchets/src/storage/session.rs b/double-ratchets/src/storage/session.rs index ea3cdfc..2598d85 100644 --- a/double-ratchets/src/storage/session.rs +++ b/double-ratchets/src/storage/session.rs @@ -13,16 +13,19 @@ use super::RatchetStorage; /// A session wrapper that automatically persists ratchet state after operations. /// Provides rollback semantics - state is only saved if the operation succeeds. -pub struct RatchetSession<'a, D: HkdfInfo + Clone = DefaultDomain> { - storage: &'a mut RatchetStorage, +/// +/// This struct owns its storage, making it easy to store in other structs +/// and use across multiple operations without lifetime concerns. +pub struct RatchetSession { + storage: RatchetStorage, conversation_id: String, state: RatchetState, } -impl<'a, D: HkdfInfo + Clone> RatchetSession<'a, D> { +impl<'a, D: HkdfInfo + Clone> RatchetSession { /// Opens an existing session from storage. pub fn open( - storage: &'a mut RatchetStorage, + storage: RatchetStorage, conversation_id: impl Into, ) -> Result { let conversation_id = conversation_id.into(); @@ -36,7 +39,7 @@ impl<'a, D: HkdfInfo + Clone> RatchetSession<'a, D> { /// Creates a new session and persists the initial state. pub fn create( - storage: &'a mut RatchetStorage, + mut storage: RatchetStorage, conversation_id: impl Into, state: RatchetState, ) -> Result { @@ -51,7 +54,7 @@ impl<'a, D: HkdfInfo + Clone> RatchetSession<'a, D> { /// Initializes a new session as a sender and persists the initial state. pub fn create_sender_session( - storage: &'a mut RatchetStorage, + storage: RatchetStorage, conversation_id: &str, shared_secret: SharedSecret, remote_pub: PublicKey, @@ -65,7 +68,7 @@ impl<'a, D: HkdfInfo + Clone> RatchetSession<'a, D> { /// Initializes a new session as a receiver and persists the initial state. pub fn create_receiver_session( - storage: &'a mut RatchetStorage, + storage: RatchetStorage, conversation_id: &str, shared_secret: SharedSecret, dh_self: InstallationKeyPair, @@ -137,6 +140,12 @@ impl<'a, D: HkdfInfo + Clone> RatchetSession<'a, D> { &self.conversation_id } + /// Consumes the session and returns the underlying storage. + /// Useful when you need to reuse the storage for another session. + pub fn into_storage(self) -> RatchetStorage { + self.storage + } + /// Manually saves the current state. pub fn save(&mut self) -> Result<(), SessionError> { self.storage @@ -164,30 +173,29 @@ mod tests { #[test] fn test_session_create_and_open() { - let mut storage = create_test_storage(); + let storage = create_test_storage(); let shared_secret = [0x42; 32]; let bob_keypair = InstallationKeyPair::generate(); let alice: RatchetState = RatchetState::init_sender(shared_secret, *bob_keypair.public()); - // Create session - { - let session = RatchetSession::create(&mut storage, "conv1", alice).unwrap(); - assert_eq!(session.conversation_id(), "conv1"); - } + // Create session - session takes ownership of storage + let session = RatchetSession::create(storage, "conv1", alice).unwrap(); + assert_eq!(session.conversation_id(), "conv1"); + + // Get storage back from session to reopen + let storage = session.into_storage(); // Open existing session - { - let session: RatchetSession = - RatchetSession::open(&mut storage, "conv1").unwrap(); - assert_eq!(session.state().msg_send, 0); - } + let session: RatchetSession = + RatchetSession::open(storage, "conv1").unwrap(); + assert_eq!(session.state().msg_send, 0); } #[test] fn test_session_encrypt_persists() { - let mut storage = create_test_storage(); + let storage = create_test_storage(); let shared_secret = [0x42; 32]; let bob_keypair = InstallationKeyPair::generate(); @@ -195,158 +203,120 @@ mod tests { RatchetState::init_sender(shared_secret, *bob_keypair.public()); // Create and encrypt - { - let mut session = RatchetSession::create(&mut storage, "conv1", alice).unwrap(); - session.encrypt_message(b"Hello").unwrap(); - assert_eq!(session.state().msg_send, 1); - } + let mut session = RatchetSession::create(storage, "conv1", alice).unwrap(); + session.encrypt_message(b"Hello").unwrap(); + assert_eq!(session.state().msg_send, 1); + + // Get storage back and reopen + let storage = session.into_storage(); // Reopen - state should be persisted - { - let session: RatchetSession = - RatchetSession::open(&mut storage, "conv1").unwrap(); - assert_eq!(session.state().msg_send, 1); - } + let session: RatchetSession = + RatchetSession::open(storage, "conv1").unwrap(); + assert_eq!(session.state().msg_send, 1); } #[test] fn test_session_full_conversation() { - let mut storage = create_test_storage(); + // Use separate in-memory storages for alice and bob (simulates different devices) + let alice_storage = create_test_storage(); + let bob_storage = create_test_storage(); let shared_secret = [0x42; 32]; let bob_keypair = InstallationKeyPair::generate(); - let alice: RatchetState = - RatchetState::init_sender(shared_secret, *bob_keypair.public()); - let bob: RatchetState = + let alice_state: RatchetState = + RatchetState::init_sender(shared_secret, bob_keypair.public().clone()); + let bob_state: RatchetState = RatchetState::init_receiver(shared_secret, bob_keypair); // Alice sends - let (ct, header) = { - let mut session = RatchetSession::create(&mut storage, "alice", alice).unwrap(); - session.encrypt_message(b"Hello Bob").unwrap() - }; + let mut alice_session = RatchetSession::create(alice_storage, "conv", alice_state).unwrap(); + let (ct, header) = alice_session.encrypt_message(b"Hello Bob").unwrap(); // Bob receives - let plaintext = { - let mut session = RatchetSession::create(&mut storage, "bob", bob).unwrap(); - session.decrypt_message(&ct, header).unwrap() - }; + let mut bob_session = RatchetSession::create(bob_storage, "conv", bob_state).unwrap(); + let plaintext = bob_session.decrypt_message(&ct, header).unwrap(); assert_eq!(plaintext, b"Hello Bob"); // Bob replies - let (ct2, header2) = { - let mut session: RatchetSession = - RatchetSession::open(&mut storage, "bob").unwrap(); - session.encrypt_message(b"Hi Alice").unwrap() - }; + let (ct2, header2) = bob_session.encrypt_message(b"Hi Alice").unwrap(); // Alice receives - let plaintext2 = { - let mut session: RatchetSession = - RatchetSession::open(&mut storage, "alice").unwrap(); - session.decrypt_message(&ct2, header2).unwrap() - }; + let plaintext2 = alice_session.decrypt_message(&ct2, header2).unwrap(); assert_eq!(plaintext2, b"Hi Alice"); } #[test] fn test_session_open_or_create() { - let mut storage = create_test_storage(); + let storage = create_test_storage(); let shared_secret = [0x42; 32]; let bob_keypair = InstallationKeyPair::generate(); let bob_pub = *bob_keypair.public(); // First call creates - { - let session: RatchetSession = RatchetSession::create_sender_session( - &mut storage, - "conv1", - shared_secret, - bob_pub, - ) - .unwrap(); - assert_eq!(session.state().msg_send, 0); - } + let session: RatchetSession = + RatchetSession::create_sender_session(storage, "conv1", shared_secret, bob_pub.clone()) + .unwrap(); + assert_eq!(session.state().msg_send, 0); + let storage = session.into_storage(); - // Second call opens existing - { - let mut session: RatchetSession = - RatchetSession::open(&mut storage, "conv1").unwrap(); - session.encrypt_message(b"test").unwrap(); - } + // Second call opens existing and encrypts + let mut session: RatchetSession = + RatchetSession::open(storage, "conv1").unwrap(); + session.encrypt_message(b"test").unwrap(); + let storage = session.into_storage(); // Verify persistence - { - let session: RatchetSession = - RatchetSession::open(&mut storage, "conv1").unwrap(); - assert_eq!(session.state().msg_send, 1); - } + let session: RatchetSession = + RatchetSession::open(storage, "conv1").unwrap(); + assert_eq!(session.state().msg_send, 1); } #[test] fn test_create_sender_session_fails_when_conversation_exists() { - let mut storage = create_test_storage(); + let storage = create_test_storage(); let shared_secret = [0x42; 32]; let bob_keypair = InstallationKeyPair::generate(); let bob_pub = *bob_keypair.public(); // First creation succeeds - { - let _session: RatchetSession = RatchetSession::create_sender_session( - &mut storage, - "conv1", - shared_secret, - bob_pub, - ) - .unwrap(); - } + let session: RatchetSession = + RatchetSession::create_sender_session(storage, "conv1", shared_secret, bob_pub.clone()) + .unwrap(); + let storage = session.into_storage(); // Second creation should fail with ConversationAlreadyExists - { - let result: Result, _> = - RatchetSession::create_sender_session( - &mut storage, - "conv1", - shared_secret, - bob_pub, - ); + let result: Result, _> = + RatchetSession::create_sender_session(storage, "conv1", shared_secret, bob_pub.clone()); - assert!(matches!(result, Err(SessionError::ConvAlreadyExists(_)))); - } + assert!(matches!(result, Err(SessionError::ConvAlreadyExists(_)))); } #[test] fn test_create_receiver_session_fails_when_conversation_exists() { - let mut storage = create_test_storage(); + let storage = create_test_storage(); let shared_secret = [0x42; 32]; let bob_keypair = InstallationKeyPair::generate(); // First creation succeeds - { - let _session: RatchetSession = RatchetSession::create_receiver_session( - &mut storage, - "conv1", - shared_secret, - bob_keypair, - ) - .unwrap(); - } + let session: RatchetSession = + RatchetSession::create_receiver_session(storage, "conv1", shared_secret, bob_keypair) + .unwrap(); + let storage = session.into_storage(); // Second creation should fail with ConversationAlreadyExists - { - let another_keypair = InstallationKeyPair::generate(); - let result: Result, _> = - RatchetSession::create_receiver_session( - &mut storage, - "conv1", - shared_secret, - another_keypair, - ); + let another_keypair = InstallationKeyPair::generate(); + let result: Result, _> = + RatchetSession::create_receiver_session( + storage, + "conv1", + shared_secret, + another_keypair, + ); - assert!(matches!(result, Err(SessionError::ConvAlreadyExists(_)))); - } + assert!(matches!(result, Err(SessionError::ConvAlreadyExists(_)))); } } diff --git a/storage/src/sqlite.rs b/storage/src/sqlite.rs index 4d42e9d..8449532 100644 --- a/storage/src/sqlite.rs +++ b/storage/src/sqlite.rs @@ -8,8 +8,11 @@ use crate::StorageError; /// Configuration for SQLite storage. #[derive(Debug, Clone)] pub enum StorageConfig { - /// In-memory database (for testing). + /// In-memory database (isolated, for simple testing). InMemory, + /// Shared in-memory database with a name (multiple connections share data). + /// Use this when you need multiple storage instances to share the same in-memory DB. + SharedInMemory(String), /// File-based SQLite database. File(String), /// SQLCipher encrypted database. @@ -29,6 +32,17 @@ impl SqliteDb { pub fn new(config: StorageConfig) -> Result { let conn = match config { StorageConfig::InMemory => Connection::open_in_memory()?, + StorageConfig::SharedInMemory(ref name) => { + // Use URI mode to create a shared in-memory database + // Multiple connections with the same name share the same data + let uri = format!("file:{}?mode=memory&cache=shared", name); + Connection::open_with_flags( + &uri, + rusqlite::OpenFlags::SQLITE_OPEN_URI + | rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE + | rusqlite::OpenFlags::SQLITE_OPEN_CREATE, + )? + } StorageConfig::File(ref path) => Connection::open(path)?, StorageConfig::Encrypted { ref path, ref key } => { let conn = Connection::open(path)?;