diff --git a/PLAN.md b/PLAN.md index 9b0d3947..25110cf7 100644 --- a/PLAN.md +++ b/PLAN.md @@ -424,37 +424,80 @@ pub fn complete_memberlist(coalesced: &BTreeMap, - A Snapshot counts only from the npub whose Refounding minted the epoch. There is deliberately no owner fallback. - The member list is `coalesced Joined ∪ observed authors − banlist`, and observation counts **forward only** (an author re-enters on activity newer than their latest Leave/Kick/Ban). A Grant holder with `ms == 0` is present by construction. -### 8.4 Chat plane (`chat.rs`) +### 8.4 Chat plane (`chat.rs`) — implemented in M4 + +Kinds (CORD-02 Appendix B): `9` message, `1111` NIP-22 comment, `7` NIP-25 reaction, +`5` NIP-09 delete, `3302` edit, `3310` WebXDC peer signal, `23311` ephemeral typing. ```rust +pub struct ChatRumor { id, author, kind, channel, epoch, at_ms, content, + expiration: Option, action: ChatAction } +pub enum ChatAction { + Message { reply_to: Option, thread_root: Option }, + Reaction { target: EventId, emoji: String }, + Edit { target: EventId, content: String }, + Delete { target: EventId, target_kind: Option }, + Typing, + Opaque, +} +pub struct ReplyRef { id: EventId, author: Option } +pub struct Target { reply: ReplyRef, kind: u16 } // the wire commits the target's kind + +pub fn build_message(author, channel, epoch, content, quote: Option<&ReplyRef>, at_ms) -> UnsignedEvent; +pub fn build_comment(author, channel, epoch, content, parent: &Target, root: Option<&Target>, at_ms) -> UnsignedEvent; +pub fn build_reaction(author, channel, epoch, target: &Target, emoji: &str, at_ms) -> UnsignedEvent; +pub fn build_edit(author, channel, epoch, target: EventId, content: &str, at_ms) -> UnsignedEvent; +pub fn build_delete(author, channel, epoch, target: EventId, target_kind: Option, at_ms) -> UnsignedEvent; +pub fn build_typing(author, channel, epoch, at_ms) -> UnsignedEvent; + +pub fn seal_rumor(rumor: &UnsignedEvent, group: &GroupKey, author: &Keys, ephemeral: bool) + -> Result<(Event, Keys), ChatError>; +pub fn open(wrap: &Event, group: &GroupKey, channel: &ChannelId, epoch: Epoch) + -> Result<(OpenedStream, ChatRumor), ChatError>; +pub fn plane_keys(held: &[(Epoch, [u8; 32])], channel: &ChannelId) -> Result>; +pub fn fold(rumors: &[ChatRumor]) -> Vec; + pub struct ChatMessage { - pub id: EventId, // recomputed rumor id + pub id: EventId, pub author: PublicKey, pub channel: ChannelId, pub epoch: Epoch, - pub kind: Kind, // 9 | 1111 | 3302 | 1740 | 15 + pub kind: Kind, pub content: String, - pub media: Vec, - pub mentions: Vec, - pub reply_to: Option, // lowercase `e`/`q` - pub thread_root: Option, // uppercase `E` for 1111 + pub reply_to: Option, // a kind 9's `q`, or a comment's lowercase `e` + pub thread_root: Option, // a comment's uppercase `E` pub at_ms: u64, pub expiration: Option, - pub edited_at: Option, // folded from 3302 - pub deleted: bool, // folded from 5 + pub edited_at: Option, + pub deleted: bool, pub reactions: BTreeMap, } ``` -Sends funnel through one function so the rules cannot drift: - -```rust -fn publish_chat(store, client, community, channel, epoch, group, rumor, at_ms, ephemeral) -> Task, Error>>; -``` - -It builds the seal + wrap, mirrors the NIP-40 tag onto the wrap for durable kinds, publishes via `send_event(..).to(relays)`, retains the ephemeral wrap key for later NIP-09 scrubbing, and locally echoes its own wrap through the same ingest path so send-then-read works without waiting on a relay round-trip. - -Disappearing messages (CORD-08) live here: `message_expiration` is read from the folded metadata, `["expiration", created_at + t]` is attached to every durable Chat rumor and to the wrap, kinds 5 and 1740 are exempt, ingest refuses an already-expired rumor, a periodic sweep purges stored ones, and the kind 1740 timer notice renders only when its author holds `MANAGE_METADATA`. +- `open` returns the `OpenedStream` next to the typed rumor because the two halves go + different ways: the caller caches the envelope, and folds the rumor. +- Ordering is `(at_ms, id)` everywhere, ties on the lower inner rumor id. The fold emits + newest first; mutations replay ascending on `(at_ms, Reverse(id))` so the last one + applied wins — the highest `at_ms` and, between equal ones, the lower id. A deletion is + terminal: a later edit never revives it. +- **M4 honors a delete only from the message's own author.** A moderator delete (a `vac` + citation under `MANAGE_MESSAGES`) needs the roster, so the builder's `citation`, the + fold's `can_delete` predicate and its tests land with M5. Failing closed here loses a + moderator's reach, never a member's authorship. +- `kind 15` is coop's own file-message convention, outside the CORD registry — it is + accepted on the read side so a second coop device's files are not dropped, and + `send_file` lands with the registry. +- A reference's author slot is a SHOULD on the wire, so it is optional. NIP-25 nonetheless + makes `p` a requirement, so a builder must be handed a `Target` built from the message it + acts on and never one whose author is empty, or a peer that requires the tag drops the result. +- `ms` orders a page but cannot page within one: a relay's `until` filter is second-granular, + so the cursor step below is what has to cope with a boundary second. +- `seal_rumor` gates the kind at publish and mirrors a NIP-40 `expiration` onto the wrap + (CORD-08 §2). The timer's policy — ingest refusal, the sweep, kind 1740 — is M8. +- **`media` and `mentions` are deliberately absent.** Both are pure post-processing of + `content` by `common` (`extract_and_remove_media_urls`, `NostrParser`) and both return a + gpui type, and a protocol crate does not take a UI dependency for a derived field. They + land with the first consumer that renders them. ### 8.5 Invites (`invite.rs`) @@ -514,11 +557,21 @@ Three layers, no new storage engine: ```rust pub async fn cache_rumor(database: &dyn NostrDatabase, channel: &ChannelId, opened: &OpenedStream) -> Result<()>; pub async fn query_rumors(database: &dyn NostrDatabase, channel: &ChannelId, until: Option, limit: usize) -> Result>; +pub async fn backfill(client: &Client, database: &dyn NostrDatabase, channel: &ChannelId, + held: &[(Epoch, [u8; 32])], until: Option, limit: usize) + -> Result>; ``` `query_rumors` returns `UnsignedEvent`, not `Event`: the cached payload *is* a rumor, which is also what `OpenedStream` carries, so the caller never has to re-parse. -**Deferred to M4:** the relay-paging `backfill`. It is network history paging whose "step past the same-second wall" policy belongs with the sync engine, and M1 has no subscription to test it against. +**Landed in M4:** `backfill` — newest-first relay paging across every held epoch. It derives +every held epoch's plane key once, fetches `kinds [1059, 21059]` by all of those addresses in +one filter with an inclusive `until`, opens each wrap against the plane whose address it +carries, caches it, and pages until the page is short of the limit, adds nothing new, or the +cursor cannot advance. That last case is real: `until` has second granularity, so a page that +begins and ends inside one boundary second has nowhere left to step and its remainder stays +unreachable until a relay serves it. Capped at `MAX_PAGES` so a relay that only ever repeats +itself cannot loop a client forever. 3. **Community state** — one local document per community, `Kind::ApplicationSpecificData` with `["d", "concord/"]`: @@ -673,6 +726,11 @@ pub fn pin(&self, id: EventId, cx: &App) -> Task>; // vsk 11, `CommunityEvent` and `ChannelEvent` mirror `ChatEvent`: one variant per thing the UI has to react to (`Updated`, `Members`, `Added`, `Removed`, `Dissolved`, `Error`, plus channel-level `Incoming`, `Reload`). +Every send funnels through one function so the rules cannot drift: it seals and wraps the +rumor, mirrors any NIP-40 `expiration` onto the wrap, publishes via `send_event(..).to(relays)`, +retains the ephemeral wrap key for a later NIP-09 scrub, and echoes its own wrap through the +same ingest path so send-then-read never waits on a relay round-trip. + ## 11. Integration with existing crates 1. **`crates/chat/src/lib.rs` — required fix, moved from M2 to the milestone that first subscribes.** `handle_notifications` currently treats *every* kind 1059 event as a NIP-59 gift wrap for the current user and pushes an unwrap failure into the trash. Concord wraps are kind 1059 with an ephemeral `p` tag, so they would flood the trash and leak error toasts. Route by `subscription_id` from `RelayMessage::Event` against `sub_id1`/`sub_id2`, and drop the `if rumor.tags.is_empty()` recipient heuristic. **M2 and M3 did not apply it**: the crate has no subscription and no `ConcordRegistry` yet, so no concord wrap can reach that handler and the change would be untestable. It lands with the sync engine (§10), as does the `concord::init` wiring in `desktop` and `web`. @@ -697,7 +755,9 @@ Each of these has burned a real implementation, or is a documented cross-client - Refuse to write a Pin List from a list the writer could not read. - Enforce the NIP-44 65 535-byte cap at every layer before publishing, and the 5-relay / 256-channel / 50-membership / 100-roles / 64-roles-per-member / 500-banlist / 25-pin caps at their ingest and write points. - Lowercase hex only; x-only pubkeys only; no version tag anywhere. -- **Enforced in M3:** a Role's `role_id` is its own coordinate and never 0; a Grant's `member` hashes to its coordinate; a `vsk 4` sits at this community's banlist locator; a banned npub's editions are dropped and a grant naming them carries no rank; a revocation carries a citation; the 100-role cap keeps the lowest ids *after* authorization; a below-floor edition is never a candidate. Still owed to M8's audit: the byte caps, the 256-channel and 25-pin caps, and the write-side counterparts. +- **Enforced in M3:** a Role's `role_id` is its own coordinate and never 0; a Grant's `member` hashes to its coordinate; a `vsk 4` sits at this community's banlist locator; a banned npub's editions are dropped and a grant naming them carries no rank; a revocation carries a citation; the 100-role cap keeps the lowest ids *after* authorization; a below-floor edition is never a candidate. +- **Enforced in M4:** the chat plane's encrypted-seal requirement, at both publish and open; `channel` **and** `epoch` strict-equal to the plane whose key opened the wrap; a retired or unregistered rumor kind rejected on both sides; a target bearing tag that appears twice rejected outright; and a delete honored only from the message's own author. +- Still owed to M8's audit: the byte caps, the 256-channel and 25-pin caps, the guestbook's future-clock and Snapshot rules, and the write-side counterparts. ## 13. Milestones @@ -706,23 +766,27 @@ Each of these has burned a real implementation, or is a documented cross-client | M0 | Crate skeleton, `derive.rs`, golden vectors, workspace wiring | ✅ `cargo test -p concord` pins every derivation; all labels match Appendix A.6 | | M1 | `stream.rs` + `store.rs` | ✅ seal/wrap/open round-trips for both seal forms; hostile wraps rejected in the documented order; the local cache reads back with the group key gone | | M2 | `edition.rs` + `control.rs` genesis + `store.rs` state document | ✅ `cargo test -p concord` (7 tests): `edition_hash` reproduces the cross-client vector `2daf42e6…`, and a community minted by one holder has both genesis wraps open for a second holder holding only the invite keys, folding to version 1 | -| M3 | Control fold + roster + metadata/channels | ✅ `cargo test -p concord` (15 tests): the chain fold, its gaps, fork tiebreak, downgrade refusal and compaction dangle are pinned; the delegation fixpoint resolves outward from the owner and refuses escalation, an unauthorized higher version, rank inversion by republish and an uncited revoke; a community minted by one holder has its metadata and channel edits fold for a second holder from the invite keys alone | -| M4 | Chat plane | send/receive/edit/delete/reaction across two identities; history pages backwards across an epoch boundary (relay `backfill` lands here); binding checks reject a foreign channel/epoch | +| M3 | Control fold + roster + metadata/channels | ✅ `cargo test -p concord` (14 tests): the chain fold, its gaps, fork tiebreak, downgrade refusal and compaction dangle are pinned; the delegation fixpoint resolves outward from the owner and refuses escalation, an unauthorized higher version, rank inversion by republish and an uncited revoke; a community minted by one holder has its metadata and channel edits fold for a second holder from the invite keys alone | +| M4 | Chat plane | ✅ `cargo test -p concord` (19 tests): a second holder folds a message's reactions, its author's edit and its author's delete, and ignores an edit or a delete from anybody else; a comment's root and parent survive the wire; a foreign channel, a replayed epoch, a plaintext seal, a retired kind and a duplicated target are each rejected; and history pages backwards across a rekey in order | | M5 | Guestbook + member list + moderation | join/leave/kick/ban converge to the same memberlist on both clients; every authority rule from §8.1 has a negative test | | M6 | Invites + Community List | link mint → fetch → join round-trips; revoked link refuses; direct invite lands in the recipient's giftwrap inbox via the `k` tag; a second device reconstructs membership from 13302 | | M7 | Rekeys + refounding + dissolution | a removed member stops reading after a rekey; continuity and race rules tested; a tombstone seals the community and a foreign-id tombstone is refused | | M8 | Pins + disappearing messages + hardening | pins verify from a keyless reader's view; expiry is refused at ingest and purged by the sweep; the audit of §12 is complete with a test per bullet | -Ordering is deliberately dependency-first: each milestone is usable on its own, and nothing in M2+ depends on a later milestone. +Ordering is deliberately dependency-first: each milestone is usable on its own, and nothing in M2+ depends on a later milestone. The sync engine and its GPUI wiring (§10, §11) have no row of their own because they are cross-cutting: every plane they consume has to exist first, so they follow M8, and they are the milestone that applies §11's `chat::handle_notifications` routing fix. M1 closed with `cargo test -p concord` (5 tests), `cargo clippy -p concord --all-targets`, and `cargo fmt -p concord --check` all clean. `rand` was added to the workspace pinned to the same `0.10.2` instance `nostr` already builds, so `Cargo.lock` gained no package. M2 closed the same way at 7 tests, with `serde` added to the crate's dependencies (`serde_json` promoted from dev to main for the metadata content types) — `Cargo.lock` gained no package again, only the `concord → serde` edge. -M3 closed at 15 tests with no dependency change at all, and `Cargo.lock` untouched. New: `src/roles.rs` (permissions, Role/Grant/banlist content, `CommunityRoles`, the delegation fixpoint) and, in `src/control.rs`, `ControlFold` / `fold_control`, the metadata-and-channel fold, `ControlWriter` and its `Edition` input. `EntityHead` and `Floors` moved from `store.rs` into `edition.rs`, where `fold_head` now composes `fold` and `bootstrap_head` for the floor-aware case. +M3 closed at 14 tests with no dependency change at all, and `Cargo.lock` untouched. New: `src/roles.rs` (permissions, Role/Grant/banlist content, `CommunityRoles`, the delegation fixpoint) and, in `src/control.rs`, `ControlFold` / `fold_control`, the metadata-and-channel fold, `ControlWriter` and its `Edition` input. `EntityHead` and `Floors` moved from `store.rs` into `edition.rs`, where `fold_head` now composes `fold` and `bootstrap_head` for the floor-aware case. + +M4 closed at 19 tests, again with no dependency change and `Cargo.lock` untouched — relay paging is `Client::fetch_events` over the same `NostrDatabase` the cache already used, so nothing new was needed. New: `src/chat.rs` (the whole channel plane) and, in `src/store.rs`, `backfill` plus the pure `advance` page step it is built from, which is what the paging test drives instead of a socket. `edition::canonical_decimal` became `pub(crate)` so the chat tag grammar shares one decimal check. What M3 still defers, and to what: the **sync engine's paging** driven by `ControlFold.gapped` and the **`chat::handle_notifications` routing fix** (both §10, together with the `concord::init` wiring — no concord wrap can reach that handler until the subscription exists); the **persisted banlist** and `CommunityState.banned` (M5, with the moderation API that writes it); the **role/grant/banlist write wrappers** (M5 — `ControlWriter::publish` already carries them, only the convenience surface is pending); and the **NIP-46 remote signer**, since `publish` takes `&Keys` rather than a `NostrSigner`. +What M4 still defers, and to what: **moderator deletes and the `can_delete` predicate** (M5 — M4 honors a delete only from the message's own author, so a moderator's reach is missing rather than forged); **`media`/`mentions`** on `ChatMessage` and **`send_file`** (the registry/UI milestone — the first needs a gpui type and the second needs the blob-upload path); and **the timer's policy** under the `expiration` tag that `seal_rumor` already mirrors (M8). + **M2's "created and published" is verified offline**: "published" is the two wraps existing and being openable by the invite keys, not a relay round-trip. There is no registry to publish through until §10, and a relay test would be testing the SDK, not the protocol. ## 14. Open questions and risks diff --git a/crates/concord/src/chat.rs b/crates/concord/src/chat.rs new file mode 100644 index 00000000..067bcccf --- /dev/null +++ b/crates/concord/src/chat.rs @@ -0,0 +1,854 @@ +use std::cmp::Reverse; +use std::collections::BTreeMap; +use std::fmt; + +use anyhow::Result; +use nostr_sdk::prelude::*; + +use crate::derive::channel_group_key; +use crate::edition::canonical_decimal; +use crate::stream::{ + KIND_WRAP, KIND_WRAP_EPHEMERAL, OpenedStream, SealForm, StreamError, build_rumor_ms, + build_seal, channel_binding_tags, check_channel_binding, open_wrap, resolve_ms_strict, + wrap_seal, +}; +use crate::{ChannelId, Epoch, GroupKey, decode_hex_32}; + +pub const KIND_MESSAGE: u16 = 9; +pub const KIND_COMMENT: u16 = 1111; +pub const KIND_REACTION: u16 = 7; +pub const KIND_DELETE: u16 = 5; +pub const KIND_EDIT: u16 = 3302; +pub const KIND_FILE: u16 = 15; +pub const KIND_WEBXDC: u16 = 3310; +pub const KIND_TYPING: u16 = 23311; + +const TAG_QUOTE: &str = "q"; +const TAG_TARGET: &str = "e"; +const TAG_TARGET_KIND: &str = "k"; +const TAG_ROOT: &str = "E"; +const TAG_ROOT_KIND: &str = "K"; +const TAG_ROOT_AUTHOR: &str = "P"; +const TAG_TARGET_AUTHOR: &str = "p"; +const TAG_EXPIRATION: &str = "expiration"; + +#[derive(Debug)] +pub enum ChatError { + Stream(StreamError), + NotEncryptedSealed, + UnknownKind(u16), + MissingTag(&'static str), + DuplicateTag(&'static str), + BadTag(&'static str), +} + +impl fmt::Display for ChatError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ChatError::Stream(error) => write!(f, "stream: {error}"), + ChatError::NotEncryptedSealed => write!(f, "chat rumor must ride an encrypted seal"), + ChatError::UnknownKind(kind) => write!(f, "not a chat rumor kind: {kind}"), + ChatError::MissingTag(name) => write!(f, "missing chat tag: {name}"), + ChatError::DuplicateTag(name) => write!(f, "duplicate chat tag: {name}"), + ChatError::BadTag(name) => write!(f, "malformed chat tag: {name}"), + } + } +} + +impl std::error::Error for ChatError {} + +impl From for ChatError { + fn from(error: StreamError) -> Self { + ChatError::Stream(error) + } +} + +/// A chat event another chat event refers to: a quote, a comment's parent, a +/// reaction's target. The author slot is a SHOULD on the wire, so it is optional. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplyRef { + pub id: EventId, + pub author: Option, +} + +/// A reference that also names the referenced event's kind, which a comment +/// (`K`/`k`) and a reaction (`k`) must commit on the wire. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Target { + pub reply: ReplyRef, + pub kind: u16, +} + +#[derive(Debug, Clone)] +pub enum ChatAction { + Message { + reply_to: Option, + thread_root: Option, + }, + Reaction { + target: EventId, + emoji: String, + }, + Edit { + target: EventId, + content: String, + }, + Delete { + target: EventId, + target_kind: Option, + }, + Typing, + Opaque, +} + +#[derive(Debug, Clone)] +pub struct ChatRumor { + pub id: EventId, + pub author: PublicKey, + pub kind: Kind, + pub channel: ChannelId, + pub epoch: Epoch, + pub at_ms: u64, + pub content: String, + pub expiration: Option, + pub action: ChatAction, +} + +/// A channel's timeline row, with every edit, delete and reaction folded in. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ChatMessage { + pub id: EventId, + pub author: PublicKey, + pub channel: ChannelId, + pub epoch: Epoch, + pub kind: Kind, + pub content: String, + pub reply_to: Option, + pub thread_root: Option, + pub at_ms: u64, + pub expiration: Option, + pub edited_at: Option, + pub deleted: bool, + pub reactions: BTreeMap, +} + +pub fn build_message( + author: PublicKey, + channel: &ChannelId, + epoch: Epoch, + content: &str, + quote: Option<&ReplyRef>, + at_ms: u64, +) -> UnsignedEvent { + let mut tags = channel_binding_tags(channel, epoch); + + if let Some(quote) = quote { + tags.push(reply_tag(TAG_QUOTE, quote)); + } + + build_rumor_ms(KIND_MESSAGE, author, content, tags, at_ms) +} + +/// A NIP-22 comment. `parent` is the immediate parent and `root` the thread's +/// immutable root; `None` means the parent is itself the root. +pub fn build_comment( + author: PublicKey, + channel: &ChannelId, + epoch: Epoch, + content: &str, + parent: &Target, + root: Option<&Target>, + at_ms: u64, +) -> UnsignedEvent { + let root = root.unwrap_or(parent); + let mut tags = channel_binding_tags(channel, epoch); + + tags.push(Tag::custom(TAG_ROOT_KIND, [root.kind.to_string()])); + tags.push(reply_tag(TAG_ROOT, &root.reply)); + if let Some(root_author) = root.reply.author { + tags.push(Tag::custom(TAG_ROOT_AUTHOR, [root_author.to_hex()])); + } + + tags.push(Tag::custom(TAG_TARGET_KIND, [parent.kind.to_string()])); + tags.push(reply_tag(TAG_TARGET, &parent.reply)); + if let Some(parent_author) = parent.reply.author { + tags.push(Tag::custom(TAG_TARGET_AUTHOR, [parent_author.to_hex()])); + } + + build_rumor_ms(KIND_COMMENT, author, content, tags, at_ms) +} + +pub fn build_reaction( + author: PublicKey, + channel: &ChannelId, + epoch: Epoch, + target: &Target, + emoji: &str, + at_ms: u64, +) -> UnsignedEvent { + let mut tags = channel_binding_tags(channel, epoch); + + tags.push(Tag::custom(TAG_TARGET, [target.reply.id.to_hex()])); + if let Some(target_author) = target.reply.author { + tags.push(Tag::custom(TAG_TARGET_AUTHOR, [target_author.to_hex()])); + } + tags.push(Tag::custom(TAG_TARGET_KIND, [target.kind.to_string()])); + + build_rumor_ms(KIND_REACTION, author, emoji, tags, at_ms) +} + +pub fn build_edit( + author: PublicKey, + channel: &ChannelId, + epoch: Epoch, + target: EventId, + content: &str, + at_ms: u64, +) -> UnsignedEvent { + let mut tags = channel_binding_tags(channel, epoch); + tags.push(Tag::custom(TAG_TARGET, [target.to_hex()])); + + build_rumor_ms(KIND_EDIT, author, content, tags, at_ms) +} + +pub fn build_delete( + author: PublicKey, + channel: &ChannelId, + epoch: Epoch, + target: EventId, + target_kind: Option, + at_ms: u64, +) -> UnsignedEvent { + let mut tags = channel_binding_tags(channel, epoch); + tags.push(Tag::custom(TAG_TARGET, [target.to_hex()])); + + if let Some(target_kind) = target_kind { + tags.push(Tag::custom(TAG_TARGET_KIND, [target_kind.to_string()])); + } + + build_rumor_ms(KIND_DELETE, author, "", tags, at_ms) +} + +pub fn build_typing( + author: PublicKey, + channel: &ChannelId, + epoch: Epoch, + at_ms: u64, +) -> UnsignedEvent { + build_rumor_ms( + KIND_TYPING, + author, + "", + channel_binding_tags(channel, epoch), + at_ms, + ) +} + +/// Seals a chat rumor and wraps it at the channel's address. `ephemeral` picks +/// the 21059 wrap, which relays must not store. +pub fn seal_rumor( + rumor: &UnsignedEvent, + group: &GroupKey, + author: &Keys, + ephemeral: bool, +) -> Result<(Event, Keys), ChatError> { + let kind = rumor.kind.as_u16(); + + if !is_chat_kind(kind) { + return Err(ChatError::UnknownKind(kind)); + } + + let seal = build_seal(rumor, SealForm::Encrypted, group, author)?; + let wrap_kind = if ephemeral { + KIND_WRAP_EPHEMERAL + } else { + KIND_WRAP + }; + + // CORD-08 §2: a NIP-40 expiration rides the wrap as well, so relays drop the + // stored event on schedule; the inner copy is what drives a local purge. + let expiration: Vec = rumor + .tags + .iter() + .filter(|tag| tag.as_slice().first().map(String::as_str) == Some(TAG_EXPIRATION)) + .cloned() + .collect(); + + Ok(wrap_seal( + &seal, + group, + wrap_kind, + rumor.created_at, + &expiration, + )?) +} + +/// Opens a wrap against the plane whose key is tried. The channel and epoch the +/// rumor claims must both be the ones that opened it, so a keyholder of two +/// planes cannot re-seal a rumor elsewhere or replay it across an epoch. +pub fn open( + wrap: &Event, + group: &GroupKey, + channel: &ChannelId, + epoch: Epoch, +) -> Result<(OpenedStream, ChatRumor), ChatError> { + let opened = open_wrap(wrap, group)?; + + if opened.seal_form != SealForm::Encrypted { + return Err(ChatError::NotEncryptedSealed); + } + + check_channel_binding(&opened.rumor, channel, epoch)?; + + let chat = typed(&opened.rumor, channel, epoch)?; + + Ok((opened, chat)) +} + +/// Every epoch's group key for one channel. `secret` is whatever feeds the +/// channel at that epoch: the `community_root` for a public one, its own key +/// for a private one. +pub fn plane_keys( + held: &[(Epoch, [u8; 32])], + channel: &ChannelId, +) -> Result> { + held.iter() + .map(|(epoch, secret)| Ok((*epoch, channel_group_key(secret, channel, *epoch)?))) + .collect() +} + +/// Folds the chat plane into timeline rows, newest first. A delete is honored +/// only from the message's own author, and a deletion is terminal: an edit or a +/// reaction arriving later never revives it. +pub fn fold(rumors: &[ChatRumor]) -> Vec { + let mut order: Vec = (0..rumors.len()).collect(); + order.sort_by_key(|&index| (rumors[index].at_ms, rumors[index].id)); + + let mut messages: Vec = Vec::new(); + let mut slot: BTreeMap = BTreeMap::new(); + + for index in order { + let rumor = &rumors[index]; + + let ChatAction::Message { + reply_to, + thread_root, + } = &rumor.action + else { + continue; + }; + + slot.insert(rumor.id, messages.len()); + messages.push(ChatMessage { + id: rumor.id, + author: rumor.author, + channel: rumor.channel, + epoch: rumor.epoch, + kind: rumor.kind, + content: rumor.content.clone(), + reply_to: reply_to.map(|reply| reply.id), + thread_root: thread_root.map(|reply| reply.id), + at_ms: rumor.at_ms, + expiration: rumor.expiration, + edited_at: None, + deleted: false, + reactions: BTreeMap::new(), + }); + } + + // Mutations replay so the last one applied is the winner: the highest + // `at_ms` and, between equal ones, the lower inner rumor id. + let mut mutations: Vec = (0..rumors.len()).collect(); + mutations.sort_by_key(|&index| (rumors[index].at_ms, Reverse(rumors[index].id))); + + for index in mutations { + let rumor = &rumors[index]; + + match &rumor.action { + ChatAction::Edit { target, content } => { + let Some(&slot) = slot.get(target) else { + continue; + }; + let message = &mut messages[slot]; + + if message.deleted || message.author != rumor.author { + continue; + } + + message.content = content.clone(); + message.edited_at = Some(rumor.at_ms); + } + ChatAction::Delete { target, .. } => { + let Some(&slot) = slot.get(target) else { + continue; + }; + + if messages[slot].author == rumor.author { + messages[slot].deleted = true; + } + } + ChatAction::Reaction { target, emoji } => { + let Some(&slot) = slot.get(target) else { + continue; + }; + + messages[slot].reactions.insert(rumor.author, emoji.clone()); + } + ChatAction::Message { .. } | ChatAction::Typing | ChatAction::Opaque => {} + } + } + + messages.sort_by_key(|message| (Reverse(message.at_ms), message.id)); + + messages +} + +fn is_chat_kind(kind: u16) -> bool { + matches!( + kind, + KIND_MESSAGE + | KIND_COMMENT + | KIND_REACTION + | KIND_DELETE + | KIND_EDIT + | KIND_FILE + | KIND_WEBXDC + | KIND_TYPING + ) +} + +fn typed(rumor: &UnsignedEvent, channel: &ChannelId, epoch: Epoch) -> Result { + Ok(ChatRumor { + id: rumor.id.unwrap_or_else(|| rumor.compute_id()), + author: rumor.pubkey, + kind: rumor.kind, + channel: *channel, + epoch, + at_ms: resolve_ms_strict(rumor)?, + content: rumor.content.clone(), + expiration: expiration_of(rumor)?, + action: action_of(rumor)?, + }) +} + +fn action_of(rumor: &UnsignedEvent) -> Result { + let kind = rumor.kind.as_u16(); + + match kind { + KIND_MESSAGE | KIND_FILE => Ok(ChatAction::Message { + reply_to: optional_reply(rumor, TAG_QUOTE)?, + thread_root: None, + }), + KIND_COMMENT => Ok(ChatAction::Message { + reply_to: optional_reply(rumor, TAG_TARGET)?, + thread_root: optional_reply(rumor, TAG_ROOT)?, + }), + KIND_REACTION => Ok(ChatAction::Reaction { + target: required_id(rumor, TAG_TARGET)?, + emoji: rumor.content.clone(), + }), + KIND_EDIT => Ok(ChatAction::Edit { + target: required_id(rumor, TAG_TARGET)?, + content: rumor.content.clone(), + }), + KIND_DELETE => Ok(ChatAction::Delete { + target: required_id(rumor, TAG_TARGET)?, + target_kind: optional_kind(rumor, TAG_TARGET_KIND)?, + }), + KIND_TYPING => Ok(ChatAction::Typing), + KIND_WEBXDC => Ok(ChatAction::Opaque), + other => Err(ChatError::UnknownKind(other)), + } +} + +fn optional_reply( + rumor: &UnsignedEvent, + name: &'static str, +) -> Result, ChatError> { + let Some(fields) = tag(rumor, name)? else { + return Ok(None); + }; + + // NIP-C7 `q` and NIP-22 `E`/`e` put a relay hint at index 2 and the + // referenced author at index 3, which is a SHOULD, so absent reads as unknown. + let author = match fields.get(3).map(String::as_str) { + Some(hex) if !hex.is_empty() => Some(pubkey(hex, name)?), + _ => None, + }; + + Ok(Some(ReplyRef { + id: hex_id(fields, name)?, + author, + })) +} + +fn required_id(rumor: &UnsignedEvent, name: &'static str) -> Result { + let fields = tag(rumor, name)?.ok_or(ChatError::MissingTag(name))?; + hex_id(fields, name) +} + +fn optional_kind(rumor: &UnsignedEvent, name: &'static str) -> Result, ChatError> { + let Some(fields) = tag(rumor, name)? else { + return Ok(None); + }; + + let raw = value(fields, name)?; + let kind = canonical_decimal(raw).ok_or(ChatError::BadTag(name))?; + + u16::try_from(kind) + .map(Some) + .map_err(|_| ChatError::BadTag(name)) +} + +fn expiration_of(rumor: &UnsignedEvent) -> Result, ChatError> { + let Some(fields) = tag(rumor, TAG_EXPIRATION)? else { + return Ok(None); + }; + + let seconds = canonical_decimal(value(fields, TAG_EXPIRATION)?) + .ok_or(ChatError::BadTag(TAG_EXPIRATION))?; + + Ok(Some(Timestamp::from_secs(seconds))) +} + +fn reply_tag(name: &str, reply: &ReplyRef) -> Tag { + Tag::custom( + name, + [ + reply.id.to_hex(), + String::new(), + reply + .author + .map(|author| author.to_hex()) + .unwrap_or_default(), + ], + ) +} + +fn tag<'a>( + rumor: &'a UnsignedEvent, + name: &'static str, +) -> Result, ChatError> { + let mut found: Option<&[String]> = None; + + for candidate in rumor.tags.iter() { + let fields = candidate.as_slice(); + + if fields.first().map(String::as_str) != Some(name) { + continue; + } + + if found.is_some() { + return Err(ChatError::DuplicateTag(name)); + } + + found = Some(fields); + } + + Ok(found) +} + +fn value<'a>(fields: &'a [String], name: &'static str) -> Result<&'a str, ChatError> { + fields + .get(1) + .map(String::as_str) + .ok_or(ChatError::BadTag(name)) +} + +fn hex_id(fields: &[String], name: &'static str) -> Result { + let bytes = decode_hex_32(value(fields, name)?).map_err(|_| ChatError::BadTag(name))?; + + EventId::from_slice(&bytes).map_err(|_| ChatError::BadTag(name)) +} + +fn pubkey(hex: &str, name: &'static str) -> Result { + let bytes = decode_hex_32(hex).map_err(|_| ChatError::BadTag(name))?; + + PublicKey::from_slice(&bytes).map_err(|_| ChatError::BadTag(name)) +} + +#[cfg(test)] +mod tests { + use super::*; + + const SECRET: [u8; 32] = [0x2du8; 32]; + const AT: u64 = 1_700_000_000_417; + + fn channel() -> ChannelId { + ChannelId::from_bytes([0x9cu8; 32]) + } + + fn group() -> GroupKey { + channel_group_key(&SECRET, &channel(), Epoch(0)).expect("derives") + } + + fn sealed(rumor: &UnsignedEvent, group: &GroupKey, author: &Keys) -> Event { + seal_rumor(rumor, group, author, false).expect("seals").0 + } + + fn read(rumor: &UnsignedEvent, group: &GroupKey, author: &Keys, epoch: Epoch) -> ChatRumor { + open(&sealed(rumor, group, author), group, &channel(), epoch) + .expect("opens") + .1 + } + + fn target(id: EventId, author: &Keys) -> Target { + Target { + reply: ReplyRef { + id, + author: Some(author.public_key()), + }, + kind: KIND_MESSAGE, + } + } + + #[test] + fn a_second_holder_folds_edits_reactions_and_a_self_delete() { + let alice = Keys::generate(); + let carol = Keys::generate(); + let group = group(); + + let message = build_message(alice.public_key(), &channel(), Epoch(0), "hello", None, AT); + let id = message.compute_id(); + + let rumors = vec![ + read(&message, &group, &alice, Epoch(0)), + read( + &build_reaction( + carol.public_key(), + &channel(), + Epoch(0), + &target(id, &alice), + "🔥", + AT + 1_000, + ), + &group, + &carol, + Epoch(0), + ), + read( + &build_edit( + alice.public_key(), + &channel(), + Epoch(0), + id, + "hello (fixed)", + AT + 2_000, + ), + &group, + &alice, + Epoch(0), + ), + read( + &build_delete( + alice.public_key(), + &channel(), + Epoch(0), + id, + Some(KIND_MESSAGE), + AT + 3_000, + ), + &group, + &alice, + Epoch(0), + ), + ]; + + let folded = fold(&rumors); + + assert_eq!(folded.len(), 1); + assert_eq!(folded[0].id, id); + assert_eq!(folded[0].content, "hello (fixed)"); + assert_eq!(folded[0].edited_at, Some(AT + 2_000)); + assert_eq!( + folded[0].reactions.get(&carol.public_key()), + Some(&"🔥".to_owned()) + ); + assert!(folded[0].deleted); + } + + #[test] + fn an_edit_or_delete_from_another_author_is_ignored() { + let alice = Keys::generate(); + let bob = Keys::generate(); + let group = group(); + + let message = build_message(alice.public_key(), &channel(), Epoch(0), "hello", None, AT); + let id = message.compute_id(); + + let rumors = vec![ + read(&message, &group, &alice, Epoch(0)), + read( + &build_edit( + bob.public_key(), + &channel(), + Epoch(0), + id, + "mine now", + AT + 1_000, + ), + &group, + &bob, + Epoch(0), + ), + read( + &build_delete( + bob.public_key(), + &channel(), + Epoch(0), + id, + Some(KIND_MESSAGE), + AT + 2_000, + ), + &group, + &bob, + Epoch(0), + ), + ]; + + let folded = fold(&rumors); + + assert_eq!(folded.len(), 1); + assert_eq!(folded[0].content, "hello"); + assert_eq!(folded[0].edited_at, None); + assert!(!folded[0].deleted); + } + + #[test] + fn a_comment_carries_its_root_and_its_parent() { + let alice = Keys::generate(); + let bob = Keys::generate(); + let group = group(); + + let root = build_message(alice.public_key(), &channel(), Epoch(0), "root", None, AT); + let root_id = root.compute_id(); + let parent = build_message( + bob.public_key(), + &channel(), + Epoch(0), + "parent", + None, + AT + 1_000, + ); + let parent_id = parent.compute_id(); + + let comment = build_comment( + alice.public_key(), + &channel(), + Epoch(0), + "deep", + &target(parent_id, &bob), + Some(&target(root_id, &alice)), + AT + 2_000, + ); + + assert!(comment.tags.iter().any(|tag| tag.as_slice() == ["K", "9"])); + assert!( + comment + .tags + .iter() + .any(|tag| { tag.as_slice()[0] == "E" && tag.as_slice()[1] == root_id.to_hex() }) + ); + assert!( + comment + .tags + .iter() + .any(|tag| { tag.as_slice()[0] == "e" && tag.as_slice()[1] == parent_id.to_hex() }) + ); + + let rumor = read(&comment, &group, &alice, Epoch(0)); + let ChatAction::Message { + reply_to, + thread_root, + } = &rumor.action + else { + panic!("a comment is a message row") + }; + + assert_eq!(reply_to.map(|reply| reply.id), Some(parent_id)); + assert_eq!(thread_root.map(|root| root.id), Some(root_id)); + } + + #[test] + fn a_rumor_bound_to_another_channel_or_epoch_is_rejected() { + let alice = Keys::generate(); + let group = group(); + + let plain = build_message(alice.public_key(), &channel(), Epoch(0), "hello", None, AT); + assert!( + open( + &sealed(&plain, &group, &alice), + &group, + &channel(), + Epoch(0) + ) + .is_ok() + ); + + // The keyholder re-addresses their own rumor: the binding is judged + // against the plane whose key opened the wrap, never the rumor's claim. + let elsewhere = ChannelId::from_bytes([0xeeu8; 32]); + assert!(matches!( + open( + &sealed(&plain, &group, &alice), + &group, + &elsewhere, + Epoch(0) + ), + Err(ChatError::Stream(StreamError::ChannelMismatch)) + )); + + let stale = build_message(alice.public_key(), &channel(), Epoch(1), "stale", None, AT); + assert!(matches!( + open( + &sealed(&stale, &group, &alice), + &group, + &channel(), + Epoch(0) + ), + Err(ChatError::Stream(StreamError::EpochMismatch)) + )); + + // Chat is encrypted-seal only (CORD-02 §5), and a retired kind is not a + // chat rumor however well-formed it looks. + let seal = build_seal(&plain, SealForm::Plaintext, &group, &alice).expect("seals"); + let (wrap, _) = wrap_seal( + &seal, + &group, + KIND_WRAP, + Timestamp::from_secs(AT / 1000), + &[], + ) + .expect("wraps"); + assert!(matches!( + open(&wrap, &group, &channel(), Epoch(0)), + Err(ChatError::NotEncryptedSealed) + )); + + let ghost = build_rumor_ms( + 3300, + alice.public_key(), + "v1 ghost", + channel_binding_tags(&channel(), Epoch(0)), + AT, + ); + assert!(matches!( + seal_rumor(&ghost, &group, &alice, false), + Err(ChatError::UnknownKind(3300)) + )); + + let mut tags = channel_binding_tags(&channel(), Epoch(0)); + tags.push(Tag::custom(TAG_TARGET, ["ab".repeat(32)])); + tags.push(Tag::custom(TAG_TARGET, ["cd".repeat(32)])); + let ambiguous = build_rumor_ms(KIND_DELETE, alice.public_key(), "", tags, AT); + assert!(matches!( + open( + &sealed(&ambiguous, &group, &alice), + &group, + &channel(), + Epoch(0) + ), + Err(ChatError::DuplicateTag(TAG_TARGET)) + )); + } +} diff --git a/crates/concord/src/edition.rs b/crates/concord/src/edition.rs index 16295ba9..83236b0d 100644 --- a/crates/concord/src/edition.rs +++ b/crates/concord/src/edition.rs @@ -399,7 +399,7 @@ impl From<&ParsedEdition> for EntityHead { /// Every entity's committed head, keyed by coordinate. pub type Floors = BTreeMap<[u8; 32], EntityHead>; -fn canonical_decimal(raw: &str) -> Option { +pub(crate) fn canonical_decimal(raw: &str) -> Option { if raw.is_empty() || !raw.bytes().all(|byte| byte.is_ascii_digit()) { return None; } diff --git a/crates/concord/src/lib.rs b/crates/concord/src/lib.rs index 0e6f96ba..4f38a0dd 100644 --- a/crates/concord/src/lib.rs +++ b/crates/concord/src/lib.rs @@ -1,3 +1,4 @@ +pub mod chat; pub mod control; pub mod derive; pub mod edition; diff --git a/crates/concord/src/store.rs b/crates/concord/src/store.rs index 0348f839..8d31818e 100644 --- a/crates/concord/src/store.rs +++ b/crates/concord/src/store.rs @@ -1,20 +1,23 @@ -use std::collections::BTreeMap; +use std::cmp::Reverse; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::LazyLock; use anyhow::{Result, anyhow}; use nostr_sdk::prelude::*; use serde::{Deserialize, Serialize}; +use crate::chat::{self, ChatRumor, plane_keys}; use crate::control::{ ChannelMetadata, CommunityGenesis, CommunityMetadata, ControlFold, ROOT_EPOCH, }; use crate::derive::control_signer_group_key; use crate::edition::{EntityHead, Floors, ParsedEdition, vsk}; -use crate::stream::OpenedStream; -use crate::{ChannelId, CommunityId, Epoch}; +use crate::stream::{KIND_WRAP_EPHEMERAL, OpenedStream}; +use crate::{ChannelId, CommunityId, Epoch, GroupKey}; static LOCAL_KEYS: LazyLock = LazyLock::new(Keys::generate); +const MAX_PAGES: usize = 8; const CHANNEL_TAG: SingleLetterTag = SingleLetterTag::LOWERCASE_C; const MARK_TAG: SingleLetterTag = SingleLetterTag::LOWERCASE_T; const MARK_VALUE: &str = "concord"; @@ -265,18 +268,179 @@ where } } +pub async fn backfill( + client: &Client, + database: &dyn NostrDatabase, + channel: &ChannelId, + held: &[(Epoch, [u8; 32])], + until: Option, + limit: usize, +) -> Result> { + let planes = plane_keys(held, channel)?; + let authors: Vec = planes.iter().map(|(_, group)| group.pk()).collect(); + + let mut cursor = until; + let mut seen: BTreeSet = BTreeSet::new(); + let mut found: Vec = Vec::new(); + + for _ in 0..MAX_PAGES { + let page = fetch_page(client, &authors, cursor, limit).await?; + + if page.is_empty() { + break; + } + + let (fresh, next) = advance(&page, &planes, channel, cursor, limit, &mut seen); + + for (opened, rumor) in fresh { + cache_rumor(database, channel, &opened).await?; + found.push(rumor); + } + + match next { + Some(next) => cursor = Some(next), + None => break, + } + } + + found.sort_by_key(|rumor| (Reverse(rumor.at_ms), rumor.id)); + found.truncate(limit); + + Ok(found) +} + +fn advance( + page: &BTreeSet, + planes: &[(Epoch, GroupKey)], + channel: &ChannelId, + cursor: Option, + limit: usize, + seen: &mut BTreeSet, +) -> (Vec<(OpenedStream, ChatRumor)>, Option) { + let mut fresh = Vec::new(); + + for wrap in page { + let Some((epoch, group)) = planes.iter().find(|(_, group)| group.pk() == wrap.pubkey) + else { + continue; + }; + + let Ok((opened, rumor)) = chat::open(wrap, group, channel, *epoch) else { + continue; + }; + + if seen.insert(rumor.id) { + fresh.push((opened, rumor)); + } + } + + if fresh.is_empty() || page.len() < limit { + return (fresh, None); + } + + let oldest = page.iter().map(|event| event.created_at).min(); + + match oldest { + Some(oldest) if cursor != Some(oldest) => (fresh, Some(oldest)), + _ => (fresh, None), + } +} + +async fn fetch_page( + client: &Client, + authors: &[PublicKey], + until: Option, + limit: usize, +) -> Result> { + let mut filter = Filter::new() + .kinds([Kind::GiftWrap, Kind::Custom(KIND_WRAP_EPHEMERAL)]) + .authors(authors.iter().copied()) + .limit(limit); + + if let Some(until) = until { + filter = filter.until(until); + } + + Ok(client.fetch_events(filter).await?) +} + #[cfg(test)] mod tests { use nostr_memory::MemoryDatabase; use super::*; use crate::Epoch; + use crate::chat::{build_message, seal_rumor}; use crate::derive::channel_group_key; use crate::stream::{ KIND_WRAP, SealForm, build_rumor_ms, build_seal, channel_binding_tags, open_wrap, wrap_seal, }; const SECRET: [u8; 32] = [0x07u8; 32]; + const NEXT_SECRET: [u8; 32] = [0x11u8; 32]; + + /// What a relay does with an inclusive `until` and a `limit`. + fn serve_page( + relay: &BTreeSet, + cursor: Option, + limit: usize, + ) -> BTreeSet { + let mut events: Vec = relay + .iter() + .filter(|event| cursor.is_none_or(|cursor| event.created_at <= cursor)) + .cloned() + .collect(); + + events.sort_by_key(|event| Reverse(event.created_at)); + events.truncate(limit); + events.into_iter().collect() + } + + #[test] + fn history_pages_back_across_a_rekey() { + let channel = ChannelId::from_bytes([0x9cu8; 32]); + let author = Keys::generate(); + let held = [(Epoch(0), SECRET), (Epoch(1), NEXT_SECRET)]; + let planes = plane_keys(&held, &channel).expect("derives"); + + // Three messages a second apart: a page boundary falls between each. + let base = 1_700_000_000_000; + let mut relay: BTreeSet = BTreeSet::new(); + + for (content, secret, epoch, at_ms) in [ + ("before the rekey", &SECRET, Epoch(0), base), + ("still before", &SECRET, Epoch(0), base + 1_000), + ("after the rekey", &NEXT_SECRET, Epoch(1), base + 2_000), + ] { + let group = channel_group_key(secret, &channel, epoch).expect("derives"); + let rumor = build_message(author.public_key(), &channel, epoch, content, None, at_ms); + relay.insert(seal_rumor(&rumor, &group, &author, false).expect("seals").0); + } + + let mut seen = BTreeSet::new(); + let mut found = Vec::new(); + let mut cursor = None; + + for _ in 0..3 { + let page = serve_page(&relay, cursor, 2); + let (fresh, next) = advance(&page, &planes, &channel, cursor, 2, &mut seen); + + found.extend(fresh.into_iter().map(|(_, rumor)| rumor)); + + match next { + Some(next) => cursor = Some(next), + None => break, + } + } + + found.sort_by_key(|rumor| (Reverse(rumor.at_ms), rumor.id)); + + let contents: Vec<&str> = found.iter().map(|rumor| rumor.content.as_str()).collect(); + assert_eq!( + contents, + ["after the rekey", "still before", "before the rekey"] + ); + } #[test] fn rumors_read_back_after_a_restart() {