mirror of
https://github.com/logos-messaging/libchat.git
synced 2026-08-25 07:01:08 +00:00
Merge branch 'main' into mch/libchat-refactoring-sender-attribution
# Conflicts: # bin/chat-cli/README.md # bin/chat-cli/src/app.rs
This commit is contained in:
Mojtaba Chenani
@@ -47,6 +47,16 @@ jobs:
|
||||
- run: rustup component add rustfmt
|
||||
- run: cargo fmt --all -- --check
|
||||
|
||||
cargo-deny:
|
||||
name: Cargo Deny
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
# Reads the dependency graph only, so no protoc/liblogosdelivery needed.
|
||||
- uses: EmbarkStudios/cargo-deny-action@v2
|
||||
with:
|
||||
command: check
|
||||
|
||||
smoketest:
|
||||
name: Smoketest
|
||||
environment: ${{ github.ref == 'refs/heads/main' && 'public-cache' || '' }}
|
||||
|
||||
Generated
+117
-19
@@ -754,9 +754,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "anyhow"
|
||||
version = "1.0.102"
|
||||
version = "1.0.104"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
|
||||
checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470"
|
||||
|
||||
[[package]]
|
||||
name = "arboard"
|
||||
@@ -836,6 +836,23 @@ dependencies = [
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-ff"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f7a806ac6c8307b929df4645776290a50ee2aac754ad09d8bdf73391309e43af"
|
||||
dependencies = [
|
||||
"ark-ff-asm 0.6.0",
|
||||
"ark-ff-macros 0.6.0",
|
||||
"ark-serialize 0.6.0",
|
||||
"ark-std 0.6.0",
|
||||
"digest 0.10.7",
|
||||
"educe",
|
||||
"num-bigint",
|
||||
"num-traits",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-ff-asm"
|
||||
version = "0.3.0"
|
||||
@@ -866,6 +883,16 @@ dependencies = [
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-ff-asm"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1479009684adc073dff49a1025d3a7065b317a9ead25aaaca38cdc70058ba8a2"
|
||||
dependencies = [
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-ff-macros"
|
||||
version = "0.3.0"
|
||||
@@ -904,6 +931,19 @@ dependencies = [
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-ff-macros"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4a0691ed21ef00ef89c1e9bda832eba493dda3ec2f8d892fb25b705f73f06bb8"
|
||||
dependencies = [
|
||||
"num-bigint",
|
||||
"num-traits",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-serialize"
|
||||
version = "0.3.0"
|
||||
@@ -937,6 +977,30 @@ dependencies = [
|
||||
"num-bigint",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-serialize"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a74dd304fd536fb95d0a328e72be759209cc496a9da094c5bc56e5fea4f9e86b"
|
||||
dependencies = [
|
||||
"ark-serialize-derive",
|
||||
"ark-std 0.6.0",
|
||||
"digest 0.10.7",
|
||||
"num-bigint",
|
||||
"serde_with",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-serialize-derive"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4f153690697a2b91e5e1251ff98411ee5371500a111a0fd317a70e588eb300f9"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.117",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-std"
|
||||
version = "0.3.0"
|
||||
@@ -967,6 +1031,16 @@ dependencies = [
|
||||
"rand 0.8.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ark-std"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "367c9c827ed431bff6868b7aa926e05b16eb46603cc8b6e768e4a5553fa1d155"
|
||||
dependencies = [
|
||||
"num-traits",
|
||||
"rand 0.8.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrayvec"
|
||||
version = "0.7.6"
|
||||
@@ -1294,6 +1368,17 @@ dependencies = [
|
||||
"cpufeatures 0.2.17",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "chacha20"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures 0.3.0",
|
||||
"rand_core 0.10.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "chacha20poly1305"
|
||||
version = "0.10.1"
|
||||
@@ -1301,7 +1386,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35"
|
||||
dependencies = [
|
||||
"aead",
|
||||
"chacha20",
|
||||
"chacha20 0.9.1",
|
||||
"cipher",
|
||||
"poly1305",
|
||||
"zeroize",
|
||||
@@ -1621,9 +1706,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-epoch"
|
||||
version = "0.9.18"
|
||||
version = "0.9.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
|
||||
checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
@@ -2121,7 +2206,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"windows-sys 0.61.2",
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -2439,11 +2524,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"js-sys",
|
||||
"libc",
|
||||
"r-efi 5.3.0",
|
||||
"wasip2",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -2453,11 +2536,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"js-sys",
|
||||
"libc",
|
||||
"r-efi 6.0.0",
|
||||
"rand_core 0.10.1",
|
||||
"wasip2",
|
||||
"wasip3",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -3274,6 +3359,7 @@ dependencies = [
|
||||
"prost",
|
||||
"rand 0.9.4",
|
||||
"rand_core 0.6.4",
|
||||
"serde",
|
||||
"shared-traits",
|
||||
"storage",
|
||||
"tempfile",
|
||||
@@ -3749,7 +3835,7 @@ version = "0.50.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
|
||||
dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
"windows-sys 0.59.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4510,15 +4596,16 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "quinn-proto"
|
||||
version = "0.11.14"
|
||||
version = "0.11.16"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098"
|
||||
checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560"
|
||||
dependencies = [
|
||||
"aws-lc-rs",
|
||||
"bytes",
|
||||
"getrandom 0.3.4",
|
||||
"getrandom 0.4.2",
|
||||
"lru-slab",
|
||||
"rand 0.9.4",
|
||||
"rand 0.10.1",
|
||||
"rand_pcg",
|
||||
"ring",
|
||||
"rustc-hash",
|
||||
"rustls",
|
||||
@@ -4600,6 +4687,7 @@ version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207"
|
||||
dependencies = [
|
||||
"chacha20 0.10.1",
|
||||
"getrandom 0.4.2",
|
||||
"rand_core 0.10.1",
|
||||
]
|
||||
@@ -4659,6 +4747,15 @@ version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
|
||||
|
||||
[[package]]
|
||||
name = "rand_pcg"
|
||||
version = "0.10.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a"
|
||||
dependencies = [
|
||||
"rand_core 0.10.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rand_xorshift"
|
||||
version = "0.4.0"
|
||||
@@ -4936,14 +5033,15 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "ruint"
|
||||
version = "1.18.0"
|
||||
version = "1.20.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0298da754d1395046b0afdc2f20ee76d29a8ae310cd30ffa84ed42acba9cb12a"
|
||||
checksum = "f5e99bff0393163bb25029a6af25d3d8d202ba5b5438a74d1bd8789f5c822970"
|
||||
dependencies = [
|
||||
"alloy-rlp",
|
||||
"ark-ff 0.3.0",
|
||||
"ark-ff 0.4.2",
|
||||
"ark-ff 0.5.0",
|
||||
"ark-ff 0.6.0",
|
||||
"bytes",
|
||||
"fastrlp 0.3.1",
|
||||
"fastrlp 0.4.0",
|
||||
@@ -5035,7 +5133,7 @@ dependencies = [
|
||||
"errno",
|
||||
"libc",
|
||||
"linux-raw-sys 0.12.1",
|
||||
"windows-sys 0.61.2",
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -5093,7 +5191,7 @@ dependencies = [
|
||||
"security-framework",
|
||||
"security-framework-sys",
|
||||
"webpki-root-certs",
|
||||
"windows-sys 0.61.2",
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -5703,7 +5801,7 @@ dependencies = [
|
||||
"getrandom 0.4.2",
|
||||
"once_cell",
|
||||
"rustix 1.1.4",
|
||||
"windows-sys 0.61.2",
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -6401,7 +6499,7 @@ version = "0.1.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
|
||||
dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -31,6 +31,9 @@ default-members = [
|
||||
"crates/generic-chat",
|
||||
]
|
||||
|
||||
[workspace.package]
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
[workspace.dependencies]
|
||||
# Internal Workspace dependency declarations (sorted)
|
||||
logos-account = { path = "core/account" }
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "chat-cli"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[[bin]]
|
||||
name = "chat-cli"
|
||||
|
||||
+13
-14
@@ -36,20 +36,20 @@ Run two instances in separate terminals:
|
||||
|
||||
```bash
|
||||
# Terminal 1
|
||||
cargo run -p chat-cli -- --name alice --port 60001
|
||||
cargo run -p chat-cli -- --name saro --port 60001
|
||||
|
||||
# Terminal 2
|
||||
cargo run -p chat-cli -- --name bob --port 60002
|
||||
cargo run -p chat-cli -- --name raya --port 60002
|
||||
```
|
||||
|
||||
For local-only testing without any network dependency, use the file transport:
|
||||
|
||||
```bash
|
||||
# Terminal 1
|
||||
cargo run -p chat-cli -- --name alice --transport file
|
||||
cargo run -p chat-cli -- --name saro --transport file
|
||||
|
||||
# Terminal 2
|
||||
cargo run -p chat-cli -- --name bob --transport file
|
||||
cargo run -p chat-cli -- --name raya --transport file
|
||||
```
|
||||
|
||||
### Starting a conversation
|
||||
@@ -60,17 +60,17 @@ and copies it to the clipboard.
|
||||
|
||||
**Direct message (1:1):**
|
||||
|
||||
1. Alice types `/account` and sends Bob her address.
|
||||
2. Bob types `/dm <paste alice's address>`.
|
||||
1. Raya runs `/account` and shares her address.
|
||||
2. Saro types `/dm <paste raya's address>`.
|
||||
3. The chat opens on both sides; either can message.
|
||||
|
||||
**Group:**
|
||||
|
||||
1. Bob types `/new weekend` to create a group named "weekend" (or
|
||||
`/new weekend <alice's address>` to invite someone at creation).
|
||||
2. Bob types `/add <alice's address>` to invite Alice; the invite stays pending
|
||||
1. Saro types `/new weekend` to create a group named "weekend". A name is
|
||||
required; addresses can follow (e.g. Pax's) to invite people at creation.
|
||||
2. Saro types `/add <raya's address>` to invite Raya; the invite stays pending
|
||||
until the group commits it.
|
||||
3. `/members` lists the roster — Alice shows `(pending)` until the commit lands,
|
||||
3. `/members` lists the roster — Raya shows `(pending)` until the commit lands,
|
||||
then appears without it. Once committed, both can chat.
|
||||
|
||||
### Optional: KeyPackage registry
|
||||
@@ -86,9 +86,9 @@ process.
|
||||
cargo run -- --bind 127.0.0.1:18080
|
||||
|
||||
# Terminal 2 / 3 — chat clients pointing at it
|
||||
cargo run -p chat-cli -- --name alice --transport file \
|
||||
cargo run -p chat-cli -- --name saro --transport file \
|
||||
--registry-url http://127.0.0.1:18080
|
||||
cargo run -p chat-cli -- --name bob --transport file \
|
||||
cargo run -p chat-cli -- --name raya --transport file \
|
||||
--registry-url http://127.0.0.1:18080
|
||||
```
|
||||
|
||||
@@ -100,7 +100,6 @@ The registry is a throwaway testnet helper; v0.3 replaces it with a
|
||||
| Flag | Default | Description |
|
||||
|------|---------|-------------|
|
||||
| `--transport <kind>` | `logos-delivery` | Transport to use (`logos-delivery` or `file`) |
|
||||
| `--group-commit <mode>` | `auto` | How fast group `/add`s commit: `fast` (~1s, for demos), `default` (production de-mls timing), or `auto` (fast on `file`, default on the network) |
|
||||
| `--data <dir>` | `tmp/chat-cli-data` | Data directory (UI state and default SQLite path) |
|
||||
| `--db <path>` | `<data>/<name>.db` | SQLite file for persistent identity |
|
||||
| `--preset <name>` | `logos.dev` | logos-delivery network preset |
|
||||
@@ -115,7 +114,7 @@ The registry is a throwaway testnet helper; v0.3 replaces it with a
|
||||
| `/help` | Show available commands |
|
||||
| `/account` | Show your account address (copies to clipboard) |
|
||||
| `/dm <address>` | Start a direct (1:1) chat |
|
||||
| `/new [name] [address...]` | Create a group chat (optionally naming it and inviting members) |
|
||||
| `/new <name> [address...]` | Create a named group chat (optionally inviting members) |
|
||||
| `/add <address>` | Add someone to the active group |
|
||||
| `/members` | List members of the active conversation |
|
||||
| `/chats` | List all established chats |
|
||||
|
||||
+110
-70
@@ -13,15 +13,38 @@ use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::utils::now;
|
||||
|
||||
/// Who a displayed message is attributed to. `Foreign` carries a short sender
|
||||
/// label; `Own` is us (and, for now, system output). Leaves room to grow —
|
||||
/// e.g. a `System` variant for membership/metadata notices.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum MessageOrigin {
|
||||
Own,
|
||||
Foreign(String),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct DisplayMessage {
|
||||
pub from_self: bool,
|
||||
pub content: String,
|
||||
pub timestamp: u64,
|
||||
/// Short label of the sender for an incoming message (`None` for our own
|
||||
/// messages and system output). Used to attribute messages in groups.
|
||||
pub message_id: Option<String>,
|
||||
#[serde(default)]
|
||||
pub sender: Option<String>,
|
||||
pub delivered_to: Vec<String>,
|
||||
/// Who sent this message; attributes incoming group messages.
|
||||
pub origin: MessageOrigin,
|
||||
}
|
||||
|
||||
impl DisplayMessage {
|
||||
fn new(from_self: bool, content: String) -> Self {
|
||||
Self {
|
||||
from_self,
|
||||
content,
|
||||
timestamp: now(),
|
||||
message_id: None,
|
||||
delivered_to: Vec::new(),
|
||||
origin: MessageOrigin::Own,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Short display label for a message's sender: the account (or device) id,
|
||||
@@ -35,36 +58,11 @@ fn sender_label(sender: &MessageSender) -> String {
|
||||
id[..8.min(id.len())].to_string()
|
||||
}
|
||||
|
||||
/// Which kind of MLS conversation this is. `Dm` is a DirectV1 1:1 — no members
|
||||
/// can be added; `Group` is an addable GroupV2 conversation.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum ChatKind {
|
||||
Dm,
|
||||
Group,
|
||||
}
|
||||
|
||||
impl Default for ChatKind {
|
||||
fn default() -> Self {
|
||||
// Chats persisted before this field existed were all DirectV1 DMs.
|
||||
ChatKind::Dm
|
||||
}
|
||||
}
|
||||
|
||||
impl ChatKind {
|
||||
pub fn badge(self) -> &'static str {
|
||||
match self {
|
||||
ChatKind::Dm => "DM",
|
||||
ChatKind::Group => "group",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ChatSession {
|
||||
pub chat_id: String,
|
||||
pub nickname: Option<String>,
|
||||
#[serde(default)]
|
||||
pub kind: ChatKind,
|
||||
pub kind: ConversationClass,
|
||||
pub messages: Vec<DisplayMessage>,
|
||||
}
|
||||
|
||||
@@ -177,7 +175,12 @@ where
|
||||
}
|
||||
|
||||
/// Insert a freshly created conversation and make it active.
|
||||
fn start_session(&mut self, chat_id: String, kind: ChatKind, nickname: Option<String>) {
|
||||
fn start_session(
|
||||
&mut self,
|
||||
chat_id: String,
|
||||
kind: ConversationClass,
|
||||
nickname: Option<String>,
|
||||
) {
|
||||
self.state.chats.insert(
|
||||
chat_id.clone(),
|
||||
ChatSession {
|
||||
@@ -228,13 +231,9 @@ where
|
||||
if self.state.chats.contains_key(&chat_id) {
|
||||
return;
|
||||
}
|
||||
let kind = match class {
|
||||
ConversationClass::Private => ChatKind::Dm,
|
||||
ConversationClass::Group => ChatKind::Group,
|
||||
};
|
||||
let label = chat_id[..8.min(chat_id.len())].to_string();
|
||||
self.status = format!("New {} ({label})! Use /nickname to name it.", kind.badge());
|
||||
self.start_session(chat_id, kind, None);
|
||||
self.status = format!("New {class:?} ({label})! Use /nickname to name it.");
|
||||
self.start_session(chat_id, class, None);
|
||||
}
|
||||
Event::MessageReceived {
|
||||
convo_id,
|
||||
@@ -246,12 +245,58 @@ where
|
||||
let Some(session) = self.state.chats.get_mut(&chat_id) else {
|
||||
return;
|
||||
};
|
||||
session.messages.push(DisplayMessage {
|
||||
from_self: false,
|
||||
content: String::from_utf8_lossy(&content).into_owned(),
|
||||
timestamp: now(),
|
||||
sender: Some(label),
|
||||
});
|
||||
let mut message =
|
||||
DisplayMessage::new(false, String::from_utf8_lossy(&content).into_owned());
|
||||
message.origin = MessageOrigin::Foreign(label);
|
||||
session.messages.push(message);
|
||||
}
|
||||
Event::MessageAcked {
|
||||
convo_id,
|
||||
message_id,
|
||||
acked_by,
|
||||
} => {
|
||||
let Some(session) = self.state.chats.get_mut(convo_id.as_ref()) else {
|
||||
return;
|
||||
};
|
||||
let Some(message) = session
|
||||
.messages
|
||||
.iter_mut()
|
||||
.find(|m| m.message_id.as_deref() == Some(message_id.as_str()))
|
||||
else {
|
||||
return; // sent before this session, or not ours
|
||||
};
|
||||
let peer = acked_by.map_or_else(
|
||||
|| "a member".to_string(),
|
||||
|s| {
|
||||
let id = s.account.unwrap_or(s.local_identity);
|
||||
format!("{}…", &id.as_str()[..8.min(id.as_str().len())])
|
||||
},
|
||||
);
|
||||
if !message.delivered_to.contains(&peer) {
|
||||
message.delivered_to.push(peer);
|
||||
}
|
||||
}
|
||||
Event::MessageMissing {
|
||||
convo_id,
|
||||
sender_hint,
|
||||
..
|
||||
} => {
|
||||
let Some(session) = self.state.chats.get(convo_id.as_ref()) else {
|
||||
return;
|
||||
};
|
||||
// The hint is not authenticated (see `Event::MessageMissing`),
|
||||
// so name the author loosely rather than as an established fact.
|
||||
let author = sender_hint.map_or_else(
|
||||
|| "a member".to_string(),
|
||||
|s| {
|
||||
let id = s.account.unwrap_or(s.local_identity);
|
||||
format!("{}…", &id.as_str()[..8.min(id.as_str().len())])
|
||||
},
|
||||
);
|
||||
self.status = format!(
|
||||
"A message from {author} never arrived in '{}'.",
|
||||
session.display_name()
|
||||
);
|
||||
}
|
||||
Event::ConversationMembersChanged { convo_id } => {
|
||||
let chat_id = convo_id.to_string();
|
||||
@@ -273,17 +318,16 @@ where
|
||||
.clone()
|
||||
.ok_or_else(|| anyhow::anyhow!("No active chat. Use /dm or /new first."))?;
|
||||
|
||||
self.client
|
||||
let message_id = self
|
||||
.client
|
||||
.send_message(&chat_id, content.as_bytes())
|
||||
.map_err(|e| anyhow::anyhow!("{e:?}"))?;
|
||||
|
||||
if let Some(session) = self.state.chats.get_mut(&chat_id) {
|
||||
session.messages.push(DisplayMessage {
|
||||
from_self: true,
|
||||
content: content.to_string(),
|
||||
timestamp: now(),
|
||||
sender: None,
|
||||
});
|
||||
let mut message = DisplayMessage::new(true, content.to_string());
|
||||
// Kept so `MessageAcked` can find this message again.
|
||||
message.message_id = Some(message_id);
|
||||
session.messages.push(message);
|
||||
}
|
||||
self.save_state()?;
|
||||
|
||||
@@ -291,12 +335,8 @@ where
|
||||
}
|
||||
|
||||
fn add_system_message(&mut self, content: &str) {
|
||||
self.command_output.push(DisplayMessage {
|
||||
from_self: true,
|
||||
content: content.to_string(),
|
||||
timestamp: now(),
|
||||
sender: None,
|
||||
});
|
||||
self.command_output
|
||||
.push(DisplayMessage::new(true, content.to_string()));
|
||||
}
|
||||
|
||||
pub fn handle_command(&mut self, cmd: &str) -> Result<Option<String>> {
|
||||
@@ -309,7 +349,7 @@ where
|
||||
self.add_system_message("── Commands ──");
|
||||
self.add_system_message("/account - Show your account address");
|
||||
self.add_system_message("/dm <address> - Start a direct (1:1) chat");
|
||||
self.add_system_message("/new [name] [address...] - Create a group chat");
|
||||
self.add_system_message("/new <name> [address...] - Create a group chat");
|
||||
self.add_system_message("/add <address> - Add someone to the active group");
|
||||
self.add_system_message("/members - List members of the active conversation");
|
||||
self.add_system_message("/nickname <name> - Name the active chat");
|
||||
@@ -343,18 +383,18 @@ where
|
||||
.create_direct_conversation(address)
|
||||
.map_err(|e| anyhow::anyhow!("{e:?}"))?;
|
||||
let label = chat_id[..8.min(chat_id.len())].to_string();
|
||||
self.start_session(chat_id, ChatKind::Dm, None);
|
||||
self.start_session(chat_id, ConversationClass::Dm, None);
|
||||
self.save_state()?;
|
||||
self.status = format!("Direct chat started ({label}). Say hello!");
|
||||
Ok(Some(format!("DM started ({label})")))
|
||||
}
|
||||
"/new" => {
|
||||
// First token is the (optional) group name; any remaining tokens
|
||||
// are addresses to invite at creation. `/new` alone makes an empty
|
||||
// group.
|
||||
// First token is the group name (required); any remaining tokens
|
||||
// are addresses to invite at creation.
|
||||
let mut tokens = args.split_whitespace();
|
||||
let name = tokens.next().unwrap_or("").to_string();
|
||||
let nickname = (!name.is_empty()).then(|| name.clone());
|
||||
let Some(name) = tokens.next().map(str::to_string) else {
|
||||
return Ok(Some("Usage: /new <name> [address...]".to_string()));
|
||||
};
|
||||
// The creator is already a member; drop self and any repeats so we
|
||||
// don't propose a duplicate signature key (which MLS rejects).
|
||||
let my_addr = self.client.addr().to_string();
|
||||
@@ -363,10 +403,10 @@ where
|
||||
members.dedup();
|
||||
let chat_id = self
|
||||
.client
|
||||
.create_group_conversation(&members, GroupMetadata::new(name, ""))
|
||||
.create_group_conversation(&members, GroupMetadata::new(name.clone(), ""))
|
||||
.map_err(|e| anyhow::anyhow!("{e:?}"))?;
|
||||
let label = chat_id[..8.min(chat_id.len())].to_string();
|
||||
self.start_session(chat_id, ChatKind::Group, nickname);
|
||||
self.start_session(chat_id, ConversationClass::Group, Some(name));
|
||||
self.save_state()?;
|
||||
let msg = if members.is_empty() {
|
||||
format!("Group created ({label}).")
|
||||
@@ -384,12 +424,12 @@ where
|
||||
if address.is_empty() {
|
||||
return Ok(Some("Usage: /add <address>".to_string()));
|
||||
}
|
||||
let chat_id = self.state.active_chat.clone().ok_or_else(|| {
|
||||
let chat_id = self.state.active_chat.as_deref().ok_or_else(|| {
|
||||
anyhow::anyhow!("No active conversation. Use /new to create a group.")
|
||||
})?;
|
||||
// DMs are 1:1 and reject adds at the protocol level; refuse early
|
||||
// with a friendly hint rather than surfacing UnsupportedFunction.
|
||||
if self.state.chats.get(&chat_id).map(|s| s.kind) == Some(ChatKind::Dm) {
|
||||
if self.state.chats.get(chat_id).map(|s| s.kind) == Some(ConversationClass::Dm) {
|
||||
return Ok(Some(
|
||||
"DMs are 1:1 — start a group with /new to add people.".to_string(),
|
||||
));
|
||||
@@ -404,7 +444,7 @@ where
|
||||
}
|
||||
let already_present = self
|
||||
.client
|
||||
.group_members(&chat_id)
|
||||
.group_members(chat_id)
|
||||
.map(|members| {
|
||||
members
|
||||
.iter()
|
||||
@@ -418,7 +458,7 @@ where
|
||||
));
|
||||
}
|
||||
self.client
|
||||
.add_group_members(&chat_id, &[address])
|
||||
.add_group_members(chat_id, &[address])
|
||||
.map_err(|e| anyhow::anyhow!("{e:?}"))?;
|
||||
self.status = "Invite pending — the group will commit it shortly.".to_string();
|
||||
Ok(Some("Invite pending".to_string()))
|
||||
@@ -487,8 +527,8 @@ where
|
||||
""
|
||||
};
|
||||
let label = format!(
|
||||
" • [{}] {} ({}){marker}",
|
||||
s.kind.badge(),
|
||||
" • [{:?}] {} ({}){marker}",
|
||||
s.kind,
|
||||
s.display_name(),
|
||||
&s.chat_id[..8.min(s.chat_id.len())]
|
||||
);
|
||||
|
||||
+17
-2
@@ -102,7 +102,7 @@ where
|
||||
// DM the single peer's label is enough.
|
||||
let is_group = matches!(
|
||||
app.current_session().map(|s| s.kind),
|
||||
Some(crate::app::ChatKind::Group)
|
||||
Some(logos_chat::ConversationClass::Group)
|
||||
);
|
||||
|
||||
// Inner width: area minus borders (2) for wrapping long content.
|
||||
@@ -115,7 +115,10 @@ where
|
||||
let (prefix, style) = if msg.from_self {
|
||||
("You".to_string(), Style::default().fg(Color::Green))
|
||||
} else if is_group {
|
||||
let label = msg.sender.as_deref().unwrap_or(remote_name);
|
||||
let label = match &msg.origin {
|
||||
crate::app::MessageOrigin::Foreign(name) => name.as_str(),
|
||||
crate::app::MessageOrigin::Own => remote_name,
|
||||
};
|
||||
(label.to_string(), Style::default().fg(Color::Yellow))
|
||||
} else {
|
||||
(remote_name.to_string(), Style::default().fg(Color::Yellow))
|
||||
@@ -165,6 +168,18 @@ where
|
||||
remaining = tail;
|
||||
}
|
||||
|
||||
// Delivery receipts for our own sends: the peers whose later
|
||||
// messages showed they hold this one.
|
||||
if !msg.delivered_to.is_empty() {
|
||||
items.push(ListItem::new(Line::from(vec![
|
||||
Span::raw(indent.clone()),
|
||||
Span::styled(
|
||||
format!("↳ delivered to {}", msg.delivered_to.join(", ")),
|
||||
Style::default().fg(Color::DarkGray),
|
||||
),
|
||||
])));
|
||||
}
|
||||
|
||||
items
|
||||
})
|
||||
.collect();
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "logos-account"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[features]
|
||||
dev = []
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "libchat"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[lib]
|
||||
crate-type = ["rlib"]
|
||||
@@ -29,6 +30,7 @@ openmls_traits = "0.5.0"
|
||||
prost = "0.14.1"
|
||||
rand = "0.9"
|
||||
rand_core = { version = "0.6" }
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
thiserror = "2.0.17"
|
||||
tracing = "0.1.44"
|
||||
x25519-dalek = { version = "2.0.1", features = [
|
||||
|
||||
@@ -9,12 +9,19 @@
|
||||
//! - assign a deterministic message ID + Lamport timestamp to outbound msgs
|
||||
//! - attach a bounded causal-history frontier to each outbound message
|
||||
//! - on receive, detect referenced-but-unseen message IDs (gaps)
|
||||
//! - on receive, detect references to *our own* messages (acknowledgements)
|
||||
//!
|
||||
//! Out of scope here: bloom-filter acknowledgements,
|
||||
//! resend / outgoing buffer, incoming reorder buffer, Store-based recovery.
|
||||
//! This is detection only — an out-of-order message is still delivered to
|
||||
//! the application, but the gap it implies is reported.
|
||||
//!
|
||||
//! The same references also show who received our messages: a peer that names
|
||||
//! one of ours must have had it. Nothing is sent back — the acknowledgement
|
||||
//! rides on whatever the peer says next — so a silent peer never acknowledges,
|
||||
//! and neither does one that speaks after our message has dropped out of its
|
||||
//! [`CAUSAL_HISTORY_LEN`]-entry frontier.
|
||||
//!
|
||||
//! State is in-memory and session-scoped, matching the crate's current
|
||||
//! in-memory MLS state.
|
||||
|
||||
@@ -72,6 +79,21 @@ pub struct MissingMessage {
|
||||
pub frontier: Frontier,
|
||||
}
|
||||
|
||||
/// A peer acknowledging one of our messages: it named that message in the
|
||||
/// causal history of a message it sent, so it held ours at the time.
|
||||
///
|
||||
/// Evidence of *delivery to a peer's client*, not of a human reading it.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct DeliveryAck {
|
||||
pub conversation_id: String,
|
||||
/// The message of ours the peer acknowledged.
|
||||
pub message_id: String,
|
||||
/// The acknowledging peer's `sender_id`, verbatim off the wire —
|
||||
/// self-asserted like [`Frontier::sender_id`], not bound to the MLS
|
||||
/// identity that sent the payload.
|
||||
pub acked_by: String,
|
||||
}
|
||||
|
||||
/// Per-conversation causal state.
|
||||
#[derive(Debug, Default)]
|
||||
struct ConvoState {
|
||||
@@ -84,6 +106,11 @@ struct ConvoState {
|
||||
frontiers: VecDeque<Frontier>,
|
||||
/// Missing IDs already reported, so a gap is surfaced exactly once.
|
||||
reported_missing: HashSet<Frontier>,
|
||||
/// IDs of messages we authored: a reference to one is an acknowledgement.
|
||||
own: HashSet<String>,
|
||||
/// Which peers have acknowledged each of our messages, so each is
|
||||
/// surfaced exactly once.
|
||||
acked_by: HashMap<String, HashSet<String>>,
|
||||
}
|
||||
|
||||
impl ConvoState {
|
||||
@@ -102,6 +129,9 @@ struct Inner {
|
||||
convos: HashMap<String, ConvoState>,
|
||||
/// Detected gaps, drained by the client (future #97 event bus).
|
||||
missing: Vec<MissingMessage>,
|
||||
/// Detected acknowledgements of our own messages, drained alongside
|
||||
/// `missing`.
|
||||
acks: Vec<DeliveryAck>,
|
||||
}
|
||||
|
||||
/// Session-scoped causal-history store shared by every `GroupV1Convo`
|
||||
@@ -143,7 +173,9 @@ impl CausalHistoryStore {
|
||||
.collect();
|
||||
|
||||
// Our own message joins the seen-set so it appears in our future
|
||||
// causal history (and, later, so we can ack peers' references to it).
|
||||
// causal history, and the own-set so a peer referencing it back is
|
||||
// recognised as acknowledging this send.
|
||||
state.own.insert(message_id.clone());
|
||||
state.record_seen(frontier);
|
||||
|
||||
ReliablePayload {
|
||||
@@ -166,7 +198,11 @@ impl CausalHistoryStore {
|
||||
payload: &ReliablePayload,
|
||||
) -> Vec<MissingMessage> {
|
||||
let mut inner = self.inner.borrow_mut();
|
||||
let Inner { convos, missing } = &mut *inner;
|
||||
let Inner {
|
||||
convos,
|
||||
missing,
|
||||
acks,
|
||||
} = &mut *inner;
|
||||
let state = convos.entry(conversation_id.to_owned()).or_default();
|
||||
|
||||
// Lamport merge: the next local send will be strictly greater than
|
||||
@@ -175,6 +211,23 @@ impl CausalHistoryStore {
|
||||
|
||||
let mut detected = Vec::new();
|
||||
for entry in &payload.causal_history {
|
||||
// The sender named one of ours, so it has it. Reported once per
|
||||
// peer per message, and never for the message's own author.
|
||||
if state.own.contains(&entry.message_id)
|
||||
&& payload.sender_id != entry.sender_id
|
||||
&& state
|
||||
.acked_by
|
||||
.entry(entry.message_id.clone())
|
||||
.or_default()
|
||||
.insert(payload.sender_id.clone())
|
||||
{
|
||||
acks.push(DeliveryAck {
|
||||
conversation_id: conversation_id.to_owned(),
|
||||
message_id: entry.message_id.clone(),
|
||||
acked_by: payload.sender_id.clone(),
|
||||
});
|
||||
}
|
||||
|
||||
let frontier = Frontier::new(entry.sender_id.clone(), entry.message_id.clone());
|
||||
if !state.seen.contains(&frontier) && state.reported_missing.insert(frontier.clone()) {
|
||||
let m = MissingMessage {
|
||||
@@ -201,6 +254,11 @@ impl CausalHistoryStore {
|
||||
pub fn take_missing(&self) -> Vec<MissingMessage> {
|
||||
std::mem::take(&mut self.inner.borrow_mut().missing)
|
||||
}
|
||||
|
||||
/// Drain all acknowledgements of our own messages detected so far.
|
||||
pub fn take_acks(&self) -> Vec<DeliveryAck> {
|
||||
std::mem::take(&mut self.inner.borrow_mut().acks)
|
||||
}
|
||||
}
|
||||
|
||||
/// Deterministic, collision-resistant message ID.
|
||||
@@ -293,6 +351,87 @@ mod tests {
|
||||
assert_eq!(missing[0].frontier.sender_id(), "alice");
|
||||
}
|
||||
|
||||
/// Bob replies after receiving Alice's message, so his causal history
|
||||
/// names it — that reference is the acknowledgement.
|
||||
#[test]
|
||||
fn a_peer_referencing_our_message_acknowledges_it() {
|
||||
let alice = CausalHistoryStore::new();
|
||||
let bob = CausalHistoryStore::new();
|
||||
|
||||
let a1 = payload(&alice, "c", "alice", b"hello");
|
||||
bob.on_receive("c", &a1);
|
||||
let b1 = payload(&bob, "c", "bob", b"hi back");
|
||||
alice.on_receive("c", &b1);
|
||||
|
||||
assert_eq!(
|
||||
alice.take_acks(),
|
||||
vec![DeliveryAck {
|
||||
conversation_id: "c".to_owned(),
|
||||
message_id: a1.message_id.clone(),
|
||||
acked_by: "bob".to_owned(),
|
||||
}]
|
||||
);
|
||||
// Draining clears the report.
|
||||
assert!(alice.take_acks().is_empty());
|
||||
}
|
||||
|
||||
/// Every member that replies acknowledges separately, which is what lets an
|
||||
/// application list the peers that hold a message.
|
||||
#[test]
|
||||
fn each_peer_acknowledges_separately() {
|
||||
let alice = CausalHistoryStore::new();
|
||||
let bob = CausalHistoryStore::new();
|
||||
let carol = CausalHistoryStore::new();
|
||||
|
||||
let a1 = payload(&alice, "c", "alice", b"hello all");
|
||||
bob.on_receive("c", &a1);
|
||||
carol.on_receive("c", &a1);
|
||||
alice.on_receive("c", &payload(&bob, "c", "bob", b"bob here"));
|
||||
alice.on_receive("c", &payload(&carol, "c", "carol", b"carol here"));
|
||||
|
||||
let holders: Vec<String> = alice
|
||||
.take_acks()
|
||||
.into_iter()
|
||||
.filter(|a| a.message_id == a1.message_id)
|
||||
.map(|a| a.acked_by)
|
||||
.collect();
|
||||
assert_eq!(holders, vec!["bob".to_owned(), "carol".to_owned()]);
|
||||
}
|
||||
|
||||
/// Bob keeps naming the message in later sends; the application is told
|
||||
/// once.
|
||||
#[test]
|
||||
fn a_peer_acknowledges_a_message_only_once() {
|
||||
let alice = CausalHistoryStore::new();
|
||||
let bob = CausalHistoryStore::new();
|
||||
|
||||
let a1 = payload(&alice, "c", "alice", b"hello");
|
||||
bob.on_receive("c", &a1);
|
||||
alice.on_receive("c", &payload(&bob, "c", "bob", b"first reply"));
|
||||
alice.take_acks();
|
||||
alice.on_receive("c", &payload(&bob, "c", "bob", b"second reply"));
|
||||
|
||||
assert!(
|
||||
alice.take_acks().is_empty(),
|
||||
"a peer's acknowledgement of one message is reported once"
|
||||
);
|
||||
}
|
||||
|
||||
/// Carol's reply names Bob's message, not ours — nothing for us to report.
|
||||
#[test]
|
||||
fn a_reference_to_someone_elses_message_is_not_our_acknowledgement() {
|
||||
let alice = CausalHistoryStore::new();
|
||||
let bob = CausalHistoryStore::new();
|
||||
let carol = CausalHistoryStore::new();
|
||||
|
||||
let b1 = payload(&bob, "c", "bob", b"bob speaks");
|
||||
carol.on_receive("c", &b1);
|
||||
// Alice observes Carol's reply, which references Bob's message only.
|
||||
alice.on_receive("c", &payload(&carol, "c", "carol", b"carol replies"));
|
||||
|
||||
assert!(alice.take_acks().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_gap_is_reported_only_once() {
|
||||
let sender = CausalHistoryStore::new();
|
||||
|
||||
@@ -16,10 +16,19 @@ use shared_traits::IdentIdRef;
|
||||
pub type ConversationId = String;
|
||||
pub type ConversationIdRef<'a> = &'a str;
|
||||
|
||||
/// Identifies one message within a conversation, as carried by the
|
||||
/// causal-history envelope. Handed back by a send so a caller can match later
|
||||
/// observations — acknowledgements, gaps — to the message that produced them.
|
||||
pub type MessageId = String;
|
||||
|
||||
/// Behaviour shared by every conversation kind.
|
||||
pub(crate) trait Convo<S: ExternalServices>: Identified + Send {
|
||||
fn send_content(&mut self, cx: &mut ServiceContext<S>, content: &[u8])
|
||||
-> Result<(), ChatError>;
|
||||
/// Encrypt and publish `content`, returning the id assigned to it.
|
||||
fn send_content(
|
||||
&mut self,
|
||||
cx: &mut ServiceContext<S>,
|
||||
content: &[u8],
|
||||
) -> Result<MessageId, ChatError>;
|
||||
|
||||
/// Decrypts and processes an incoming encrypted frame.
|
||||
///
|
||||
|
||||
@@ -2,7 +2,7 @@ use chat_proto::logoschat::encryption::EncryptedPayload;
|
||||
use shared_traits::IdentIdRef;
|
||||
|
||||
use crate::{
|
||||
ChatError, ExternalServices,
|
||||
ChatError, ExternalServices, MessageId,
|
||||
conversation::{ConversationIdRef, Convo, GroupConvo, GroupV1Convo, Identified},
|
||||
service_context::ServiceContext,
|
||||
};
|
||||
@@ -43,7 +43,7 @@ where
|
||||
&mut self,
|
||||
cx: &mut ServiceContext<S>,
|
||||
content: &[u8],
|
||||
) -> Result<(), super::ChatError> {
|
||||
) -> Result<MessageId, ChatError> {
|
||||
self.inner_group.send_content(cx, content)
|
||||
}
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@ use shared_traits::IdentIdRef;
|
||||
use std::collections::VecDeque;
|
||||
use tracing::debug;
|
||||
|
||||
use crate::conversation::ConversationIdRef;
|
||||
use crate::conversation::{ConversationIdRef, MessageId};
|
||||
use crate::inbox_v2::MlsProvider;
|
||||
use crate::service_context::{ExternalServices, ServiceContext};
|
||||
|
||||
@@ -177,7 +177,7 @@ impl GroupV1Convo {
|
||||
&mut self,
|
||||
content: &[u8],
|
||||
cx: &mut ServiceContext<S>,
|
||||
) -> Result<(), ChatError> {
|
||||
) -> Result<MessageId, ChatError> {
|
||||
let sender_id = cx.mls_identity.id().as_str();
|
||||
let reliable = cx.causal.on_send(&self.convo_id, sender_id, content);
|
||||
let wire = reliable.encode_to_vec();
|
||||
@@ -188,7 +188,8 @@ impl GroupV1Convo {
|
||||
.unwrap();
|
||||
|
||||
let msg_bytes = mls_message_out.to_bytes().unwrap();
|
||||
self.send_payload(cx, msg_bytes)
|
||||
self.send_payload(cx, msg_bytes)?;
|
||||
Ok(reliable.message_id)
|
||||
}
|
||||
|
||||
// Publish outbound payloads to the DeliveryService
|
||||
@@ -233,7 +234,7 @@ impl<S: ExternalServices> Convo<S> for GroupV1Convo {
|
||||
&mut self,
|
||||
cx: &mut ServiceContext<S>,
|
||||
content: &[u8],
|
||||
) -> Result<(), ChatError> {
|
||||
) -> Result<MessageId, ChatError> {
|
||||
self.send_message(content, cx)
|
||||
}
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@ use crate::{Content, WakeupService};
|
||||
use alloy::signers::local::PrivateKeySigner;
|
||||
use blake2::{Blake2b, Digest, digest::consts::U6};
|
||||
use chat_proto::logoschat::encryption::{EncryptedPayload, Plaintext, encrypted_payload};
|
||||
use chat_proto::logoschat::reliability::ReliablePayload;
|
||||
use de_mls::protos::de_mls::messages::v1::{
|
||||
AppMessage as AppMessageProto, MemberWelcome, app_message,
|
||||
};
|
||||
@@ -31,7 +32,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
use tracing::{info, instrument};
|
||||
|
||||
use crate::IdentityProvider;
|
||||
use crate::conversation::{ConversationIdRef, ExternalServices, ServiceContext};
|
||||
use crate::conversation::{ConversationIdRef, ExternalServices, MessageId, ServiceContext};
|
||||
use crate::{
|
||||
ConvoOutcome, DeliveryService, RegistrationService,
|
||||
conversation::{ChatError, Convo, GroupConvo, Identified},
|
||||
@@ -304,14 +305,20 @@ where
|
||||
&mut self,
|
||||
service_ctx: &mut super::ServiceContext<S>,
|
||||
content: &[u8],
|
||||
) -> Result<(), ChatError> {
|
||||
) -> Result<MessageId, ChatError> {
|
||||
let reliable = service_ctx.causal.on_send(
|
||||
&self.convo_id,
|
||||
service_ctx.mls_identity.id().as_str(),
|
||||
content,
|
||||
);
|
||||
|
||||
self.conversation.send_message(
|
||||
&service_ctx.mls_provider,
|
||||
&service_ctx.mls_identity,
|
||||
content.to_vec(),
|
||||
reliable.encode_to_vec(),
|
||||
)?;
|
||||
self.after_op(service_ctx)?;
|
||||
Ok(())
|
||||
Ok(reliable.message_id)
|
||||
}
|
||||
|
||||
#[instrument(name = "groupv2.handle_frame", skip_all, fields(user_id = %service_ctx.mls_identity.display_name()))]
|
||||
@@ -341,7 +348,7 @@ where
|
||||
self.conversation
|
||||
.poll(&service_ctx.mls_provider, &service_ctx.mls_identity);
|
||||
let events = self.after_op(service_ctx)?; // route + publish + re-arm, returns events
|
||||
Ok(self.outcome_from_events(&events))
|
||||
self.outcome_from_events(service_ctx, &events)
|
||||
}
|
||||
|
||||
#[instrument(name = "groupv2.wakeup", skip_all, fields(user_id = %ctx.mls_identity.display_name()))]
|
||||
@@ -355,7 +362,7 @@ where
|
||||
tracing::warn!(convo = %self.convo_id, "conversation requested teardown");
|
||||
}
|
||||
let events = self.after_op(ctx)?; // publish what poll produced + re-arm alarm
|
||||
Ok(self.outcome_from_events(&events))
|
||||
self.outcome_from_events(ctx, &events)
|
||||
}
|
||||
|
||||
fn members(&self) -> Result<Vec<Vec<u8>>, ChatError> {
|
||||
@@ -491,27 +498,47 @@ impl GroupV2Convo {
|
||||
Ok(events)
|
||||
}
|
||||
|
||||
fn outcome_from_events(&self, events: &[ConversationEvent]) -> ConvoOutcome {
|
||||
let content = events.iter().find_map(|evt| match evt {
|
||||
ConversationEvent::ConversationMessage(AppMessageProto {
|
||||
payload: Some(app_message::Payload::ConversationMessage(cm)),
|
||||
}) => Some(Content {
|
||||
bytes: cm.message.clone(),
|
||||
encoded_credential: cm.sender.clone(),
|
||||
}),
|
||||
_ => None,
|
||||
});
|
||||
/// Turn drained de-mls events into a [`ConvoOutcome`], unwrapping the
|
||||
/// message from its causal-history envelope.
|
||||
///
|
||||
/// An outcome holds one message and de-mls emits at most one per frame, so
|
||||
/// the first wins. A second would be dropped without being recorded as
|
||||
/// seen, leaving a later reference to report it missing.
|
||||
fn outcome_from_events<S: ExternalServices>(
|
||||
&self,
|
||||
service_ctx: &ServiceContext<S>,
|
||||
events: &[ConversationEvent],
|
||||
) -> Result<ConvoOutcome, ChatError> {
|
||||
let content = events
|
||||
.iter()
|
||||
.find_map(|evt| match evt {
|
||||
ConversationEvent::ConversationMessage(AppMessageProto {
|
||||
payload: Some(app_message::Payload::ConversationMessage(cm)),
|
||||
}) => Some(cm),
|
||||
_ => None,
|
||||
})
|
||||
.map(|cm| -> Result<Content, ChatError> {
|
||||
let reliable =
|
||||
ReliablePayload::decode(cm.message.as_slice()).map_err(ChatError::generic)?;
|
||||
service_ctx.causal.on_receive(&self.convo_id, &reliable);
|
||||
Ok(Content {
|
||||
bytes: reliable.content.to_vec(),
|
||||
encoded_credential: cm.sender.clone(),
|
||||
})
|
||||
})
|
||||
.transpose()?;
|
||||
|
||||
let members_changed = events.iter().any(|evt| {
|
||||
matches!(
|
||||
evt,
|
||||
ConversationEvent::CommitApplied(_) | ConversationEvent::WelcomeReady { .. }
|
||||
)
|
||||
});
|
||||
ConvoOutcome {
|
||||
Ok(ConvoOutcome {
|
||||
convo_id: self.convo_id.clone(),
|
||||
content,
|
||||
members_changed,
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use crate::causal_history::{CausalHistoryStore, MissingMessage};
|
||||
use crate::causal_history::{CausalHistoryStore, DeliveryAck, MissingMessage};
|
||||
use crate::conversation::{
|
||||
ConversationIdRef, DirectV1Convo, GroupV1Convo, GroupV2Convo, Identified,
|
||||
ConversationIdRef, DirectV1Convo, GroupV1Convo, GroupV2Convo, Identified, MessageId,
|
||||
};
|
||||
use crate::service_context::{ExternalServices, ServiceContext};
|
||||
use crate::types::ConvoMetadata;
|
||||
@@ -329,8 +329,16 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
|
||||
self.services.causal.take_missing()
|
||||
}
|
||||
|
||||
/// Encrypt and publish `content` to an existing conversation.
|
||||
pub fn send_content(&mut self, convo_id: &str, content: &[u8]) -> Result<(), ChatError> {
|
||||
/// Drain the acknowledgements observed since the last call: peers that
|
||||
/// referenced one of our messages, and so demonstrably hold it.
|
||||
pub fn take_acks(&self) -> Vec<DeliveryAck> {
|
||||
self.services.causal.take_acks()
|
||||
}
|
||||
|
||||
/// Encrypt and publish `content` to an existing conversation, returning the
|
||||
/// id assigned to the message so later acknowledgements can be matched to
|
||||
/// it.
|
||||
pub fn send_content(&mut self, convo_id: &str, content: &[u8]) -> Result<MessageId, ChatError> {
|
||||
if self.cached_convos.contains_key(convo_id) {
|
||||
let convo = self
|
||||
.cached_convos
|
||||
@@ -508,7 +516,7 @@ impl<S: ExternalServices> Convo<S> for ConvoTypeOwned<S> {
|
||||
&mut self,
|
||||
cx: &mut ServiceContext<S>,
|
||||
content: &[u8],
|
||||
) -> Result<(), ChatError> {
|
||||
) -> Result<MessageId, ChatError> {
|
||||
match self {
|
||||
ConvoTypeOwned::Group(group_convo) => group_convo.send_content(cx, content),
|
||||
ConvoTypeOwned::Direct(convo) => convo.send_content(cx, content),
|
||||
|
||||
@@ -122,7 +122,7 @@ impl InboxV2 {
|
||||
|
||||
/// The convo built from an invite, paired with the display class its invite
|
||||
/// type implies: `InviteType::GroupV1` carries the pairwise DirectV1 welcome,
|
||||
/// so it is `Private`; `InviteType::GroupV2` is a real group.
|
||||
/// so it is `Dm`; `InviteType::GroupV2` is a real group.
|
||||
#[instrument(name = "inboxV2.handle_frame", skip_all, fields(user_id = %service_ctx.mls_identity.display_name()))]
|
||||
pub fn handle_frame<S: ExternalServices>(
|
||||
&self,
|
||||
@@ -143,7 +143,7 @@ impl InboxV2 {
|
||||
match payload {
|
||||
InviteType::GroupV1(inv) => {
|
||||
let convo = self.handle_heavy_invite(service_ctx, inv)?;
|
||||
Ok(Some((Box::new(convo), ConversationClass::Private)))
|
||||
Ok(Some((Box::new(convo), ConversationClass::Dm)))
|
||||
}
|
||||
InviteType::GroupV2(welcome_bytes) => {
|
||||
info!("Process WelcomeMessage");
|
||||
|
||||
@@ -10,10 +10,10 @@ mod service_traits;
|
||||
mod types;
|
||||
mod utils;
|
||||
|
||||
pub use causal_history::{Frontier, MissingMessage};
|
||||
pub use causal_history::{DeliveryAck, Frontier, MissingMessage};
|
||||
pub use chat_sqlite::ChatStorage;
|
||||
pub use chat_sqlite::StorageConfig;
|
||||
pub use conversation::GroupV2Clock;
|
||||
pub use conversation::{GroupV2Clock, MessageId};
|
||||
pub use core::{ConversationId, Core};
|
||||
/// Timing/policy for GroupV2 conversations (de-mls's per-conversation config).
|
||||
/// Defaults to the de-mls library defaults; inject via
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
//! initial [`ConvoOutcome`].
|
||||
//! - [`PayloadOutcome`] — the union of the above, plus `Empty`.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use storage::ConversationKind;
|
||||
|
||||
use crate::conversation::ConversationId;
|
||||
@@ -68,9 +69,9 @@ impl From<InboxOutcome> for PayloadOutcome {
|
||||
}
|
||||
|
||||
/// Stable across protocol versions of the same conversation shape.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum ConversationClass {
|
||||
Private,
|
||||
Dm,
|
||||
Group,
|
||||
}
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "crypto"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
# External dependencies (sorted)
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "double-ratchets"
|
||||
version = "0.0.1"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[lib]
|
||||
crate-type = ["rlib"]
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "integration_tests_core"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
# [[test]]
|
||||
# name = "integration_tests_core"
|
||||
|
||||
@@ -84,6 +84,13 @@ impl TestClient {
|
||||
outcomes
|
||||
}
|
||||
|
||||
/// Poll and discard every payload waiting for this client — simulates
|
||||
/// frames the transport never delivered.
|
||||
pub fn drop_pending_payloads(&mut self) {
|
||||
let ds = self.inner.ds();
|
||||
while ds.poll().is_some() {}
|
||||
}
|
||||
|
||||
pub fn received_messages(&self) -> &[ReceivedMessage<Vec<u8>>] {
|
||||
&self.received_messages
|
||||
}
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use integration_tests_core::TestHarness;
|
||||
use libchat::{DeliveryAck, MissingMessage};
|
||||
use tracing::info;
|
||||
|
||||
#[test]
|
||||
@@ -310,3 +311,145 @@ fn direct_v1_then_group_v2_reuses_key_package() {
|
||||
"raya did not join the group after a direct chat"
|
||||
);
|
||||
}
|
||||
|
||||
/// End-to-end causal-history gap detection on GroupV2.
|
||||
///
|
||||
/// Saro and Raya share a GroupV2 conversation. Saro sends three messages; the
|
||||
/// second never reaches Raya. The third carries the second in its causal
|
||||
/// history, so Raya must detect and report the gap.
|
||||
#[test]
|
||||
fn missing_group_v2_message_is_detected() {
|
||||
let mut harness = TestHarness::<2>::new(|_, _| {});
|
||||
|
||||
let participants = &[&harness.raya().addr()];
|
||||
let convo_id = harness
|
||||
.saro()
|
||||
.create_group_convo_v2(participants, "", "")
|
||||
.expect("saro create group");
|
||||
|
||||
// Carry the invite through (commit, WelcomeReady, inbox routing, welcome
|
||||
// accept) until Raya has joined.
|
||||
harness.process_until_label("raya joins", |h| h.raya().convo_count() == 1);
|
||||
|
||||
// M1 is delivered normally.
|
||||
harness
|
||||
.saro()
|
||||
.send_content(&convo_id, b"first")
|
||||
.expect("saro send m1");
|
||||
harness.process_until_label("raya gets m1", |h| h.raya().check(&convo_id, b"first"));
|
||||
assert!(
|
||||
harness.raya().take_missing_messages().is_empty(),
|
||||
"no gap expected while every message is delivered"
|
||||
);
|
||||
|
||||
// M2 is published but never reaches Raya. The settle above drained Raya's
|
||||
// queue, and `send_content` publishes synchronously, so discarding what is
|
||||
// pending now drops that message and nothing of the group protocol.
|
||||
harness
|
||||
.saro()
|
||||
.send_content(&convo_id, b"second")
|
||||
.expect("saro send m2");
|
||||
harness.raya().drop_pending_payloads();
|
||||
|
||||
// M3 is delivered; its causal history references the dropped M2.
|
||||
harness
|
||||
.saro()
|
||||
.send_content(&convo_id, b"third")
|
||||
.expect("saro send m3");
|
||||
harness.process_until_label("raya gets m3", |h| h.raya().check(&convo_id, b"third"));
|
||||
|
||||
let missing: Vec<MissingMessage> = harness.raya().take_missing_messages();
|
||||
assert_eq!(missing.len(), 1, "exactly one message should be missing");
|
||||
assert_eq!(missing[0].conversation_id, convo_id);
|
||||
assert!(
|
||||
!missing[0].frontier.message_id().is_empty(),
|
||||
"the missing message must be identified"
|
||||
);
|
||||
// The causal sender hint carries the MLS identity id ("saro") — the same
|
||||
// value de-mls stamps as the message's authenticated member id — not the
|
||||
// signer id the inbox and registry key on.
|
||||
assert_eq!(
|
||||
missing[0].frontier.sender_id(),
|
||||
"saro",
|
||||
"missing-message sender hint should attribute to Saro"
|
||||
);
|
||||
|
||||
// Draining clears the report; a reported gap is not surfaced again.
|
||||
assert!(harness.raya().take_missing_messages().is_empty());
|
||||
}
|
||||
|
||||
/// End-to-end acknowledgement detection on GroupV2.
|
||||
///
|
||||
/// Saro sends a message; Raya and Pax reply. Each reply carries Saro's message
|
||||
/// in its causal history, so Saro learns both peers hold it — without either
|
||||
/// sending anything back on purpose.
|
||||
#[test]
|
||||
fn replies_acknowledge_the_message_they_were_sent_after() {
|
||||
let mut harness = TestHarness::<3>::new(|_, _| {});
|
||||
|
||||
let participants = &[&harness.raya().addr(), &harness.pax().addr()];
|
||||
let convo_id = harness
|
||||
.saro()
|
||||
.create_group_convo_v2(participants, "", "")
|
||||
.expect("saro create group");
|
||||
|
||||
harness.process_until_label("peers join", |h| {
|
||||
h.raya().convo_count() == 1 && h.pax().convo_count() == 1
|
||||
});
|
||||
|
||||
let message_id = harness
|
||||
.saro()
|
||||
.send_content(&convo_id, b"anyone there?")
|
||||
.expect("saro send");
|
||||
harness.process_until_label("peers get the message", |h| {
|
||||
h.raya().check(&convo_id, b"anyone there?") && h.pax().check(&convo_id, b"anyone there?")
|
||||
});
|
||||
assert!(
|
||||
harness.saro().take_acks().is_empty(),
|
||||
"holding a message is only observable once the peer sends"
|
||||
);
|
||||
|
||||
// Each reply names Saro's message in its causal history.
|
||||
harness
|
||||
.raya()
|
||||
.send_content(&convo_id, b"raya here")
|
||||
.expect("raya reply");
|
||||
harness
|
||||
.pax()
|
||||
.send_content(&convo_id, b"pax here")
|
||||
.expect("pax reply");
|
||||
harness.process_until_label("saro gets both replies", |h| {
|
||||
h.saro().check(&convo_id, b"raya here") && h.saro().check(&convo_id, b"pax here")
|
||||
});
|
||||
|
||||
let acks: Vec<DeliveryAck> = harness.saro().take_acks();
|
||||
let mut holders: Vec<&str> = acks
|
||||
.iter()
|
||||
.filter(|a| a.conversation_id == convo_id && a.message_id == message_id)
|
||||
.map(|a| a.acked_by.as_str())
|
||||
.collect();
|
||||
holders.sort_unstable();
|
||||
assert_eq!(
|
||||
holders,
|
||||
vec!["pax", "raya"],
|
||||
"both peers that replied should be reported as holding the message"
|
||||
);
|
||||
|
||||
// Draining clears the reports, and neither peer acknowledges twice.
|
||||
assert!(harness.saro().take_acks().is_empty());
|
||||
harness
|
||||
.raya()
|
||||
.send_content(&convo_id, b"raya again")
|
||||
.expect("raya second reply");
|
||||
harness.process_until_label("saro gets the second reply", |h| {
|
||||
h.saro().check(&convo_id, b"raya again")
|
||||
});
|
||||
assert!(
|
||||
harness
|
||||
.saro()
|
||||
.take_acks()
|
||||
.iter()
|
||||
.all(|a| a.message_id != message_id),
|
||||
"a peer acknowledges one message only once"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ name = "shared-traits"
|
||||
description = "Shared traits for the Logos Ecosystem"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
crypto = { workspace = true }
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "chat-sqlite"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
description = "SQLite storage implementation for libchat"
|
||||
|
||||
[dependencies]
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "storage"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
description = "Shared storage layer for libchat"
|
||||
|
||||
[dependencies]
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "logos-generic-chat"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[lib]
|
||||
crate-type = ["rlib"]
|
||||
|
||||
@@ -6,8 +6,9 @@ use components::{ThreadedWakeupService, WakeupEvent};
|
||||
use crossbeam_channel::{Receiver, Sender, select};
|
||||
use crypto::Ed25519VerifyingKey;
|
||||
use libchat::{
|
||||
ConversationId, ConvoMetadata, ConvoOutcome, Core, DeliveryService, GroupV2Config, IdentId,
|
||||
IdentIdRef, InboxOutcome, PayloadOutcome, RegistrationService,
|
||||
ConversationId, ConvoMetadata, ConvoOutcome, Core, DeliveryAck, DeliveryService, GroupV2Config,
|
||||
IdentId, IdentIdRef, InboxOutcome, MessageId, MissingMessage, PayloadOutcome,
|
||||
RegistrationService,
|
||||
};
|
||||
use logos_account::{AccountDirectory, resolve_device_ids};
|
||||
use parking_lot::Mutex;
|
||||
@@ -275,7 +276,14 @@ where
|
||||
|
||||
/// Encrypt and send `content` to an existing conversation. The core
|
||||
/// publishes the outbound envelope.
|
||||
pub fn send_message(&mut self, convo_id: &str, content: &[u8]) -> Result<(), ClientError> {
|
||||
///
|
||||
/// Returns the message's id, which later [`Event::MessageAcked`] events
|
||||
/// carry — hold onto it to show which peers have the message.
|
||||
pub fn send_message(
|
||||
&mut self,
|
||||
convo_id: &str,
|
||||
content: &[u8],
|
||||
) -> Result<MessageId, ClientError> {
|
||||
self.core
|
||||
.lock()
|
||||
.send_content(convo_id, content)
|
||||
@@ -347,7 +355,7 @@ fn worker_loop<T, R, S: ChatStore + 'static>(
|
||||
};
|
||||
let events = {
|
||||
let mut core = core.lock();
|
||||
match core.handle_payload(&bytes) {
|
||||
let mut events = match core.handle_payload(&bytes) {
|
||||
Ok(outcome) => events_from_inbound(outcome, &directory),
|
||||
Err(e) => {
|
||||
tracing::warn!("inbound handle_payload failed: {e:?}");
|
||||
@@ -355,7 +363,10 @@ fn worker_loop<T, R, S: ChatStore + 'static>(
|
||||
message: e.to_string(),
|
||||
}]
|
||||
}
|
||||
}
|
||||
};
|
||||
events.extend(delivery_ack_events(core.take_acks(), &directory));
|
||||
events.extend(missing_events(core.take_missing_messages(), &directory));
|
||||
events
|
||||
};
|
||||
for event in events {
|
||||
if event_tx.send(event).is_err() {
|
||||
@@ -368,12 +379,18 @@ fn worker_loop<T, R, S: ChatStore + 'static>(
|
||||
return; // wakeup service's sender dropped
|
||||
};
|
||||
// A wakeup can drive the steward's own commit, so it yields events too.
|
||||
let events = match core.lock().wakeup(&convo_id) {
|
||||
Ok(outcome) => events_from_inbound(outcome, &directory),
|
||||
Err(e) => {
|
||||
tracing::warn!("wakeup failed: {e:?}");
|
||||
Vec::new()
|
||||
}
|
||||
let events = {
|
||||
let mut core = core.lock();
|
||||
let mut events = match core.wakeup(&convo_id) {
|
||||
Ok(outcome) => events_from_inbound(outcome, &directory),
|
||||
Err(e) => {
|
||||
tracing::warn!("wakeup failed: {e:?}");
|
||||
Vec::new()
|
||||
}
|
||||
};
|
||||
events.extend(delivery_ack_events(core.take_acks(), &directory));
|
||||
events.extend(missing_events(core.take_missing_messages(), &directory));
|
||||
events
|
||||
};
|
||||
for event in events {
|
||||
if event_tx.send(event).is_err() {
|
||||
@@ -398,6 +415,57 @@ fn events_from_inbound(result: PayloadOutcome, directory: &impl AccountDirectory
|
||||
}
|
||||
}
|
||||
|
||||
/// Map the acknowledgements the core observed while processing one payload onto
|
||||
/// [`Event::MessageAcked`], one per peer per message.
|
||||
///
|
||||
/// Drained from the same place as [`missing_events`]: the causal history of the
|
||||
/// message just processed is what carried the acknowledgement.
|
||||
fn delivery_ack_events(acks: Vec<DeliveryAck>, directory: &impl AccountDirectory) -> Vec<Event> {
|
||||
acks.into_iter()
|
||||
.map(|a| Event::MessageAcked {
|
||||
convo_id: Arc::from(a.conversation_id),
|
||||
message_id: a.message_id,
|
||||
acked_by: sender_hint(directory, &a.acked_by),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Map the causal-history gaps the core detected while processing one payload
|
||||
/// onto [`Event::MessageMissing`].
|
||||
///
|
||||
/// Drained right after each drive of the core, so a gap arrives with the batch
|
||||
/// of events for the message that revealed it — and after them, so a gap on a
|
||||
/// conversation this payload just started still follows its
|
||||
/// [`Event::ConversationStarted`].
|
||||
fn missing_events(missing: Vec<MissingMessage>, directory: &impl AccountDirectory) -> Vec<Event> {
|
||||
missing
|
||||
.into_iter()
|
||||
.map(|m| Event::MessageMissing {
|
||||
convo_id: Arc::from(m.conversation_id),
|
||||
message_id: m.frontier.message_id().to_owned(),
|
||||
sender_hint: sender_hint(directory, m.frontier.sender_id()),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Resolve a participant a causal-history observation named — the author of a
|
||||
/// message we never saw, or the peer acknowledging one of ours.
|
||||
///
|
||||
/// Same credential decoding as a delivered message's sender, but the claim is
|
||||
/// self-asserted rather than authenticated, so an unconfirmable account yields
|
||||
/// the device alone rather than dropping the observation. `None` when the value
|
||||
/// is not a credential at all.
|
||||
fn sender_hint(directory: &impl AccountDirectory, encoded: &str) -> Option<MessageSender> {
|
||||
let (device, claim) = parse_credential(directory, encoded.as_bytes()).ok()?;
|
||||
Some(MessageSender {
|
||||
account: match claim {
|
||||
AccountClaim::Verified(account) => Some(account),
|
||||
AccountClaim::None | AccountClaim::Unverified(_) => None,
|
||||
},
|
||||
local_identity: device,
|
||||
})
|
||||
}
|
||||
|
||||
/// Interpret a hex account address as an Ed25519 account verifying key.
|
||||
fn account_key_from_hex(addr: &str) -> Option<Ed25519VerifyingKey> {
|
||||
let bytes: [u8; 32] = hex::decode(addr).ok()?.try_into().ok()?;
|
||||
@@ -601,10 +669,11 @@ mod sender_check_tests {
|
||||
use logos_account::{DeviceSet, SignedDeviceBundle};
|
||||
|
||||
use super::{
|
||||
GroupMember, MessageSender, SenderError, decode_sender, dedup_members, member_key,
|
||||
roster_member,
|
||||
Event, GroupMember, MessageSender, SenderError, decode_sender, dedup_members,
|
||||
delivery_ack_events, member_key, missing_events, roster_member,
|
||||
};
|
||||
use crate::delegate::DelegateCredential;
|
||||
use libchat::{DeliveryAck, Frontier, MissingMessage};
|
||||
|
||||
/// In-test account → device directory. Holds device id sets keyed by the hex
|
||||
/// account key, and can be made to fail to simulate a directory outage.
|
||||
@@ -913,4 +982,112 @@ mod sender_check_tests {
|
||||
vec![committed]
|
||||
);
|
||||
}
|
||||
|
||||
/// A gap reported by the causal history, as the core hands it over: the
|
||||
/// sender hint travels in the same encoding a message's credential does.
|
||||
fn gap(sender_hint: &str) -> MissingMessage {
|
||||
MissingMessage {
|
||||
conversation_id: "convo".to_owned(),
|
||||
frontier: Frontier::new(sender_hint.to_owned(), "msg-id".to_owned()),
|
||||
}
|
||||
}
|
||||
|
||||
fn hex_cred(cred: DelegateCredential) -> String {
|
||||
hex::encode(cred.serialize())
|
||||
}
|
||||
|
||||
/// Unwrap the single `MessageMissing` a one-gap batch produces.
|
||||
fn only_missing(events: Vec<Event>) -> (String, Option<MessageSender>) {
|
||||
match <[Event; 1]>::try_from(events)
|
||||
.expect("one gap produces one event")
|
||||
.into_iter()
|
||||
.next()
|
||||
.unwrap()
|
||||
{
|
||||
Event::MessageMissing {
|
||||
convo_id,
|
||||
message_id,
|
||||
sender_hint,
|
||||
} => {
|
||||
assert_eq!(&*convo_id, "convo");
|
||||
(message_id, sender_hint)
|
||||
}
|
||||
other => panic!("expected MessageMissing, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// An account claim the directory contradicts drops a *delivered* message,
|
||||
/// but a gap is still worth reporting: the hint keeps the device and
|
||||
/// forgoes the account, since nothing about an unseen message is verifiable
|
||||
/// anyway.
|
||||
#[test]
|
||||
fn missing_message_hint_keeps_the_device_when_the_account_claim_fails() {
|
||||
let account = key();
|
||||
let endorsed = key();
|
||||
let spoofer = key();
|
||||
let dir = FakeDir::with_devices(&account, &[&endorsed]);
|
||||
let cred = DelegateCredential::associated(&spoofer, &hex::encode(account.as_ref()));
|
||||
|
||||
let (_, sender) = only_missing(missing_events(vec![gap(&hex_cred(cred))], &dir));
|
||||
assert_eq!(
|
||||
sender,
|
||||
Some(MessageSender {
|
||||
account: None,
|
||||
local_identity: local_id(&spoofer),
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
/// A hint that is not a credential at all still reports the gap — the
|
||||
/// message id is the part the application needs.
|
||||
#[test]
|
||||
fn missing_message_without_a_resolvable_hint_is_still_reported() {
|
||||
let (message_id, sender) =
|
||||
only_missing(missing_events(vec![gap("saro")], &FakeDir::default()));
|
||||
assert_eq!(message_id, "msg-id");
|
||||
assert_eq!(sender, None);
|
||||
}
|
||||
|
||||
/// One acknowledgement per peer per message, each naming the peer an
|
||||
/// application would list against the message.
|
||||
#[test]
|
||||
fn acks_name_the_peers_that_hold_the_message() {
|
||||
let account = key();
|
||||
let device = key();
|
||||
let dir = FakeDir::with_devices(&account, &[&device]);
|
||||
let peer = DelegateCredential::associated(&device, &hex::encode(account.as_ref()));
|
||||
|
||||
let events = delivery_ack_events(
|
||||
vec![DeliveryAck {
|
||||
conversation_id: "convo".to_owned(),
|
||||
message_id: "msg-id".to_owned(),
|
||||
acked_by: hex_cred(peer),
|
||||
}],
|
||||
&dir,
|
||||
);
|
||||
|
||||
match <[Event; 1]>::try_from(events)
|
||||
.expect("one ack produces one event")
|
||||
.into_iter()
|
||||
.next()
|
||||
.unwrap()
|
||||
{
|
||||
Event::MessageAcked {
|
||||
convo_id,
|
||||
message_id,
|
||||
acked_by,
|
||||
} => {
|
||||
assert_eq!(&*convo_id, "convo");
|
||||
assert_eq!(message_id, "msg-id");
|
||||
assert_eq!(
|
||||
acked_by,
|
||||
Some(MessageSender {
|
||||
account: Some(local_id(&account)),
|
||||
local_identity: local_id(&device),
|
||||
})
|
||||
);
|
||||
}
|
||||
other => panic!("expected MessageAcked, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -38,6 +38,36 @@ pub enum Event {
|
||||
content: Vec<u8>,
|
||||
sender: MessageSender,
|
||||
},
|
||||
/// A peer acknowledged a message this client sent: it referenced that
|
||||
/// message in the causal history of a message of its own, so it held ours
|
||||
/// when it sent. `message_id` is the id the send returned.
|
||||
///
|
||||
/// Evidence of delivery to the peer's client, not of a human reading it.
|
||||
/// The acknowledgement is passive — nothing is sent back on purpose — so a
|
||||
/// peer that never sends never acknowledges, and an application should
|
||||
/// treat the absence of one as "not confirmed" rather than "not delivered".
|
||||
///
|
||||
/// `acked_by` is resolved from the peer's self-asserted `sender_id` and is
|
||||
/// **not authenticated**; see [`Self::MessageMissing`]'s `sender_hint`.
|
||||
/// `None` when it could not be resolved to a device.
|
||||
MessageAcked {
|
||||
convo_id: Arc<str>,
|
||||
message_id: String,
|
||||
acked_by: Option<MessageSender>,
|
||||
},
|
||||
/// A message this client never received, revealed by the causal history of
|
||||
/// one that did arrive. Detection only — nothing is fetched or replayed,
|
||||
/// and the gap is reported once.
|
||||
///
|
||||
/// `sender_hint` is the author the *referencing* peer named, resolved the
|
||||
/// same way as [`Self::MessageReceived`]'s sender but **not authenticated**:
|
||||
/// nothing about a message we never saw can be verified, so treat it as a
|
||||
/// display hint. `None` when the hint could not be resolved to a device.
|
||||
MessageMissing {
|
||||
convo_id: Arc<str>,
|
||||
message_id: String,
|
||||
sender_hint: Option<MessageSender>,
|
||||
},
|
||||
/// A commit changed a conversation's membership.
|
||||
ConversationMembersChanged {
|
||||
convo_id: Arc<str>,
|
||||
|
||||
@@ -15,7 +15,8 @@ pub use event::{Event, MessageSender};
|
||||
// Re-export types callers need to interact with ChatClient.
|
||||
pub use libchat::{
|
||||
AddressedEnvelope, ChatStore, ConversationClass, ConversationId, ConvoMetadata,
|
||||
DeliveryService, GroupV2Config, IdentityProvider, RegistrationService, StorageConfig,
|
||||
DeliveryService, GroupV2Config, IdentityProvider, MessageId, RegistrationService,
|
||||
StorageConfig,
|
||||
};
|
||||
// The directory trait bounds ChatClient's registry parameter, so callers
|
||||
// writing code generic over ChatClient need it too.
|
||||
|
||||
@@ -479,3 +479,63 @@ fn group_metadata_defaults_to_empty() {
|
||||
assert_eq!(meta.name, "");
|
||||
assert_eq!(meta.desc, "");
|
||||
}
|
||||
|
||||
/// The peers that hold a sent message surface as `MessageAcked` events keyed by
|
||||
/// the id the send returned — what an application needs to list the peers that
|
||||
/// hold a message. The acknowledgement is passive: Raya and Pax only send
|
||||
/// ordinary replies, never a receipt.
|
||||
#[test]
|
||||
fn a_sent_message_is_acknowledged_by_the_peers_that_reply() {
|
||||
let bus = MessageBus::default();
|
||||
let reg = EphemeralRegistry::new();
|
||||
|
||||
let (mut saro, saro_events, saro_addr) = create_test_client(bus.clone(), reg.clone());
|
||||
let (mut raya, raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone());
|
||||
let (mut pax, pax_events, pax_addr) = create_test_client(bus.clone(), reg.clone());
|
||||
|
||||
let convo_id = saro
|
||||
.create_group_conversation(&[&raya_addr, &pax_addr], unnamed_group())
|
||||
.expect("saro create group");
|
||||
wait_for_group_started(&raya_events, "raya ConversationStarted");
|
||||
wait_for_group_started(&pax_events, "pax ConversationStarted");
|
||||
wait_for_members(&mut saro, &convo_id, &[&saro_addr, &raya_addr, &pax_addr]);
|
||||
|
||||
let message_id = saro
|
||||
.send_message(&convo_id, b"anyone there?")
|
||||
.expect("saro send");
|
||||
wait_for_message(&raya_events, b"anyone there?");
|
||||
wait_for_message(&pax_events, b"anyone there?");
|
||||
|
||||
// Ordinary replies; their causal history carries the acknowledgement.
|
||||
raya.send_message(&convo_id, b"raya here")
|
||||
.expect("raya reply");
|
||||
pax.send_message(&convo_id, b"pax here").expect("pax reply");
|
||||
|
||||
let mut holders = Vec::new();
|
||||
while holders.len() < 2 {
|
||||
let peer = wait_for_event(
|
||||
&saro_events,
|
||||
"saro MessageAcked",
|
||||
Duration::from_secs(10),
|
||||
|e| match e {
|
||||
Event::MessageAcked {
|
||||
convo_id: id,
|
||||
message_id: acked,
|
||||
acked_by,
|
||||
} if **id == *convo_id && *acked == message_id => Some(
|
||||
acked_by
|
||||
.as_ref()
|
||||
.and_then(|a| a.account.as_ref())
|
||||
.map(|a| a.as_str().to_string()),
|
||||
),
|
||||
_ => None,
|
||||
},
|
||||
);
|
||||
holders.push(peer.expect("the acknowledging peer's account should be directory-verified"));
|
||||
}
|
||||
holders.sort();
|
||||
|
||||
let mut expected = vec![raya_addr.clone(), pax_addr.clone()];
|
||||
expected.sort();
|
||||
assert_eq!(holders, expected, "both replying peers should be listed");
|
||||
}
|
||||
|
||||
@@ -54,6 +54,25 @@ where
|
||||
f(event).unwrap_or_else(|other| panic!("expected {label}, got {other:?}"))
|
||||
}
|
||||
|
||||
/// [`expect_event`] for a back-and-forth exchange, skipping acknowledgements.
|
||||
///
|
||||
/// Each reply acknowledges the message it was sent after, so `MessageAcked`
|
||||
/// lands at points a test driving one direction at a time does not control.
|
||||
fn expect_event_ignoring_acks<F, T>(events: &Receiver<Event>, label: &str, mut f: F) -> T
|
||||
where
|
||||
F: FnMut(Event) -> Result<T, Event>,
|
||||
{
|
||||
loop {
|
||||
let event = events
|
||||
.recv_timeout(Duration::from_secs(5))
|
||||
.unwrap_or_else(|_| panic!("timed out waiting for {label}"));
|
||||
if matches!(event, Event::MessageAcked { .. }) {
|
||||
continue;
|
||||
}
|
||||
return f(event).unwrap_or_else(|other| panic!("expected {label}, got {other:?}"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn direct_v1_integration() {
|
||||
let bus = MessageBus::default();
|
||||
@@ -170,7 +189,7 @@ fn direct_v1_by_account_address() {
|
||||
// though its welcome rides the InboxV2 (GroupV1 invite) path.
|
||||
let raya_convo_id = expect_event(&raya_events, "ConversationStarted", |e| match e {
|
||||
Event::ConversationStarted { convo_id, class } => {
|
||||
assert_eq!(class, ConversationClass::Private);
|
||||
assert_eq!(class, ConversationClass::Dm);
|
||||
Ok(convo_id)
|
||||
}
|
||||
other => Err(other),
|
||||
@@ -255,7 +274,7 @@ fn saro_raya_message_exchange() {
|
||||
for i in 0u8..5 {
|
||||
let msg = format!("msg {i}");
|
||||
saro.send_message(&saro_convo_id, msg.as_bytes()).unwrap();
|
||||
expect_event(
|
||||
expect_event_ignoring_acks(
|
||||
&raya_events,
|
||||
&format!("MessageReceived(msg {i})"),
|
||||
|e| match e {
|
||||
@@ -269,7 +288,7 @@ fn saro_raya_message_exchange() {
|
||||
|
||||
let reply = format!("reply {i}");
|
||||
raya.send_message(&raya_convo_id, reply.as_bytes()).unwrap();
|
||||
expect_event(
|
||||
expect_event_ignoring_acks(
|
||||
&saro_events,
|
||||
&format!("MessageReceived(reply {i})"),
|
||||
|e| match e {
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "logos-chat"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[lib]
|
||||
crate-type = ["rlib"]
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
# cargo-deny configuration — run with: cargo deny check
|
||||
|
||||
[graph]
|
||||
all-features = true
|
||||
|
||||
[advisories]
|
||||
version = 2
|
||||
# libcrux crypto crates, exact-pinned (0.0.x) via openmls_libcrux_crypto 0.3.1.
|
||||
# The fix only exists on the openmls 0.9 prerelease line, which de-mls does not
|
||||
# support yet — revisit once openmls 0.9 is stable.
|
||||
ignore = [
|
||||
"RUSTSEC-2026-0073", # libcrux-poly1305
|
||||
"RUSTSEC-2026-0075", # libcrux-ed25519
|
||||
"RUSTSEC-2026-0124", # libcrux-chacha20poly1305
|
||||
"RUSTSEC-2026-0207", # libcrux-sha3
|
||||
"RUSTSEC-2026-0208", # libcrux-sha3
|
||||
"RUSTSEC-2026-0212", # libcrux-secrets
|
||||
"RUSTSEC-2026-0209", # libcrux-aesgcm — AES-GCM, unused by our ciphersuite
|
||||
"RUSTSEC-2026-0211", # libcrux-aesgcm — AES-GCM, unused by our ciphersuite
|
||||
"RUSTSEC-2026-0210", # libcrux-aesgcm renamed
|
||||
"RUSTSEC-2024-0436", # paste, unmaintained
|
||||
"RUSTSEC-2026-0173", # proc-macro-error2, unmaintained
|
||||
]
|
||||
|
||||
[licenses]
|
||||
version = 2
|
||||
allow = [
|
||||
"MIT",
|
||||
"Apache-2.0",
|
||||
"Apache-2.0 WITH LLVM-exception",
|
||||
"BSD-2-Clause",
|
||||
"BSD-3-Clause",
|
||||
"ISC",
|
||||
"Zlib",
|
||||
"Unicode-3.0",
|
||||
"Unicode-DFS-2016",
|
||||
"CC0-1.0",
|
||||
"Unlicense",
|
||||
"BSL-1.0",
|
||||
"MPL-2.0",
|
||||
"CDLA-Permissive-2.0",
|
||||
]
|
||||
confidence-threshold = 0.8
|
||||
|
||||
# chat-proto ships no license field yet; treat it as the workspace license.
|
||||
[[licenses.clarify]]
|
||||
crate = "chat-proto"
|
||||
expression = "MIT OR Apache-2.0"
|
||||
license-files = []
|
||||
|
||||
[bans]
|
||||
multiple-versions = "warn"
|
||||
wildcards = "allow"
|
||||
|
||||
[sources]
|
||||
unknown-registry = "deny"
|
||||
unknown-git = "deny"
|
||||
allow-git = [
|
||||
"https://github.com/logos-messaging/chat_proto",
|
||||
"https://github.com/vacp2p/de-mls",
|
||||
]
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "components"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
# Workspace dependencies (sorted)
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "embedded-logos-delivery"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
# Workspace dependencies (sorted)
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
name = "logos-delivery"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
license.workspace = true
|
||||
links = "logosdelivery"
|
||||
|
||||
[dependencies]
|
||||
|
||||
Reference in New Issue
Block a user