logos-messaging/libchat
1
0
Fork 0
mirror of https://github.com/logos-messaging/libchat.git synced 2026-08-12 01:53:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: durable MLS storage folded into a single ChatStore

Browse Source 
Implements issue #112: MLS group state must survive process restarts. The MLS
provider hardcoded an in-memory store separate from the chat store, so a restart
lost every group. Fold the OpenMLS StorageProvider into libchat's own ChatStore
so one durable store holds both chat state and MLS group state.

- storage: ChatStore subsumes StorageProvider<CURRENT_VERSION>, so a ChatStore is
  the whole durable-storage contract (identity/ephemeral/conversation/ratchet
  sub-stores plus the MLS key-value surface).
- chat-sqlite: ChatStorage implements StorageProvider over an mls_kv table
  (migration 003_mls_storage), a byte-faithful port of MemoryStorage with decode
  errors surfaced instead of unwrapped and the reference clear_proposal_queue
  orphan-key bug fixed. The separate SqliteMlsStorage type is removed.
- conversations: MlsPqProvider becomes a transient view borrowing the shared
  store and a long-lived crypto backend, so the store stays singly owned;
  ServiceContext hands it out via mls_provider(). ExternalServices drops its ST
  associated type, leaving one CS. GroupV2/de_mls's thread-safe-error bound is
  restated on CS because an associated-type bound does not elaborate through a
  supertrait.
- client: the store is always ChatStorage, so fix it as such and drop the
  vestigial store generic from ChatClient and the builder. Retire the stubbed
  in-memory MemStore in favour of ChatStorage::in_memory().
- GroupV1 conversations resume across a restart through the existing
  MlsGroup::load path, proven by a drop-and-reload integration test.
This commit is contained in:
osmaczko 2026-07-01 17:56:10 +02:00
parent d2cb3017e9
commit 7332c152c0
No known key found for this signature in database
GPG Key ID: 28DEFC027B185BB8
29 changed files with 1292 additions and 377 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Cargo.lock generated
View File

@ -1341,9 +1341,13 @@ version = "0.1.0"
dependencies = [
"crypto",
"hex",
"openmls_traits 0.5.0",
"rusqlite",
"serde",
"serde_json",
"storage",
"tempfile",
"thiserror",
"zeroize",
]
@ -5533,6 +5537,7 @@ name = "storage"
version = "0.1.0"
dependencies = [
"crypto",
"openmls_traits 0.5.0",
"thiserror",
]

12
bin/chat-cli/src/app.rs
View File

@ -5,7 +5,7 @@ use std::path::{Path, PathBuf};
use anyhow::Result;
use arboard::Clipboard;
use crossbeam_channel::Receiver;
use logos_chat::{ChatClient, ChatStore, Event, IdentityProvider, RegistrationService, Transport};
use logos_chat::{ChatClient, Event, IdentityProvider, RegistrationService, Transport};
use serde::{Deserialize, Serialize};
use crate::utils::now;
@ -41,14 +41,13 @@ pub struct AppState {
pub active_chat: Option<String>,
}
pub struct ChatApp<I, T, R, S>
pub struct ChatApp<I, T, R>
where
I: IdentityProvider + Send + 'static,
T: Transport,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
pub client: ChatClient<I, T, R, S>,
pub client: ChatClient<I, T, R>,
events: Receiver<Event>,
pub state: AppState,
/// Ephemeral command output — not persisted, cleared on chat switch.
@ -59,15 +58,14 @@ where
state_path: PathBuf,
}
impl<I, T, R, S> ChatApp<I, T, R, S>
impl<I, T, R> ChatApp<I, T, R>
where
I: IdentityProvider + Send,
T: Transport,
R: RegistrationService + Send + 'static,
S: ChatStore + Send,
{
pub fn new(
client: ChatClient<I, T, R, S>,
client: ChatClient<I, T, R>,
events: Receiver<Event>,
user_name: &str,
data_dir: &Path,

10
bin/chat-cli/src/main.rs
View File

@ -9,7 +9,7 @@ use anyhow::{Context, Result};
use clap::{Parser, ValueEnum};
use crossbeam_channel::Receiver;
use logos_chat::{
ChatClient, ChatStore, Event, IdentityProvider, LogosChatClient, RegistrationService, Transport,
ChatClient, Event, IdentityProvider, LogosChatClient, RegistrationService, Transport,
};
use components::{EmbeddedP2pDeliveryService, P2pConfig};
@ -134,8 +134,8 @@ fn run<T: Transport>(transport: T, cli: &Cli) -> Result<()> {
launch_tui(client, events, cli)
}
fn launch_tui<I, T, R, S>(
client: ChatClient<I, T, R, S>,
fn launch_tui<I, T, R>(
client: ChatClient<I, T, R>,
events: Receiver<Event>,
cli: &Cli,
) -> Result<()>
@ -143,7 +143,6 @@ where
I: IdentityProvider + Send,
T: Transport,
R: RegistrationService + Send + 'static,
S: ChatStore + Send,
{
let mut app = ChatApp::new(client, events, &cli.name, &cli.data)?;
@ -157,12 +156,11 @@ where
result
}
fn run_app<I, T, R, S>(terminal: &mut ui::Tui, app: &mut ChatApp<I, T, R, S>) -> Result<()>
fn run_app<I, T, R>(terminal: &mut ui::Tui, app: &mut ChatApp<I, T, R>) -> Result<()>
where
I: IdentityProvider + Send,
T: Transport,
R: RegistrationService + Send + 'static,
S: ChatStore + Send,
{
loop {
app.process_incoming()?;

20
bin/chat-cli/src/ui.rs
View File

@ -16,7 +16,7 @@ use ratatui::{
widgets::{Block, Borders, List, ListItem, Paragraph, Wrap},
};
use logos_chat::{ChatStore, IdentityProvider, RegistrationService, Transport};
use logos_chat::{IdentityProvider, RegistrationService, Transport};
use crate::app::ChatApp;
@ -38,12 +38,11 @@ pub fn restore() -> io::Result<()> {
}
/// Draw the UI.
pub fn draw<I, D, R, S>(frame: &mut Frame, app: &ChatApp<I, D, R, S>)
pub fn draw<I, D, R>(frame: &mut Frame, app: &ChatApp<I, D, R>)
where
I: IdentityProvider + Send + 'static,
D: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
let chunks = Layout::default()
.direction(Direction::Vertical)
@ -61,12 +60,11 @@ where
draw_status(frame, app, chunks[3]);
}
fn draw_header<I, D, R, S>(frame: &mut Frame, app: &ChatApp<I, D, R, S>, area: Rect)
fn draw_header<I, D, R>(frame: &mut Frame, app: &ChatApp<I, D, R>, area: Rect)
where
I: IdentityProvider + Send + 'static,
D: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
let title = match app.current_session() {
Some(session) => {
@ -90,12 +88,11 @@ where
frame.render_widget(header, area);
}
fn draw_messages<I, D, R, S>(frame: &mut Frame, app: &ChatApp<I, D, R, S>, area: Rect)
fn draw_messages<I, D, R>(frame: &mut Frame, app: &ChatApp<I, D, R>, area: Rect)
where
I: IdentityProvider + Send + 'static,
D: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
let remote_name = app
.current_session()
@ -182,12 +179,11 @@ where
frame.render_stateful_widget(messages_widget, area, &mut list_state);
}
fn draw_input<I, D, R, S>(frame: &mut Frame, app: &ChatApp<I, D, R, S>, area: Rect)
fn draw_input<I, D, R>(frame: &mut Frame, app: &ChatApp<I, D, R>, area: Rect)
where
I: IdentityProvider + Send + 'static,
D: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
// Inner width: area minus borders (2).
let inner_width = area.width.saturating_sub(2) as usize;
@ -215,12 +211,11 @@ where
frame.set_cursor_position((cursor_x, area.y + 1));
}
fn draw_status<I, D, R, S>(frame: &mut Frame, app: &ChatApp<I, D, R, S>, area: Rect)
fn draw_status<I, D, R>(frame: &mut Frame, app: &ChatApp<I, D, R>, area: Rect)
where
I: IdentityProvider + Send + 'static,
D: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
let status = Paragraph::new(app.status.as_str())
.style(Style::default().fg(Color::Gray))
@ -231,12 +226,11 @@ where
}
/// Handle keyboard events.
pub fn handle_events<I, D, R, S>(app: &mut ChatApp<I, D, R, S>) -> io::Result<bool>
pub fn handle_events<I, D, R>(app: &mut ChatApp<I, D, R>) -> io::Result<bool>
where
I: IdentityProvider + Send + 'static,
D: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
// Poll for events with a short timeout to allow checking incoming messages
if event::poll(std::time::Duration::from_millis(100))?

4
core/account/src/account.rs
View File

@ -6,6 +6,10 @@ use libchat::IdentityProvider;
/// A Test Focused LogosAccount using a pre-defined identifier.
/// The test account is not persisted, and uses a single user provided id.
/// This account type should not be used in a production system.
///
/// `Clone` lets a test reuse the same identity across a simulated restart (a
/// fresh `new` would mint a new random key).
#[derive(Clone)]
pub struct TestLogosAccount {
id: IdentId,
signing_key: Ed25519SigningKey,

45
core/conversations/src/conversation/group_v1.rs
View File

@ -14,7 +14,7 @@ use tracing::debug;
use crate::account_directory::{AccountDirectory, resolve_device_ids};
use crate::conversation::ConversationIdRef;
use crate::inbox_v2::MlsProvider;
use crate::inbox_v2::{MlsPqProvider, MlsProvider};
use crate::service_context::{ExternalServices, ServiceContext};
use crate::utils::{blake2b_hex, hash_size};
@ -49,7 +49,7 @@ impl GroupV1Convo {
pub fn new<S: ExternalServices>(cx: &mut ServiceContext<S>) -> Result<Self, ChatError> {
let config = Self::mls_create_config(cx);
let mls_group = MlsGroup::new(
&cx.mls_provider,
&cx.mls_provider(),
&cx.mls_identity,
&config,
cx.mls_identity.get_credential(),
@ -70,13 +70,16 @@ impl GroupV1Convo {
cx: &mut ServiceContext<S>,
welcome: Welcome,
) -> Result<Self, ChatError> {
let mls_group =
StagedWelcome::build_from_welcome(&cx.mls_provider, &Self::mls_join_config(), welcome)
.unwrap()
.build()
.unwrap()
.into_group(&cx.mls_provider)
.unwrap();
let mls_group = StagedWelcome::build_from_welcome(
&cx.mls_provider(),
&Self::mls_join_config(),
welcome,
)
.unwrap()
.build()
.unwrap()
.into_group(&cx.mls_provider())
.unwrap();
let convo_id = hex::encode(mls_group.group_id().as_slice());
Self::subscribe(&mut cx.ds, &convo_id)?;
@ -93,7 +96,7 @@ impl GroupV1Convo {
convo_id: String,
group_id: GroupId,
) -> Result<Self, ChatError> {
let mls_group = MlsGroup::load(cx.mls_provider.storage(), &group_id)
let mls_group = MlsGroup::load(cx.mls_provider().storage(), &group_id)
.map_err(ChatError::generic)?
.ok_or_else(|| ChatError::NoConvo("mls group not found".into()))?;
@ -116,7 +119,7 @@ impl GroupV1Convo {
fn mls_create_config<S: ExternalServices>(cx: &mut ServiceContext<S>) -> MlsGroupCreateConfig {
MlsGroupCreateConfig::builder()
.ciphersuite(cx.mls_provider.crypto().supported_ciphersuites()[0])
.ciphersuite(cx.mls_provider().crypto().supported_ciphersuites()[0])
.use_ratchet_tree_extension(true) // This is handy for now, until there is central store for this data
.build()
}
@ -182,7 +185,7 @@ impl GroupV1Convo {
let mls_message_out = self
.mls_group
.create_message(&cx.mls_provider, &cx.mls_identity, &wire)
.create_message(&cx.mls_provider(), &cx.mls_identity, &wire)
.unwrap();
let msg_bytes = mls_message_out.to_bytes().unwrap();
@ -271,7 +274,7 @@ impl<S: ExternalServices> Convo<S> for GroupV1Convo {
let processed = self
.mls_group
.process_message(&cx.mls_provider, protocol_message)
.process_message(&cx.mls_provider(), protocol_message)
.map_err(ChatError::generic)?;
let cred_bytes = processed.credential().serialized_content().to_vec();
@ -287,7 +290,7 @@ impl<S: ExternalServices> Convo<S> for GroupV1Convo {
}
ProcessedMessageContent::StagedCommitMessage(commit) => {
self.mls_group
.merge_staged_commit(&cx.mls_provider, *commit)
.merge_staged_commit(&cx.mls_provider(), *commit)
.map_err(ChatError::generic)?;
None
}
@ -329,25 +332,31 @@ impl<S: ExternalServices> GroupConvo<S> for GroupV1Convo {
// leaf, so all of a user's installations join the group.
let mut keypkgs = Vec::with_capacity(members.len());
for ident in members {
keypkgs.extend(self.key_packages_for_account(ident, &cx.mls_provider, &cx.registry)?);
keypkgs.extend(self.key_packages_for_account(
ident,
&cx.mls_provider(),
&cx.registry,
)?);
}
let (commit, welcome, _group_info) = self
.mls_group
.add_members(
&cx.mls_provider,
&cx.mls_provider(),
&cx.mls_identity,
keypkgs.iter().as_slice(),
)
.unwrap();
self.mls_group
.merge_pending_commit(&cx.mls_provider)
.merge_pending_commit(&cx.mls_provider())
.unwrap();
// TODO: (P3) Evaluate privacy/performance implications of an aggregated Welcome for multiple users
for account_id in members {
cx.mls_provider
// Built inline (not via `cx.mls_provider()`) so the immutable store
// borrow and the `&mut cx.ds` publish borrow stay disjoint.
MlsPqProvider::new(&cx.crypto, &cx.store)
.invite_user(&mut cx.ds, account_id, &welcome)?;
}

16
core/conversations/src/conversation/group_v2.rs
View File

@ -100,7 +100,7 @@ impl GroupV2Convo {
let conversation = Conversation::create(
&convo_id,
&member_id(service_ctx),
&service_ctx.mls_provider,
&service_ctx.mls_provider(),
service_ctx.mls_identity.get_credential(),
CIPHER_SUITE,
&service_ctx.mls_identity,
@ -131,7 +131,7 @@ impl GroupV2Convo {
) -> Result<Self, ChatError> {
let Some(conv) = Conversation::join(
&member_id(service_ctx),
&service_ctx.mls_provider,
&service_ctx.mls_provider(),
&service_ctx.mls_identity,
&welcome.welcome_bytes,
&welcome.conversation_sync_bytes,
@ -198,7 +198,7 @@ where
content: &[u8],
) -> Result<(), ChatError> {
self.conversation.send_message(
&service_ctx.mls_provider,
&service_ctx.mls_provider(),
&service_ctx.mls_identity,
content.to_vec(),
)?;
@ -225,13 +225,13 @@ where
};
self.conversation.process_inbound(
&service_ctx.mls_provider,
&service_ctx.mls_provider(),
&service_ctx.mls_identity,
&frame.sender_app_id,
&inner,
)?;
self.conversation
.poll(&service_ctx.mls_provider, &service_ctx.mls_identity);
.poll(&service_ctx.mls_provider(), &service_ctx.mls_identity);
let events = self.after_op(service_ctx)?; // route + publish + re-arm, returns events
match self.events_to_content(&events) {
@ -247,7 +247,9 @@ where
fn wakeup(&mut self, ctx: &mut ServiceContext<S>) -> Result<(), ChatError> {
info!(convo = %self.convo_id, "Wakeup");
let outcome = self.conversation.poll(&ctx.mls_provider, &ctx.mls_identity);
let outcome = self
.conversation
.poll(&ctx.mls_provider(), &ctx.mls_identity);
if outcome.leave_requested {
// Commit ejected us (or join expired). Real handling - drops
// this convo from its map;
@ -280,7 +282,7 @@ where
self.pending_invites
.push(member.as_str().as_bytes().to_vec());
self.conversation.add_member(
&service_ctx.mls_provider,
&service_ctx.mls_provider(),
&service_ctx.mls_identity,
member.as_str().as_bytes(),
&kp_bytes,

16
core/conversations/src/core.rs
View File

@ -8,12 +8,14 @@ use crate::{
conversation::{Convo, GroupConvo},
errors::ChatError,
inbox::Inbox,
inbox_v2::{InboxV2, MlsEphemeralPqProvider, MlsIdentityProvider},
inbox_v2::{InboxV2, MlsIdentityProvider},
outcomes::{ConvoOutcome, InboxOutcome, PayloadOutcome},
proto::{EncryptedPayload, EnvelopeV1, Message},
};
use crypto::{Identity, PublicKey};
use openmls::group::GroupId;
use openmls_libcrux_crypto::CryptoProvider as LibcruxCryptoProvider;
use openmls_traits::storage::{CURRENT_VERSION, StorageProvider};
use shared_traits::IdentIdRef;
use std::collections::HashMap;
use std::fmt::Debug;
@ -37,15 +39,17 @@ pub struct Core<S: ExternalServices> {
cached_convos: HashMap<String, ConvoTypeOwned<S>>,
}
// 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.
// Constructors live on the `(IP, DS, RS, WS, CS)` tuple form: `S` can't be
// inferred backwards through `S::DS`, so the bundle is built from the args here.
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,
CS: ChatStore
+ StorageProvider<CURRENT_VERSION, Error: std::error::Error + Send + Sync>
+ 'static,
{
/// Opens or creates a `Core` with the given storage configuration.
///
@ -114,7 +118,7 @@ where
let inbox = Inbox::new(&identity);
let ident_id = ident.id().clone();
let mls_identity = MlsIdentityProvider::new(ident);
let mls_provider = MlsEphemeralPqProvider::new().map_err(ChatError::generic)?;
let crypto = LibcruxCryptoProvider::new().map_err(ChatError::generic)?;
let causal = CausalHistoryStore::new();
let pq_inbox = InboxV2::new(ident_id);
@ -132,7 +136,7 @@ where
registry: registration,
store,
mls_identity,
mls_provider,
crypto,
causal,
identity,
wakeup_service,

4
core/conversations/src/inbox_v2.rs
View File

@ -12,7 +12,7 @@ use tracing::info;
use tracing::instrument;
pub use identity::MlsIdentityProvider;
pub(crate) use mls_provider::MlsEphemeralPqProvider;
pub(crate) use mls_provider::MlsPqProvider;
use crate::ChatError;
use crate::DeliveryService;
@ -195,7 +195,7 @@ impl InboxV2 {
.leaf_node_capabilities(capabilities)
.build(
CIPHER_SUITE,
&cx.mls_provider,
&cx.mls_provider(),
&cx.mls_identity,
cx.mls_identity.get_credential(),
)

36
core/conversations/src/inbox_v2/mls_provider.rs
View File

@ -1,8 +1,7 @@
use openmls::framing::MlsMessageOut;
use openmls_libcrux_crypto::CryptoProvider as LibcruxCryptoProvider;
use openmls_memory_storage::MemoryStorage;
use openmls_traits::OpenMlsProvider;
use openmls_traits::types::CryptoError;
use openmls_traits::storage::{CURRENT_VERSION, StorageProvider};
use prost::Message;
use shared_traits::IdentIdRef;
@ -13,22 +12,23 @@ use super::{
conversation_id_for, delivery_address_for,
};
/// This is a Post-Quantum based MLS provider with in memory storage
pub struct MlsEphemeralPqProvider {
crypto: LibcruxCryptoProvider,
storage: MemoryStorage,
/// Post-Quantum MLS provider: a transient view pairing the libcrux crypto/RNG
/// backend with an OpenMLS [`StorageProvider`], both borrowed. Holding borrows
/// rather than owning lets one store instance serve MLS operations while it is
/// separately owned and mutated as the chat store. Crypto/RNG are always libcrux
/// (PQ).
pub struct MlsPqProvider<'a, St: StorageProvider<CURRENT_VERSION>> {
crypto: &'a LibcruxCryptoProvider,
storage: &'a St,
}
impl MlsEphemeralPqProvider {
pub fn new() -> Result<Self, CryptoError> {
let crypto = LibcruxCryptoProvider::new()?;
let storage = MemoryStorage::default();
Ok(Self { crypto, storage })
impl<'a, St: StorageProvider<CURRENT_VERSION>> MlsPqProvider<'a, St> {
pub fn new(crypto: &'a LibcruxCryptoProvider, storage: &'a St) -> Self {
Self { crypto, storage }
}
}
impl MlsProvider for MlsEphemeralPqProvider {
impl<St: StorageProvider<CURRENT_VERSION>> MlsProvider for MlsPqProvider<'_, St> {
fn invite_user<DS: DeliveryService>(
&self,
ds: &mut DS,
@ -59,20 +59,20 @@ impl MlsProvider for MlsEphemeralPqProvider {
}
}
impl OpenMlsProvider for MlsEphemeralPqProvider {
impl<St: StorageProvider<CURRENT_VERSION>> OpenMlsProvider for MlsPqProvider<'_, St> {
type CryptoProvider = LibcruxCryptoProvider;
type RandProvider = LibcruxCryptoProvider;
type StorageProvider = openmls_memory_storage::MemoryStorage;
type StorageProvider = St;
fn storage(&self) -> &Self::StorageProvider {
&self.storage
self.storage
}
fn crypto(&self) -> &Self::CryptoProvider {
&self.crypto
self.crypto
}
fn rand(&self) -> &Self::RandProvider {
&self.crypto
self.crypto
}
}

12
core/conversations/src/lib.rs
View File

@ -20,6 +20,7 @@ pub use account_directory::{
};
pub use causal_history::{Frontier, MissingMessage};
pub use chat_sqlite::ChatStorage;
pub use chat_sqlite::MlsStorageError;
pub use chat_sqlite::StorageConfig;
pub use core::{ConversationId, Core, Introduction};
pub use errors::ChatError;
@ -32,3 +33,14 @@ pub use shared_traits::{IdentId, IdentIdRef, IdentityProvider};
pub use storage::{ChatStore, ConversationKind};
pub use types::AddressedEnvelope;
pub use utils::{hex_trunc, trunc};
/// OpenMLS storage requirements, re-exported so external providers can implement
/// a durable [`StorageProvider`](openmls_traits::storage::StorageProvider)
/// without depending on `openmls_traits` directly. [`ChatStorage`] is libchat's
/// own durable implementation, and [`ChatStore`] folds this surface in so one
/// store type serves both chat and MLS state.
pub mod mls_storage {
pub use openmls_memory_storage::MemoryStorage;
pub use openmls_traits::OpenMlsProvider;
pub use openmls_traits::storage::{CURRENT_VERSION, Entity, Key, StorageProvider, traits};
}

44
core/conversations/src/service_context.rs
View File

@ -1,22 +1,34 @@
//! Bundles the services a conversation operation needs into one [`ServiceContext`].
use crypto::Identity;
use openmls_libcrux_crypto::CryptoProvider as LibcruxCryptoProvider;
use openmls_traits::storage::{CURRENT_VERSION, StorageProvider};
use storage::ChatStore;
use crate::IdentityProvider;
use crate::causal_history::CausalHistoryStore;
use crate::inbox_v2::{MlsEphemeralPqProvider, MlsIdentityProvider};
use crate::inbox_v2::{MlsIdentityProvider, MlsPqProvider};
use crate::service_traits::WakeupService;
use crate::{DeliveryService, RegistrationService};
/// Bundles the external service types (`DS`, `RS`, `CS`) behind one `S`. The
/// `(DS, RS, CS)` tuple impl lets them still be supplied separately.
/// Bundles the external service types behind one `S`; the `(IP, DS, RS, WS, CS)`
/// tuple impl lets them still be supplied separately. `CS` is the single durable
/// store: [`ChatStore`] subsumes the OpenMLS storage surface, so chat state and
/// MLS group state share one type.
///
/// The extra `StorageProvider` bound pins the store's OpenMLS error to a
/// thread-safe, `'static` `std::error::Error`, which GroupV2/de_mls needs.
/// It's stated here (not on `ChatStore`) because an associated-type bound doesn't
/// elaborate through a supertrait, whereas one on this associated type reaches
/// every `S: ExternalServices` consumer.
pub trait ExternalServices {
type IP: IdentityProvider;
type DS: DeliveryService;
type RS: RegistrationService;
type WS: WakeupService;
type CS: ChatStore;
type CS: ChatStore
+ StorageProvider<CURRENT_VERSION, Error: std::error::Error + Send + Sync>
+ 'static;
}
impl<IP, DS, RS, WS, CS> ExternalServices for (IP, DS, RS, WS, CS)
@ -25,7 +37,9 @@ where
DS: DeliveryService,
RS: RegistrationService,
WS: WakeupService,
CS: ChatStore,
CS: ChatStore
+ StorageProvider<CURRENT_VERSION, Error: std::error::Error + Send + Sync>
+ 'static,
{
type IP = IP;
type DS = DS;
@ -40,12 +54,20 @@ pub(crate) struct ServiceContext<S: ExternalServices> {
pub(crate) registry: S::RS,
pub(crate) store: S::CS,
pub(crate) mls_identity: MlsIdentityProvider<S::IP>,
pub(crate) mls_provider: MlsEphemeralPqProvider,
pub(crate) crypto: LibcruxCryptoProvider,
pub(crate) causal: CausalHistoryStore,
pub(crate) identity: Identity,
pub(crate) wakeup_service: S::WS,
}
impl<S: ExternalServices> ServiceContext<S> {
/// A transient MLS provider over the shared chat store and the long-lived
/// crypto backend. Rebuilt per call so `store` stays singly owned.
pub(crate) fn mls_provider(&self) -> MlsPqProvider<'_, S::CS> {
MlsPqProvider::new(&self.crypto, &self.store)
}
}
#[cfg(test)]
mod test_support {
use super::*;
@ -118,8 +140,12 @@ mod test_support {
fn wakeup_in(&mut self, _: std::time::Duration, _: crate::ConversationId) {}
}
impl<IP: IdentityProvider, CS: ChatStore>
ServiceContext<(IP, NoopDelivery, NoopRegistration, NoopWakeups, CS)>
impl<IP, CS> ServiceContext<(IP, NoopDelivery, NoopRegistration, NoopWakeups, CS)>
where
IP: IdentityProvider,
CS: ChatStore
+ StorageProvider<CURRENT_VERSION, Error: std::error::Error + Send + Sync>
+ 'static,
{
/// Builds a context around a real store, stubbing other services.
pub(crate) fn for_test(ident: IP, store: CS) -> Result<Self, ChatError> {
@ -129,7 +155,7 @@ mod test_support {
registry: NoopRegistration,
store,
mls_identity: MlsIdentityProvider::new(ident),
mls_provider: MlsEphemeralPqProvider::new().map_err(ChatError::generic)?,
crypto: LibcruxCryptoProvider::new().map_err(ChatError::generic)?,
causal: CausalHistoryStore::new(),
identity: Identity::new(name),
wakeup_service: NoopWakeups {},

17
core/integration_tests_core/src/test_client.rs
View File

@ -1,3 +1,4 @@
use chat_sqlite::ChatStorage;
use libchat::{ConversationId, Core, IdentityProvider, PayloadOutcome};
use logos_account::TestLogosAccount;
use shared_traits::IdentId;
@ -7,7 +8,7 @@ use std::ops::{Deref, DerefMut};
use std::time::Duration;
use tracing::{info, warn};
use components::{EphemeralRegistry, LocalBroadcaster, MemStore};
use components::{EphemeralRegistry, LocalBroadcaster};
use crate::wakeup::{TestWakeupProvider, TestWakeupService, WakeupRecord};
@ -21,13 +22,12 @@ const RAYA: usize = 1;
const PAX: usize = 2;
const MIRA: usize = 3;
// type ClientType = CoreClient<TestLogosAccount, LocalBroadcaster, EphemeralRegistry, WP, MemStore>;
type ClientType = Core<(
TestLogosAccount,
LocalBroadcaster,
EphemeralRegistry,
WP,
MemStore,
ChatStorage,
)>;
#[derive(Debug)]
@ -154,9 +154,14 @@ impl<const N: usize> TestHarness<N> {
let ident = TestLogosAccount::new(Self::names(i));
addresses.insert(i, ident.id().clone());
let core_client =
ClientType::new_with_name(ident, ds.clone(), rs.clone(), wp, MemStore::new())
.unwrap();
let core_client = ClientType::new_with_name(
ident,
ds.clone(),
rs.clone(),
wp,
ChatStorage::in_memory(),
)
.unwrap();
let client = TestClient::init(core_client);

13
core/integration_tests_core/tests/causal_history.rs
View File

@ -6,7 +6,8 @@
use std::ops::{Deref, DerefMut};
use components::{EphemeralRegistry, LocalBroadcaster, MemStore};
use chat_sqlite::ChatStorage;
use components::{EphemeralRegistry, LocalBroadcaster};
use libchat::{Core, MissingMessage, WakeupService};
use logos_account::TestLogosAccount;
@ -22,7 +23,7 @@ struct Client {
LocalBroadcaster,
EphemeralRegistry,
NoopWakeupService,
MemStore,
ChatStorage,
)>,
}
@ -33,7 +34,7 @@ impl Client {
LocalBroadcaster,
EphemeralRegistry,
NoopWakeupService,
MemStore,
ChatStorage,
)>,
) -> Self {
Client { inner: core }
@ -64,7 +65,7 @@ impl Deref for Client {
LocalBroadcaster,
EphemeralRegistry,
NoopWakeupService,
MemStore,
ChatStorage,
)>;
fn deref(&self) -> &Self::Target {
&self.inner
@ -88,7 +89,7 @@ fn missing_group_message_is_detected() {
ds.new_consumer(),
rs.clone(),
NoopWakeupService {},
MemStore::new(),
ChatStorage::in_memory(),
)
.unwrap();
@ -98,7 +99,7 @@ fn missing_group_message_is_detected() {
ds.clone(),
rs.clone(),
NoopWakeupService {},
MemStore::new(),
ChatStorage::in_memory(),
)
.unwrap();

106
core/integration_tests_core/tests/mls_persistence.rs Normal file
View File

@ -0,0 +1,106 @@
//! Proves a GroupV1 conversation survives a full `Core` restart once MLS state
//! is backed by the durable SQLite `StorageProvider` (issue #112).
//!
//! Saro and Raya exchange over a group; Saro's `Core` is dropped and rebuilt
//! against the same DB files and identity; Saro sends again and Raya still
//! receives it. The post-restart send rehydrates the MLS group via
//! `MlsGroup::load` from durable storage — with the in-memory provider the
//! group would be gone and the send would fail.
use chat_sqlite::{ChatStorage, StorageConfig};
use components::{EphemeralRegistry, LocalBroadcaster};
use libchat::{ConvoOutcome, Core, PayloadOutcome, WakeupService};
use logos_account::TestLogosAccount;
#[derive(Debug)]
struct NoopWakeupService {}
impl WakeupService for NoopWakeupService {
fn wakeup_in(&mut self, _: std::time::Duration, _: libchat::ConversationId) {}
}
type TestCore = Core<(
TestLogosAccount,
LocalBroadcaster,
EphemeralRegistry,
NoopWakeupService,
ChatStorage,
)>;
/// Builds a `Core` whose single store holds both the conversation metadata and
/// the MLS group state in one on-disk DB, so both persist across a restart.
fn build(
account: TestLogosAccount,
ds: LocalBroadcaster,
rs: EphemeralRegistry,
db_path: &str,
) -> TestCore {
let chat = ChatStorage::new(StorageConfig::File(db_path.to_string())).unwrap();
Core::new_with_name(account, ds, rs, NoopWakeupService {}, chat).unwrap()
}
/// Drains everything queued for `core`, returning the bytes of any received
/// conversation messages.
fn drain(core: &mut TestCore) -> Vec<Vec<u8>> {
let payloads: Vec<_> = {
let ds = core.ds();
std::iter::from_fn(|| ds.poll()).collect()
};
let mut received = vec![];
for data in payloads {
if let PayloadOutcome::Convo(ConvoOutcome {
content: Some(content),
..
}) = core.handle_payload(&data).unwrap()
{
received.push(content.bytes);
}
}
received
}
#[test]
fn group_v1_resumes_after_core_restart() {
let dir = tempfile::tempdir().unwrap();
let saro_db = dir.path().join("saro.db").to_string_lossy().into_owned();
let raya_db = dir.path().join("raya.db").to_string_lossy().into_owned();
let ds = LocalBroadcaster::new();
let rs = EphemeralRegistry::new();
// Cloned so Saro keeps the same identity across the restart (delegate
// persistence is a separate concern; here we hold identity fixed to isolate
// the MLS-storage contribution).
let saro_account = TestLogosAccount::new("saro");
let raya_account = TestLogosAccount::new("raya");
let mut saro = build(
saro_account.clone(),
ds.new_consumer(),
rs.clone(),
&saro_db,
);
let mut raya = build(raya_account, ds.new_consumer(), rs.clone(), &raya_db);
// Saro creates a group with Raya; Raya processes the Welcome and joins.
let raya_id = raya.ident_id().clone();
let convo_id = saro.create_group_convo_v1(&[&raya_id]).unwrap().to_string();
drain(&mut raya);
saro.send_content(&convo_id, b"before restart").unwrap();
assert_eq!(drain(&mut raya), vec![b"before restart".to_vec()]);
// Restart Saro: drop the Core, rebuild against the same DB file and
// identity. The only thing carrying the group forward is the MLS state
// persisted in `saro.db`.
drop(saro);
let mut saro = build(saro_account, ds.new_consumer(), rs.clone(), &saro_db);
// The send rehydrates the group via `MlsGroup::load` from durable storage.
saro.send_content(&convo_id, b"after restart").unwrap();
assert_eq!(
drain(&mut raya),
vec![b"after restart".to_vec()],
"Raya must receive Saro's post-restart message: the MLS group resumed \
from durable SQLite storage"
);
}

4
core/sqlite/Cargo.toml
View File

@ -11,9 +11,13 @@ storage = { workspace = true }
# External dependencies (sorted)
hex = "0.4.3"
openmls_traits = "0.5.0"
rusqlite = { version = "0.35", features = ["bundled-sqlcipher-vendored-openssl"] }
serde_json = "1.0"
thiserror = "2"
zeroize = { version = "1.8.2", features = ["derive"] }
[dev-dependencies]
# External dependencies (sorted)
serde = { version = "1.0", features = ["derive"] }
tempfile = "3"

2
core/sqlite/src/lib.rs
View File

@ -3,6 +3,7 @@
mod common;
mod errors;
mod migrations;
pub mod mls_storage;
mod types;
use std::collections::HashSet;
@ -22,6 +23,7 @@ use crate::{
};
pub use common::StorageConfig;
pub use mls_storage::MlsStorageError;
/// Chat-specific storage operations.
///

4
core/sqlite/src/migrations.rs
View File

@ -19,6 +19,10 @@ pub fn get_migrations() -> Vec<(&'static str, &'static str)> {
"002_ratchet_state",
include_str!("migrations/002_ratchet_state.sql"),
),
(
"003_mls_storage",
include_str!("migrations/003_mls_storage.sql"),
),
]
}

9
core/sqlite/src/migrations/003_mls_storage.sql Normal file
View File

@ -0,0 +1,9 @@
-- Backing table for the OpenMLS StorageProvider (SqliteMlsStorage).
--
-- A byte-faithful mirror of openmls_memory_storage's HashMap<Vec<u8>, Vec<u8>>:
-- `key` is `label ++ serde_json(logical_key) ++ version_be`, `value` is the
-- serde_json blob (a single value, or a JSON array for the two list labels).
CREATE TABLE IF NOT EXISTS mls_kv (
key BLOB PRIMARY KEY,
value BLOB NOT NULL
);

925
core/sqlite/src/mls_storage.rs Normal file
View File

@ -0,0 +1,925 @@
//! The MLS half of [`ChatStorage`](crate::ChatStorage): its OpenMLS
//! [`StorageProvider`] impl, backed by a `mls_kv(key, value)` table in the same
//! SQLite database as the chat schema.
//!
//! A byte-faithful port of `openmls_memory_storage::MemoryStorage`: the same
//! `label ++ serde_json(logical_key) ++ version_be` key derivation and JSON
//! value encoding, but persisted to the table instead of a `HashMap`, so an MLS
//! group's state survives process restarts (`MlsGroup::load` reads it back)
//! where the in-memory provider cannot.
//!
//! Two deviations from the reference impl, both deliberate:
//! - decode paths return `Err` instead of `unwrap`ing, since disk bytes can be
//! corrupt where an in-memory map cannot;
//! - `clear_proposal_queue` deletes queued proposals with the same composite
//! key they were written under (the reference impl uses a bare key and
//! orphans them).
use openmls_traits::storage::*;
use rusqlite::{OptionalExtension, params};
use crate::ChatStorage;
/// Errors surfaced by [`ChatStorage`]'s MLS [`StorageProvider`] impl as its
/// `Error` type.
#[derive(Debug, thiserror::Error)]
pub enum MlsStorageError {
#[error("sqlite: {0}")]
Sqlite(#[from] rusqlite::Error),
#[error("codec: {0}")]
Codec(#[from] serde_json::Error),
#[error("inconsistent storage: {0}")]
Inconsistent(&'static str),
}
// Raw and typed key/value helpers backing the `StorageProvider` impl below. They
// run on the same connection as the chat schema; the trait's all-`&self` API
// maps onto rusqlite's `&self` methods directly, and a `Core` drives storage
// single-threaded so no external synchronization is needed.
impl ChatStorage {
fn conn(&self) -> &rusqlite::Connection {
self.db.connection()
}
fn put(&self, storage_key: &[u8], value: &[u8]) -> Result<(), MlsStorageError> {
self.conn().execute(
"INSERT OR REPLACE INTO mls_kv (key, value) VALUES (?1, ?2)",
params![storage_key, value],
)?;
Ok(())
}
fn get(&self, storage_key: &[u8]) -> Result<Option<Vec<u8>>, MlsStorageError> {
Ok(self
.conn()
.query_row(
"SELECT value FROM mls_kv WHERE key = ?1",
params![storage_key],
|row| row.get::<_, Vec<u8>>(0),
)
.optional()?)
}
fn del(&self, storage_key: &[u8]) -> Result<(), MlsStorageError> {
self.conn()
.execute("DELETE FROM mls_kv WHERE key = ?1", params![storage_key])?;
Ok(())
}
// --- typed helpers mirroring MemoryStorage's private helpers ---
fn write<const V: u16>(
&self,
label: &[u8],
key: &[u8],
value: Vec<u8>,
) -> Result<(), MlsStorageError> {
self.put(&build_key_from_vec::<V>(label, key.to_vec()), &value)
}
fn read<const V: u16, E: Entity<V>>(
&self,
label: &[u8],
key: &[u8],
) -> Result<Option<E>, MlsStorageError> {
match self.get(&build_key_from_vec::<V>(label, key.to_vec()))? {
Some(bytes) => Ok(Some(serde_json::from_slice(&bytes)?)),
None => Ok(None),
}
}
fn read_list<const V: u16, E: Entity<V>>(
&self,
label: &[u8],
key: &[u8],
) -> Result<Vec<E>, MlsStorageError> {
let list = self.read_raw_list::<V>(label, key)?;
list.iter()
.map(|bytes| serde_json::from_slice(bytes))
.collect::<Result<Vec<E>, _>>()
.map_err(MlsStorageError::from)
}
/// The raw JSON-array-of-blobs behind a list key (empty when absent).
fn read_raw_list<const V: u16>(
&self,
label: &[u8],
key: &[u8],
) -> Result<Vec<Vec<u8>>, MlsStorageError> {
match self.get(&build_key_from_vec::<V>(label, key.to_vec()))? {
Some(bytes) => Ok(serde_json::from_slice(&bytes)?),
None => Ok(vec![]),
}
}
fn append<const V: u16>(
&self,
label: &[u8],
key: &[u8],
value: Vec<u8>,
) -> Result<(), MlsStorageError> {
let storage_key = build_key_from_vec::<V>(label, key.to_vec());
let tx = self.conn().unchecked_transaction()?;
let mut list = load_list(&tx, &storage_key)?;
list.push(value);
store_list(&tx, &storage_key, &list)?;
tx.commit()?;
Ok(())
}
fn remove_item<const V: u16>(
&self,
label: &[u8],
key: &[u8],
value: Vec<u8>,
) -> Result<(), MlsStorageError> {
let storage_key = build_key_from_vec::<V>(label, key.to_vec());
let tx = self.conn().unchecked_transaction()?;
let mut list = load_list(&tx, &storage_key)?;
if let Some(pos) = list.iter().position(|item| item == &value) {
list.remove(pos);
}
store_list(&tx, &storage_key, &list)?;
tx.commit()?;
Ok(())
}
fn delete<const V: u16>(&self, label: &[u8], key: &[u8]) -> Result<(), MlsStorageError> {
self.del(&build_key_from_vec::<V>(label, key.to_vec()))
}
}
fn build_key_from_vec<const V: u16>(label: &[u8], key: Vec<u8>) -> Vec<u8> {
let mut out = label.to_vec();
out.extend_from_slice(&key);
out.extend_from_slice(&u16::to_be_bytes(V));
out
}
fn epoch_key_pairs_id(
group_id: &impl traits::GroupId<CURRENT_VERSION>,
epoch: &impl traits::EpochKey<CURRENT_VERSION>,
leaf_index: u32,
) -> Result<Vec<u8>, MlsStorageError> {
let mut key = serde_json::to_vec(group_id)?;
key.extend_from_slice(&serde_json::to_vec(epoch)?);
key.extend_from_slice(&serde_json::to_vec(&leaf_index)?);
Ok(key)
}
fn load_list(
tx: &rusqlite::Transaction<'_>,
storage_key: &[u8],
) -> Result<Vec<Vec<u8>>, MlsStorageError> {
let existing: Option<Vec<u8>> = tx
.query_row(
"SELECT value FROM mls_kv WHERE key = ?1",
params![storage_key],
|row| row.get::<_, Vec<u8>>(0),
)
.optional()?;
match existing {
Some(bytes) => Ok(serde_json::from_slice(&bytes)?),
None => Ok(vec![]),
}
}
fn store_list(
tx: &rusqlite::Transaction<'_>,
storage_key: &[u8],
list: &[Vec<u8>],
) -> Result<(), MlsStorageError> {
let encoded = serde_json::to_vec(list)?;
tx.execute(
"INSERT OR REPLACE INTO mls_kv (key, value) VALUES (?1, ?2)",
params![storage_key, encoded],
)?;
Ok(())
}
const KEY_PACKAGE_LABEL: &[u8] = b"KeyPackage";
const PSK_LABEL: &[u8] = b"Psk";
const ENCRYPTION_KEY_PAIR_LABEL: &[u8] = b"EncryptionKeyPair";
const SIGNATURE_KEY_PAIR_LABEL: &[u8] = b"SignatureKeyPair";
const EPOCH_KEY_PAIRS_LABEL: &[u8] = b"EpochKeyPairs";
const TREE_LABEL: &[u8] = b"Tree";
const GROUP_CONTEXT_LABEL: &[u8] = b"GroupContext";
const INTERIM_TRANSCRIPT_HASH_LABEL: &[u8] = b"InterimTranscriptHash";
const CONFIRMATION_TAG_LABEL: &[u8] = b"ConfirmationTag";
const JOIN_CONFIG_LABEL: &[u8] = b"MlsGroupJoinConfig";
const OWN_LEAF_NODES_LABEL: &[u8] = b"OwnLeafNodes";
const GROUP_STATE_LABEL: &[u8] = b"GroupState";
const QUEUED_PROPOSAL_LABEL: &[u8] = b"QueuedProposal";
const PROPOSAL_QUEUE_REFS_LABEL: &[u8] = b"ProposalQueueRefs";
const OWN_LEAF_NODE_INDEX_LABEL: &[u8] = b"OwnLeafNodeIndex";
const EPOCH_SECRETS_LABEL: &[u8] = b"EpochSecrets";
const RESUMPTION_PSK_STORE_LABEL: &[u8] = b"ResumptionPsk";
const MESSAGE_SECRETS_LABEL: &[u8] = b"MessageSecrets";
impl StorageProvider<CURRENT_VERSION> for ChatStorage {
type Error = MlsStorageError;
fn queue_proposal<
GroupId: traits::GroupId<CURRENT_VERSION>,
ProposalRef: traits::ProposalRef<CURRENT_VERSION>,
QueuedProposal: traits::QueuedProposal<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
proposal_ref: &ProposalRef,
proposal: &QueuedProposal,
) -> Result<(), Self::Error> {
let key = serde_json::to_vec(&(group_id, proposal_ref))?;
let value = serde_json::to_vec(proposal)?;
self.write::<CURRENT_VERSION>(QUEUED_PROPOSAL_LABEL, &key, value)?;
let key = serde_json::to_vec(group_id)?;
let value = serde_json::to_vec(proposal_ref)?;
self.append::<CURRENT_VERSION>(PROPOSAL_QUEUE_REFS_LABEL, &key, value)
}
fn write_tree<
GroupId: traits::GroupId<CURRENT_VERSION>,
TreeSync: traits::TreeSync<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
tree: &TreeSync,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
TREE_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(tree)?,
)
}
fn write_interim_transcript_hash<
GroupId: traits::GroupId<CURRENT_VERSION>,
InterimTranscriptHash: traits::InterimTranscriptHash<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
interim_transcript_hash: &InterimTranscriptHash,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
INTERIM_TRANSCRIPT_HASH_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(interim_transcript_hash)?,
)
}
fn write_context<
GroupId: traits::GroupId<CURRENT_VERSION>,
GroupContext: traits::GroupContext<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
group_context: &GroupContext,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
GROUP_CONTEXT_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(group_context)?,
)
}
fn write_confirmation_tag<
GroupId: traits::GroupId<CURRENT_VERSION>,
ConfirmationTag: traits::ConfirmationTag<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
confirmation_tag: &ConfirmationTag,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
CONFIRMATION_TAG_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(confirmation_tag)?,
)
}
fn write_signature_key_pair<
SignaturePublicKey: traits::SignaturePublicKey<CURRENT_VERSION>,
SignatureKeyPair: traits::SignatureKeyPair<CURRENT_VERSION>,
>(
&self,
public_key: &SignaturePublicKey,
signature_key_pair: &SignatureKeyPair,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
SIGNATURE_KEY_PAIR_LABEL,
&serde_json::to_vec(public_key)?,
serde_json::to_vec(signature_key_pair)?,
)
}
fn queued_proposal_refs<
GroupId: traits::GroupId<CURRENT_VERSION>,
ProposalRef: traits::ProposalRef<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Vec<ProposalRef>, Self::Error> {
self.read_list::<CURRENT_VERSION, _>(
PROPOSAL_QUEUE_REFS_LABEL,
&serde_json::to_vec(group_id)?,
)
}
fn queued_proposals<
GroupId: traits::GroupId<CURRENT_VERSION>,
ProposalRef: traits::ProposalRef<CURRENT_VERSION>,
QueuedProposal: traits::QueuedProposal<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Vec<(ProposalRef, QueuedProposal)>, Self::Error> {
let refs: Vec<ProposalRef> = self.read_list::<CURRENT_VERSION, _>(
PROPOSAL_QUEUE_REFS_LABEL,
&serde_json::to_vec(group_id)?,
)?;
refs.into_iter()
.map(|proposal_ref| {
let key = serde_json::to_vec(&(group_id, &proposal_ref))?;
let proposal = self
.read::<CURRENT_VERSION, _>(QUEUED_PROPOSAL_LABEL, &key)?
.ok_or(MlsStorageError::Inconsistent(
"queued proposal missing for stored reference",
))?;
Ok((proposal_ref, proposal))
})
.collect::<Result<Vec<_>, Self::Error>>()
}
fn tree<
GroupId: traits::GroupId<CURRENT_VERSION>,
TreeSync: traits::TreeSync<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<TreeSync>, Self::Error> {
self.read::<CURRENT_VERSION, _>(TREE_LABEL, &serde_json::to_vec(group_id)?)
}
fn group_context<
GroupId: traits::GroupId<CURRENT_VERSION>,
GroupContext: traits::GroupContext<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<GroupContext>, Self::Error> {
self.read::<CURRENT_VERSION, _>(GROUP_CONTEXT_LABEL, &serde_json::to_vec(group_id)?)
}
fn interim_transcript_hash<
GroupId: traits::GroupId<CURRENT_VERSION>,
InterimTranscriptHash: traits::InterimTranscriptHash<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<InterimTranscriptHash>, Self::Error> {
self.read::<CURRENT_VERSION, _>(
INTERIM_TRANSCRIPT_HASH_LABEL,
&serde_json::to_vec(group_id)?,
)
}
fn confirmation_tag<
GroupId: traits::GroupId<CURRENT_VERSION>,
ConfirmationTag: traits::ConfirmationTag<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<ConfirmationTag>, Self::Error> {
self.read::<CURRENT_VERSION, _>(CONFIRMATION_TAG_LABEL, &serde_json::to_vec(group_id)?)
}
fn signature_key_pair<
SignaturePublicKey: traits::SignaturePublicKey<CURRENT_VERSION>,
SignatureKeyPair: traits::SignatureKeyPair<CURRENT_VERSION>,
>(
&self,
public_key: &SignaturePublicKey,
) -> Result<Option<SignatureKeyPair>, Self::Error> {
self.read::<CURRENT_VERSION, _>(SIGNATURE_KEY_PAIR_LABEL, &serde_json::to_vec(public_key)?)
}
fn write_key_package<
HashReference: traits::HashReference<CURRENT_VERSION>,
KeyPackage: traits::KeyPackage<CURRENT_VERSION>,
>(
&self,
hash_ref: &HashReference,
key_package: &KeyPackage,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
KEY_PACKAGE_LABEL,
&serde_json::to_vec(hash_ref)?,
serde_json::to_vec(key_package)?,
)
}
fn write_psk<
PskId: traits::PskId<CURRENT_VERSION>,
PskBundle: traits::PskBundle<CURRENT_VERSION>,
>(
&self,
psk_id: &PskId,
psk: &PskBundle,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
PSK_LABEL,
&serde_json::to_vec(psk_id)?,
serde_json::to_vec(psk)?,
)
}
fn write_encryption_key_pair<
EncryptionKey: traits::EncryptionKey<CURRENT_VERSION>,
HpkeKeyPair: traits::HpkeKeyPair<CURRENT_VERSION>,
>(
&self,
public_key: &EncryptionKey,
key_pair: &HpkeKeyPair,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
ENCRYPTION_KEY_PAIR_LABEL,
&serde_json::to_vec(public_key)?,
serde_json::to_vec(key_pair)?,
)
}
fn key_package<
KeyPackageRef: traits::HashReference<CURRENT_VERSION>,
KeyPackage: traits::KeyPackage<CURRENT_VERSION>,
>(
&self,
hash_ref: &KeyPackageRef,
) -> Result<Option<KeyPackage>, Self::Error> {
self.read::<CURRENT_VERSION, _>(KEY_PACKAGE_LABEL, &serde_json::to_vec(hash_ref)?)
}
fn psk<PskBundle: traits::PskBundle<CURRENT_VERSION>, PskId: traits::PskId<CURRENT_VERSION>>(
&self,
psk_id: &PskId,
) -> Result<Option<PskBundle>, Self::Error> {
self.read::<CURRENT_VERSION, _>(PSK_LABEL, &serde_json::to_vec(psk_id)?)
}
fn encryption_key_pair<
HpkeKeyPair: traits::HpkeKeyPair<CURRENT_VERSION>,
EncryptionKey: traits::EncryptionKey<CURRENT_VERSION>,
>(
&self,
public_key: &EncryptionKey,
) -> Result<Option<HpkeKeyPair>, Self::Error> {
self.read::<CURRENT_VERSION, _>(ENCRYPTION_KEY_PAIR_LABEL, &serde_json::to_vec(public_key)?)
}
fn delete_signature_key_pair<
SignaturePublicKey: traits::SignaturePublicKey<CURRENT_VERSION>,
>(
&self,
public_key: &SignaturePublicKey,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(SIGNATURE_KEY_PAIR_LABEL, &serde_json::to_vec(public_key)?)
}
fn delete_encryption_key_pair<EncryptionKey: traits::EncryptionKey<CURRENT_VERSION>>(
&self,
public_key: &EncryptionKey,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(ENCRYPTION_KEY_PAIR_LABEL, &serde_json::to_vec(public_key)?)
}
fn delete_key_package<KeyPackageRef: traits::HashReference<CURRENT_VERSION>>(
&self,
hash_ref: &KeyPackageRef,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(KEY_PACKAGE_LABEL, &serde_json::to_vec(hash_ref)?)
}
fn delete_psk<PskKey: traits::PskId<CURRENT_VERSION>>(
&self,
psk_id: &PskKey,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(PSK_LABEL, &serde_json::to_vec(psk_id)?)
}
fn group_state<
GroupState: traits::GroupState<CURRENT_VERSION>,
GroupId: traits::GroupId<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<GroupState>, Self::Error> {
self.read::<CURRENT_VERSION, _>(GROUP_STATE_LABEL, &serde_json::to_vec(group_id)?)
}
fn write_group_state<
GroupState: traits::GroupState<CURRENT_VERSION>,
GroupId: traits::GroupId<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
group_state: &GroupState,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
GROUP_STATE_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(group_state)?,
)
}
fn delete_group_state<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(GROUP_STATE_LABEL, &serde_json::to_vec(group_id)?)
}
fn message_secrets<
GroupId: traits::GroupId<CURRENT_VERSION>,
MessageSecrets: traits::MessageSecrets<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<MessageSecrets>, Self::Error> {
self.read::<CURRENT_VERSION, _>(MESSAGE_SECRETS_LABEL, &serde_json::to_vec(group_id)?)
}
fn write_message_secrets<
GroupId: traits::GroupId<CURRENT_VERSION>,
MessageSecrets: traits::MessageSecrets<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
message_secrets: &MessageSecrets,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
MESSAGE_SECRETS_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(message_secrets)?,
)
}
fn delete_message_secrets<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(MESSAGE_SECRETS_LABEL, &serde_json::to_vec(group_id)?)
}
fn resumption_psk_store<
GroupId: traits::GroupId<CURRENT_VERSION>,
ResumptionPskStore: traits::ResumptionPskStore<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<ResumptionPskStore>, Self::Error> {
self.read::<CURRENT_VERSION, _>(RESUMPTION_PSK_STORE_LABEL, &serde_json::to_vec(group_id)?)
}
fn write_resumption_psk_store<
GroupId: traits::GroupId<CURRENT_VERSION>,
ResumptionPskStore: traits::ResumptionPskStore<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
resumption_psk_store: &ResumptionPskStore,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
RESUMPTION_PSK_STORE_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(resumption_psk_store)?,
)
}
fn delete_all_resumption_psk_secrets<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(RESUMPTION_PSK_STORE_LABEL, &serde_json::to_vec(group_id)?)
}
fn own_leaf_index<
GroupId: traits::GroupId<CURRENT_VERSION>,
LeafNodeIndex: traits::LeafNodeIndex<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<LeafNodeIndex>, Self::Error> {
self.read::<CURRENT_VERSION, _>(OWN_LEAF_NODE_INDEX_LABEL, &serde_json::to_vec(group_id)?)
}
fn write_own_leaf_index<
GroupId: traits::GroupId<CURRENT_VERSION>,
LeafNodeIndex: traits::LeafNodeIndex<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
own_leaf_index: &LeafNodeIndex,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
OWN_LEAF_NODE_INDEX_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(own_leaf_index)?,
)
}
fn delete_own_leaf_index<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(OWN_LEAF_NODE_INDEX_LABEL, &serde_json::to_vec(group_id)?)
}
fn group_epoch_secrets<
GroupId: traits::GroupId<CURRENT_VERSION>,
GroupEpochSecrets: traits::GroupEpochSecrets<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<GroupEpochSecrets>, Self::Error> {
self.read::<CURRENT_VERSION, _>(EPOCH_SECRETS_LABEL, &serde_json::to_vec(group_id)?)
}
fn write_group_epoch_secrets<
GroupId: traits::GroupId<CURRENT_VERSION>,
GroupEpochSecrets: traits::GroupEpochSecrets<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
group_epoch_secrets: &GroupEpochSecrets,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
EPOCH_SECRETS_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(group_epoch_secrets)?,
)
}
fn delete_group_epoch_secrets<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(EPOCH_SECRETS_LABEL, &serde_json::to_vec(group_id)?)
}
fn write_encryption_epoch_key_pairs<
GroupId: traits::GroupId<CURRENT_VERSION>,
EpochKey: traits::EpochKey<CURRENT_VERSION>,
HpkeKeyPair: traits::HpkeKeyPair<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
epoch: &EpochKey,
leaf_index: u32,
key_pairs: &[HpkeKeyPair],
) -> Result<(), Self::Error> {
let key = epoch_key_pairs_id(group_id, epoch, leaf_index)?;
let value = serde_json::to_vec(key_pairs)?;
self.write::<CURRENT_VERSION>(EPOCH_KEY_PAIRS_LABEL, &key, value)
}
fn encryption_epoch_key_pairs<
GroupId: traits::GroupId<CURRENT_VERSION>,
EpochKey: traits::EpochKey<CURRENT_VERSION>,
HpkeKeyPair: traits::HpkeKeyPair<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
epoch: &EpochKey,
leaf_index: u32,
) -> Result<Vec<HpkeKeyPair>, Self::Error> {
let key = epoch_key_pairs_id(group_id, epoch, leaf_index)?;
match self.get(&build_key_from_vec::<CURRENT_VERSION>(
EPOCH_KEY_PAIRS_LABEL,
key,
))? {
Some(bytes) => Ok(serde_json::from_slice(&bytes)?),
None => Ok(vec![]),
}
}
fn delete_encryption_epoch_key_pairs<
GroupId: traits::GroupId<CURRENT_VERSION>,
EpochKey: traits::EpochKey<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
epoch: &EpochKey,
leaf_index: u32,
) -> Result<(), Self::Error> {
let key = epoch_key_pairs_id(group_id, epoch, leaf_index)?;
self.delete::<CURRENT_VERSION>(EPOCH_KEY_PAIRS_LABEL, &key)
}
fn clear_proposal_queue<
GroupId: traits::GroupId<CURRENT_VERSION>,
ProposalRef: traits::ProposalRef<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
let proposal_refs: Vec<ProposalRef> = self.read_list::<CURRENT_VERSION, _>(
PROPOSAL_QUEUE_REFS_LABEL,
&serde_json::to_vec(group_id)?,
)?;
for proposal_ref in proposal_refs {
// Delete under the same composite key `queue_proposal` wrote it with;
// the reference impl uses a bare key here and orphans these rows.
let key = serde_json::to_vec(&(group_id, &proposal_ref))?;
self.delete::<CURRENT_VERSION>(QUEUED_PROPOSAL_LABEL, &key)?;
}
self.delete::<CURRENT_VERSION>(PROPOSAL_QUEUE_REFS_LABEL, &serde_json::to_vec(group_id)?)
}
fn mls_group_join_config<
GroupId: traits::GroupId<CURRENT_VERSION>,
MlsGroupJoinConfig: traits::MlsGroupJoinConfig<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Option<MlsGroupJoinConfig>, Self::Error> {
self.read::<CURRENT_VERSION, _>(JOIN_CONFIG_LABEL, &serde_json::to_vec(group_id)?)
}
fn write_mls_join_config<
GroupId: traits::GroupId<CURRENT_VERSION>,
MlsGroupJoinConfig: traits::MlsGroupJoinConfig<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
config: &MlsGroupJoinConfig,
) -> Result<(), Self::Error> {
self.write::<CURRENT_VERSION>(
JOIN_CONFIG_LABEL,
&serde_json::to_vec(group_id)?,
serde_json::to_vec(config)?,
)
}
fn own_leaf_nodes<
GroupId: traits::GroupId<CURRENT_VERSION>,
LeafNode: traits::LeafNode<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
) -> Result<Vec<LeafNode>, Self::Error> {
self.read_list::<CURRENT_VERSION, _>(OWN_LEAF_NODES_LABEL, &serde_json::to_vec(group_id)?)
}
fn append_own_leaf_node<
GroupId: traits::GroupId<CURRENT_VERSION>,
LeafNode: traits::LeafNode<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
leaf_node: &LeafNode,
) -> Result<(), Self::Error> {
let key = serde_json::to_vec(group_id)?;
let value = serde_json::to_vec(leaf_node)?;
self.append::<CURRENT_VERSION>(OWN_LEAF_NODES_LABEL, &key, value)
}
fn delete_own_leaf_nodes<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(OWN_LEAF_NODES_LABEL, &serde_json::to_vec(group_id)?)
}
fn delete_group_config<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(JOIN_CONFIG_LABEL, &serde_json::to_vec(group_id)?)
}
fn delete_tree<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(TREE_LABEL, &serde_json::to_vec(group_id)?)
}
fn delete_confirmation_tag<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(CONFIRMATION_TAG_LABEL, &serde_json::to_vec(group_id)?)
}
fn delete_context<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(GROUP_CONTEXT_LABEL, &serde_json::to_vec(group_id)?)
}
fn delete_interim_transcript_hash<GroupId: traits::GroupId<CURRENT_VERSION>>(
&self,
group_id: &GroupId,
) -> Result<(), Self::Error> {
self.delete::<CURRENT_VERSION>(
INTERIM_TRANSCRIPT_HASH_LABEL,
&serde_json::to_vec(group_id)?,
)
}
fn remove_proposal<
GroupId: traits::GroupId<CURRENT_VERSION>,
ProposalRef: traits::ProposalRef<CURRENT_VERSION>,
>(
&self,
group_id: &GroupId,
proposal_ref: &ProposalRef,
) -> Result<(), Self::Error> {
let key = serde_json::to_vec(group_id)?;
let value = serde_json::to_vec(proposal_ref)?;
self.remove_item::<CURRENT_VERSION>(PROPOSAL_QUEUE_REFS_LABEL, &key, value)?;
let key = serde_json::to_vec(&(group_id, proposal_ref))?;
self.delete::<CURRENT_VERSION>(QUEUED_PROPOSAL_LABEL, &key)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::common::StorageConfig;
// Minimal Key/Entity newtypes so the KV layer can be exercised without
// dragging in openmls' concrete group types.
#[derive(serde::Serialize)]
struct TestKey(Vec<u8>);
impl Key<CURRENT_VERSION> for TestKey {}
impl traits::GroupId<CURRENT_VERSION> for TestKey {}
#[derive(serde::Serialize, serde::Deserialize, PartialEq, Debug, Clone)]
struct TestVal(u32);
impl Entity<CURRENT_VERSION> for TestVal {}
impl traits::GroupState<CURRENT_VERSION> for TestVal {}
impl traits::LeafNode<CURRENT_VERSION> for TestVal {}
fn store() -> ChatStorage {
ChatStorage::in_memory()
}
#[test]
fn single_value_round_trip() {
let s = store();
let gid = TestKey(b"group-a".to_vec());
assert_eq!(s.group_state::<TestVal, _>(&gid).unwrap(), None);
s.write_group_state(&gid, &TestVal(7)).unwrap();
assert_eq!(s.group_state::<TestVal, _>(&gid).unwrap(), Some(TestVal(7)));
// Overwrite replaces in place.
s.write_group_state(&gid, &TestVal(9)).unwrap();
assert_eq!(s.group_state::<TestVal, _>(&gid).unwrap(), Some(TestVal(9)));
s.delete_group_state(&gid).unwrap();
assert_eq!(s.group_state::<TestVal, _>(&gid).unwrap(), None);
}
#[test]
fn list_append_and_read() {
let s = store();
let gid = TestKey(b"group-b".to_vec());
assert!(s.own_leaf_nodes::<_, TestVal>(&gid).unwrap().is_empty());
s.append_own_leaf_node(&gid, &TestVal(1)).unwrap();
s.append_own_leaf_node(&gid, &TestVal(2)).unwrap();
let nodes: Vec<TestVal> = s.own_leaf_nodes(&gid).unwrap();
assert_eq!(nodes, vec![TestVal(1), TestVal(2)]);
s.delete_own_leaf_nodes(&gid).unwrap();
assert!(s.own_leaf_nodes::<_, TestVal>(&gid).unwrap().is_empty());
}
#[test]
fn persists_across_reopen() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("mls.db").to_str().unwrap().to_string();
let gid = TestKey(b"group-c".to_vec());
{
let s = ChatStorage::new(StorageConfig::File(path.clone())).unwrap();
s.write_group_state(&gid, &TestVal(42)).unwrap();
}
let reopened = ChatStorage::new(StorageConfig::File(path)).unwrap();
assert_eq!(
reopened.group_state::<TestVal, _>(&gid).unwrap(),
Some(TestVal(42))
);
}
}

1
core/storage/Cargo.toml
View File

@ -9,4 +9,5 @@ description = "Shared storage layer for libchat"
crypto = { workspace = true }
# External dependencies (sorted)
openmls_traits = "0.5.0"
thiserror = "2"

23
core/storage/src/store.rs
View File

@ -1,4 +1,5 @@
use crypto::{Identity, PrivateKey};
use openmls_traits::storage::{CURRENT_VERSION, StorageProvider};
use crate::StorageError;
@ -125,7 +126,23 @@ pub trait RatchetStore {
// TODO: (P2) this should be defined in the ConversationType
pub trait ChatStore: IdentityStore + EphemeralKeyStore + ConversationStore + RatchetStore {}
/// The full durable-storage contract libchat needs: the chat-domain sub-stores
/// plus the OpenMLS [`StorageProvider`], so one type holds both chat state and
/// MLS group state.
pub trait ChatStore:
IdentityStore
+ EphemeralKeyStore
+ ConversationStore
+ RatchetStore
+ StorageProvider<CURRENT_VERSION>
{
}
impl<T> ChatStore for T where T: IdentityStore + EphemeralKeyStore + ConversationStore + RatchetStore
{}
impl<T> ChatStore for T where
T: IdentityStore
+ EphemeralKeyStore
+ ConversationStore
+ RatchetStore
+ StorageProvider<CURRENT_VERSION>
{
}

162
crates/client/src/builder.rs
View File

@ -1,7 +1,6 @@
use components::EphemeralRegistry;
use crossbeam_channel::Receiver;
use libchat::{ChatError, ChatStorage, IdentityProvider, RegistrationService, StorageConfig};
use storage::ChatStore;
use crate::Transport;
use crate::client::ChatClient;
@ -13,11 +12,14 @@ use crate::event::Event;
/// component will be filled in with a sensible default when `build()` is called.
pub struct Unset;
pub struct ChatClientBuilder<I = Unset, T = Unset, R = Unset, S = Unset> {
pub struct ChatClientBuilder<I = Unset, T = Unset, R = Unset> {
ident: I,
transport: T,
registration: R,
storage: S,
/// The durable store; defaults to an ephemeral in-memory `ChatStorage` at
/// `build()` when left unset by [`storage`](Self::storage) or
/// [`storage_config`](Self::storage_config).
storage: Option<ChatStorage>,
}
impl Default for ChatClientBuilder {
@ -26,7 +28,7 @@ impl Default for ChatClientBuilder {
ident: Unset,
transport: Unset,
registration: Unset,
storage: Unset,
storage: None,
}
}
}
@ -37,8 +39,8 @@ impl ChatClientBuilder {
}
}
impl<I, T, R, S> ChatClientBuilder<I, T, R, S> {
pub fn ident<NI>(self, ident: NI) -> ChatClientBuilder<NI, T, R, S> {
impl<I, T, R> ChatClientBuilder<I, T, R> {
pub fn ident<NI>(self, ident: NI) -> ChatClientBuilder<NI, T, R> {
ChatClientBuilder {
ident,
transport: self.transport,
@ -47,7 +49,7 @@ impl<I, T, R, S> ChatClientBuilder<I, T, R, S> {
}
}
pub fn transport<NT>(self, transport: NT) -> ChatClientBuilder<I, NT, R, S> {
pub fn transport<NT>(self, transport: NT) -> ChatClientBuilder<I, NT, R> {
ChatClientBuilder {
ident: self.ident,
transport,
@ -56,7 +58,7 @@ impl<I, T, R, S> ChatClientBuilder<I, T, R, S> {
}
}
pub fn registration<NR>(self, registration: NR) -> ChatClientBuilder<I, T, NR, S> {
pub fn registration<NR>(self, registration: NR) -> ChatClientBuilder<I, T, NR> {
ChatClientBuilder {
ident: self.ident,
transport: self.transport,
@ -65,151 +67,79 @@ impl<I, T, R, S> ChatClientBuilder<I, T, R, S> {
}
}
pub fn storage<NS>(self, storage: NS) -> ChatClientBuilder<I, T, R, NS> {
ChatClientBuilder {
ident: self.ident,
transport: self.transport,
registration: self.registration,
storage,
}
pub fn storage(mut self, storage: ChatStorage) -> Self {
self.storage = Some(storage);
self
}
pub fn storage_config(self, config: StorageConfig) -> ChatClientBuilder<I, T, R, ChatStorage> {
pub fn storage_config(mut self, config: StorageConfig) -> Self {
let storage = ChatStorage::new(config)
.map_err(ChatError::from)
.expect("Storage config file should be valid");
ChatClientBuilder {
ident: self.ident,
transport: self.transport,
registration: self.registration,
storage,
}
self.storage = Some(storage);
self
}
}
type Built<I, T, R, S> = Result<(ChatClient<I, T, R, S>, Receiver<Event>), ClientError>;
type Built<I, T, R> = Result<(ChatClient<I, T, R>, Receiver<Event>), ClientError>;
// All four explicitly provided.
impl<I, T, R, S> ChatClientBuilder<I, T, R, S>
where
I: IdentityProvider + Send + 'static,
T: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
pub fn build(self) -> Built<I, T, R, S> {
ChatClient::new(self.ident, self.transport, self.registration, self.storage)
}
}
// Transport only; I, R, S all default.
impl<T: Transport + Send + 'static> ChatClientBuilder<Unset, T, Unset, Unset> {
pub fn build(self) -> Built<DelegateSigner, T, EphemeralRegistry, ChatStorage> {
ChatClient::new(
DelegateSigner::random(),
self.transport,
EphemeralRegistry::new(),
ChatStorage::in_memory(),
)
}
}
// I and T; R and S default.
impl<I, T> ChatClientBuilder<I, T, Unset, Unset>
where
I: IdentityProvider + Send + 'static,
T: Transport + Send + 'static,
{
pub fn build(self) -> Built<I, T, EphemeralRegistry, ChatStorage> {
ChatClient::new(
self.ident,
self.transport,
EphemeralRegistry::new(),
ChatStorage::in_memory(),
)
}
}
// T and R; I and S default.
impl<T, R> ChatClientBuilder<Unset, T, R, Unset>
where
T: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
{
pub fn build(self) -> Built<DelegateSigner, T, R, ChatStorage> {
ChatClient::new(
DelegateSigner::random(),
self.transport,
self.registration,
ChatStorage::in_memory(),
)
}
}
// T and S; I and R default.
impl<T, S> ChatClientBuilder<Unset, T, Unset, S>
where
T: Transport + Send + 'static,
S: ChatStore + Send + 'static,
{
pub fn build(self) -> Built<DelegateSigner, T, EphemeralRegistry, S> {
ChatClient::new(
DelegateSigner::random(),
self.transport,
EphemeralRegistry::new(),
self.storage,
)
}
}
// I, T, and R; S defaults.
impl<I, T, R> ChatClientBuilder<I, T, R, Unset>
// I and R explicitly provided.
impl<I, T, R> ChatClientBuilder<I, T, R>
where
I: IdentityProvider + Send + 'static,
T: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
{
pub fn build(self) -> Built<I, T, R, ChatStorage> {
pub fn build(self) -> Built<I, T, R> {
ChatClient::new(
self.ident,
self.transport,
self.registration,
ChatStorage::in_memory(),
self.storage.unwrap_or_else(ChatStorage::in_memory),
)
}
}
// T, R, and S; I defaults.
impl<T, R, S> ChatClientBuilder<Unset, T, R, S>
where
T: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
pub fn build(self) -> Built<DelegateSigner, T, R, S> {
// Transport only; I and R default.
impl<T: Transport + Send + 'static> ChatClientBuilder<Unset, T, Unset> {
pub fn build(self) -> Built<DelegateSigner, T, EphemeralRegistry> {
ChatClient::new(
DelegateSigner::random(),
self.transport,
self.registration,
self.storage,
EphemeralRegistry::new(),
self.storage.unwrap_or_else(ChatStorage::in_memory),
)
}
}
// I, T, and S; R defaults.
impl<I, T, S> ChatClientBuilder<I, T, Unset, S>
// I and T; R defaults.
impl<I, T> ChatClientBuilder<I, T, Unset>
where
I: IdentityProvider + Send + 'static,
T: Transport + Send + 'static,
S: ChatStore + Send + 'static,
{
pub fn build(self) -> Built<I, T, EphemeralRegistry, S> {
pub fn build(self) -> Built<I, T, EphemeralRegistry> {
ChatClient::new(
self.ident,
self.transport,
EphemeralRegistry::new(),
self.storage,
self.storage.unwrap_or_else(ChatStorage::in_memory),
)
}
}
// T and R; I defaults.
impl<T, R> ChatClientBuilder<Unset, T, R>
where
T: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
{
pub fn build(self) -> Built<DelegateSigner, T, R> {
ChatClient::new(
DelegateSigner::random(),
self.transport,
self.registration,
self.storage.unwrap_or_else(ChatStorage::in_memory),
)
}
}

29
crates/client/src/client.rs
View File

@ -5,17 +5,19 @@ use components::{ThreadedWakeupService, WakeupEvent};
use crossbeam_channel::{Receiver, Sender, select};
use crypto::Ed25519VerifyingKey;
use libchat::{
AccountDirectory, ConversationId, ConvoOutcome, Core, DeliveryService, IdentId, IdentIdRef,
IdentityProvider, InboxOutcome, Introduction, PayloadOutcome, RegistrationService,
AccountDirectory, ChatStorage, ConversationId, ConvoOutcome, Core, DeliveryService, IdentId,
IdentIdRef, IdentityProvider, InboxOutcome, Introduction, PayloadOutcome, RegistrationService,
};
use parking_lot::Mutex;
use storage::ChatStore;
use crate::delegate::DelegateCredential;
use crate::errors::ClientError;
use crate::event::{Event, MessageSender};
type ClientCore<I, T, R, S> = Core<(I, T, R, ThreadedWakeupService, S)>;
// The client always persists to `ChatStorage` (a durable SQLite store that holds
// both chat and MLS state, in-memory when the config is ephemeral), so the store
// is fixed rather than a public generic parameter on `ChatClient`.
type ClientCore<I, T, R> = Core<(I, T, R, ThreadedWakeupService, ChatStorage)>;
type AccountAddressRef<'a> = &'a str;
type LocalSignerId = IdentId;
@ -40,16 +42,15 @@ pub trait Transport: DeliveryService + Send + 'static {
/// caller's thread: they briefly lock the core, invoke it, and return — no
/// message-passing round-trip. The `Arc`/`Mutex`/threads live entirely here;
/// the core never mentions threads.
pub struct ChatClient<I, T, R, S>
pub struct ChatClient<I, T, R>
where
I: IdentityProvider + Send + 'static,
T: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
/// `parking_lot::Mutex` for its eventual fairness: an inbound burst can't
/// starve caller operations of the lock.
core: Arc<Mutex<ClientCore<I, T, R, S>>>,
core: Arc<Mutex<ClientCore<I, T, R>>>,
/// Dropped on `Drop` to wake the worker's `select!` and shut it down.
shutdown: Option<Sender<()>>,
worker: Option<JoinHandle<()>>,
@ -57,18 +58,17 @@ where
}
// -- GenericChatClient
impl<I, T, R, S> ChatClient<I, T, R, S>
impl<I, T, R> ChatClient<I, T, R>
where
I: IdentityProvider + Send + 'static,
T: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
pub fn new(
ident: I,
mut transport: T,
reg: R,
storage: S,
storage: ChatStorage,
) -> Result<(Self, Receiver<Event>), ClientError> {
let inbound = transport.inbound();
@ -79,7 +79,7 @@ where
}
fn spawn(
core: ClientCore<I, T, R, S>,
core: ClientCore<I, T, R>,
inbound: Receiver<Vec<u8>>,
wakeup_events: Receiver<WakeupEvent>,
) -> (Self, Receiver<Event>) {
@ -172,12 +172,11 @@ where
}
}
impl<I, T, R, S> Drop for ChatClient<I, T, R, S>
impl<I, T, R> Drop for ChatClient<I, T, R>
where
I: IdentityProvider + Send + 'static,
T: Transport + Send + 'static,
R: RegistrationService + Send + 'static,
S: ChatStore + Send + 'static,
{
fn drop(&mut self) {
// Dropping the sender disconnects the worker's shutdown channel, waking
@ -192,8 +191,8 @@ where
/// Background loop: block until an inbound payload or shutdown arrives, drive
/// the core on each payload, and forward events. No polling — `select!` parks
/// the thread until one of the channels is ready.
fn worker_loop<I: IdentityProvider + 'static, T, R, S: ChatStore + 'static>(
core: Arc<Mutex<ClientCore<I, T, R, S>>>,
fn worker_loop<I: IdentityProvider + 'static, T, R>(
core: Arc<Mutex<ClientCore<I, T, R>>>,
inbound: Receiver<Vec<u8>>,
wakeup_events: Receiver<WakeupEvent>,
shutdown: Receiver<()>,

7
crates/client/src/logos.rs
View File

@ -12,7 +12,7 @@
//! environment-specific configuration that belong to the binary, not here.
use crossbeam_channel::Receiver;
use libchat::{ChatStorage, StorageConfig};
use libchat::StorageConfig;
use crate::ChatClientBuilder;
use crate::client::{ChatClient, Transport};
@ -27,8 +27,9 @@ const REGISTRY_ENDPOINT: &str = "https://devnet.chat-kc.logos.co";
/// A [`ChatClient`] wired to the Logos service stack: a [`DelegateSigner`]
/// identity, the HTTP keypackage + account registry ([`HttpRegistry`], which is
/// both the keypackage store and the account → device directory), and encrypted
/// [`ChatStorage`]. Only the transport `T` is supplied by the caller.
pub type LogosChatClient<T> = ChatClient<DelegateSigner, T, HttpRegistry, ChatStorage>;
/// [`ChatStorage`](libchat::ChatStorage). Only the transport `T` is supplied by
/// the caller.
pub type LogosChatClient<T> = ChatClient<DelegateSigner, T, HttpRegistry>;
impl<T> LogosChatClient<T>
where

2
crates/client/tests/saro_and_raya.rs
View File

@ -33,7 +33,7 @@ fn create_test_client(
reg: EphemeralRegistry,
) -> Result<
(
ChatClient<DelegateSigner, InProcessDelivery, EphemeralRegistry, libchat::ChatStorage>,
ChatClient<DelegateSigner, InProcessDelivery, EphemeralRegistry>,
Receiver<Event>,
),
logos_chat::ClientError,

2
extensions/components/src/lib.rs
View File

@ -1,10 +1,8 @@
mod contact_registry;
pub mod delivery;
mod storage;
mod wakeup;
pub use contact_registry::ephemeral::EphemeralRegistry;
pub use contact_registry::http::{HttpRegistry, HttpRegistryError};
pub use delivery::*;
pub use storage::*;
pub use wakeup::*;

3
extensions/components/src/storage.rs
View File

@ -1,3 +0,0 @@
mod in_memory_store;
pub use in_memory_store::MemStore;

136
extensions/components/src/storage/in_memory_store.rs
View File

@ -1,136 +0,0 @@
use std::collections::HashMap;
use storage::{
// TODO: (P4) Importable crates need to be prefixed with a project name to avoid conflicts
ConversationMeta,
ConversationStore,
EphemeralKeyStore,
IdentityStore,
RatchetStore,
};
/// An Test focused StorageService which holds data in a hashmap
pub struct MemStore {
convos: HashMap<String, ConversationMeta>,
}
impl MemStore {
pub fn new() -> Self {
Self {
convos: HashMap::new(),
}
}
}
impl Default for MemStore {
fn default() -> Self {
Self::new()
}
}
impl ConversationStore for MemStore {
fn save_conversation(
&mut self,
meta: &storage::ConversationMeta,
) -> Result<(), storage::StorageError> {
self.convos
.insert(meta.local_convo_id.clone(), meta.clone());
Ok(())
}
fn load_conversation(
&self,
local_convo_id: &str,
) -> Result<Option<storage::ConversationMeta>, storage::StorageError> {
let a = self.convos.get(local_convo_id).cloned();
Ok(a)
}
fn remove_conversation(&mut self, _local_convo_id: &str) -> Result<(), storage::StorageError> {
todo!()
}
fn load_conversations(&self) -> Result<Vec<storage::ConversationMeta>, storage::StorageError> {
Ok(self.convos.values().cloned().collect())
}
fn has_conversation(&self, local_convo_id: &str) -> Result<bool, storage::StorageError> {
Ok(self.convos.contains_key(local_convo_id))
}
}
impl IdentityStore for MemStore {
fn load_identity(&self) -> Result<Option<crypto::Identity>, storage::StorageError> {
// todo!()
Ok(None)
}
fn save_identity(&mut self, _identity: &crypto::Identity) -> Result<(), storage::StorageError> {
// todo!()
Ok(())
}
}
impl EphemeralKeyStore for MemStore {
fn save_ephemeral_key(
&mut self,
_public_key_hex: &str,
_private_key: &crypto::PrivateKey,
) -> Result<(), storage::StorageError> {
todo!()
}
fn load_ephemeral_key(
&self,
_public_key_hex: &str,
) -> Result<Option<crypto::PrivateKey>, storage::StorageError> {
todo!()
}
fn remove_ephemeral_key(&mut self, _public_key_hex: &str) -> Result<(), storage::StorageError> {
todo!()
}
}
impl RatchetStore for MemStore {
fn save_ratchet_state(
&mut self,
_conversation_id: &str,
_state: &storage::RatchetStateRecord,
_skipped_keys: &[storage::SkippedKeyRecord],
) -> Result<(), storage::StorageError> {
todo!()
}
fn load_ratchet_state(
&self,
_conversation_id: &str,
) -> Result<storage::RatchetStateRecord, storage::StorageError> {
todo!()
}
fn load_skipped_keys(
&self,
_conversation_id: &str,
) -> Result<Vec<storage::SkippedKeyRecord>, storage::StorageError> {
todo!()
}
fn has_ratchet_state(&self, _conversation_id: &str) -> Result<bool, storage::StorageError> {
todo!()
}
fn delete_ratchet_state(
&mut self,
_conversation_id: &str,
) -> Result<(), storage::StorageError> {
todo!()
}
fn cleanup_old_skipped_keys(
&mut self,
_max_age_secs: i64,
) -> Result<usize, storage::StorageError> {
todo!()
}
}