use nostr_sdk::prelude::*;
use super::derive::channel_pseudonym;
use super::envelope::{open_message_multi, seal_message_with_ephemeral, seal_with_signed_inner, OpenedMessage};
#[cfg(test)]
use super::envelope::open_message;
use super::transport::{Query, Transport};
use super::{Channel, Community};
use crate::stored_event::event_kind;
pub async fn publish_message<T: Transport + ?Sized>(
transport: &T,
community: &Community,
channel: &Channel,
author: &Keys,
content: &str,
ms: u64,
) -> Result<(Event, Keys), String> {
let ephemeral = Keys::generate();
let outer = seal_message_with_ephemeral(
&ephemeral, author, &channel.key, &channel.id, channel.epoch, content, ms,
)
.map_err(|e| e.to_string())?;
transport.publish(&outer, &community.relays).await?;
Ok((outer, ephemeral))
}
pub async fn publish_signed_message<T: Transport + ?Sized>(
transport: &T,
community: &Community,
channel: &Channel,
inner: &Event,
durable: bool,
) -> Result<(Event, Keys), String> {
let ephemeral = Keys::generate();
let outer = seal_with_signed_inner(&ephemeral, inner, &channel.key, &channel.id, channel.epoch)
.map_err(|e| e.to_string())?;
if durable {
transport.publish_durable(&outer, &community.relays).await?;
} else {
transport.publish(&outer, &community.relays).await?;
}
Ok((outer, ephemeral))
}
pub async fn delete_own_message<T: Transport + ?Sized>(
transport: &T,
relays: &[String],
ephemeral: &Keys,
outer_event_id: EventId,
) -> Result<(), String> {
let deletion = EventBuilder::delete(EventDeletionRequest::new().ids([outer_event_id]))
.sign_with_keys(ephemeral)
.map_err(|e| e.to_string())?;
transport.publish_durable(&deletion, relays).await
}
fn channel_read_pseudonyms(channel: &Channel) -> Vec<String> {
channel
.read_epoch_keys()
.iter()
.map(|(epoch, key)| channel_pseudonym(key, &channel.id, *epoch).to_hex())
.collect()
}
pub async fn fetch_channel_messages<T: Transport + ?Sized>(
transport: &T,
community: &Community,
channel: &Channel,
) -> Result<Vec<OpenedMessage>, String> {
let query = Query {
kinds: vec![event_kind::COMMUNITY_MESSAGE],
z_tags: channel_read_pseudonyms(channel),
since: None,
..Default::default()
};
let events = transport.fetch(&query, &community.relays).await?;
let epoch_keys = channel.read_epoch_keys();
let mut opened: Vec<OpenedMessage> = Vec::new();
let mut dropped = 0usize;
for ev in &events {
match open_message_multi(ev, &channel.id, &epoch_keys) {
Ok(msg) => opened.push(msg),
Err(e) => {
dropped += 1;
crate::log_debug!("[community] dropped event {}: {}", ev.id.to_hex(), e);
}
}
}
if dropped > 0 {
crate::log_debug!(
"[community] channel {} fetch: {} opened, {} dropped",
channel.id.to_hex(),
opened.len(),
dropped
);
}
{
let mut seen = std::collections::HashSet::new();
opened.retain(|m| seen.insert(m.message_id));
}
opened.sort_by(|a, b| {
a.ms.unwrap_or(0)
.cmp(&b.ms.unwrap_or(0))
.then_with(|| a.message_id.to_hex().cmp(&b.message_id.to_hex()))
});
Ok(opened)
}
pub async fn fetch_channel_events<T: Transport + ?Sized>(
transport: &T,
community: &Community,
channel: &Channel,
) -> Result<Vec<Event>, String> {
let query = Query {
kinds: vec![
event_kind::COMMUNITY_MESSAGE,
event_kind::COMMUNITY_REACTION,
event_kind::COMMUNITY_EDIT,
event_kind::COMMUNITY_DELETE,
event_kind::COMMUNITY_PRESENCE,
event_kind::COMMUNITY_KICK,
event_kind::COMMUNITY_WEBXDC,
],
z_tags: channel_read_pseudonyms(channel),
since: None,
..Default::default()
};
transport.fetch(&query, &community.relays).await
}
pub async fn fetch_channel_page<T: Transport + ?Sized>(
transport: &T,
community: &Community,
channel: &Channel,
until: Option<u64>,
since: Option<u64>,
limit: usize,
) -> Result<Vec<Event>, String> {
let query = Query {
kinds: vec![
event_kind::COMMUNITY_MESSAGE,
event_kind::COMMUNITY_REACTION,
event_kind::COMMUNITY_EDIT,
event_kind::COMMUNITY_DELETE,
event_kind::COMMUNITY_PRESENCE,
event_kind::COMMUNITY_KICK,
event_kind::COMMUNITY_WEBXDC,
],
z_tags: channel_read_pseudonyms(channel),
until,
since,
limit: Some(limit),
..Default::default()
};
transport.fetch(&query, &community.relays).await
}
#[cfg(test)]
mod tests {
use super::*;
use crate::community::transport::memory::MemoryRelay;
use crate::community::{Channel, ChannelKey, Epoch};
fn community() -> Community {
Community::create("HQ", "general", vec!["r1".into(), "r2".into(), "r3".into()])
}
fn member_view(of: &Community) -> Community {
Community {
id: of.id,
server_root_key: of.server_root_key.clone(),
server_root_epoch: of.server_root_epoch,
name: of.name.clone(),
description: of.description.clone(),
icon: of.icon.clone(),
banner: of.banner.clone(),
relays: of.relays.clone(),
channels: of.channels.clone(),
owner_attestation: of.owner_attestation.clone(),
dissolved: of.dissolved,
}
}
#[tokio::test]
async fn two_clients_exchange_via_relay() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
publish_message(&relay, &community, &channel, &alice, "gm from alice", 100)
.await
.unwrap();
let bob_view = member_view(&community);
let msgs = fetch_channel_messages(&relay, &bob_view, &bob_view.channels[0])
.await
.unwrap();
assert_eq!(msgs.len(), 1);
assert_eq!(msgs[0].content, "gm from alice");
assert_eq!(msgs[0].author, alice.public_key());
}
fn epoch_view(base: &Channel, key: ChannelKey, epoch: u64) -> Channel {
Channel {
id: base.id, key, epoch: Epoch(epoch), name: base.name.clone(),
banned: Vec::new(), protected: Vec::new(), roster: Default::default(), epoch_keys: Vec::new(),
dissolved: false,
}
}
#[tokio::test]
async fn fetch_spans_held_epochs_after_rekeys() {
let relay = MemoryRelay::new();
let community = community();
let base = community.channels[0].clone();
let alice = Keys::generate();
let k0 = base.key.clone();
let k1 = ChannelKey([0x11u8; 32]);
let k2 = ChannelKey([0x22u8; 32]);
publish_message(&relay, &community, &epoch_view(&base, k0.clone(), 0), &alice, "epoch0", 100).await.unwrap();
publish_message(&relay, &community, &epoch_view(&base, k1.clone(), 1), &alice, "epoch1", 200).await.unwrap();
publish_message(&relay, &community, &epoch_view(&base, k2.clone(), 2), &alice, "epoch2", 300).await.unwrap();
let mut reader = member_view(&community);
reader.channels[0] = Channel {
id: base.id, key: k2.clone(), epoch: Epoch(2), name: base.name.clone(),
banned: Vec::new(), protected: Vec::new(), roster: Default::default(),
epoch_keys: vec![(Epoch(0), k0), (Epoch(1), k1), (Epoch(2), k2.clone())],
dissolved: false,
};
let msgs = fetch_channel_messages(&relay, &reader, &reader.channels[0]).await.unwrap();
let contents: Vec<&str> = msgs.iter().map(|m| m.content.as_str()).collect();
assert_eq!(contents, vec!["epoch0", "epoch1", "epoch2"], "every held epoch's messages returned, none stranded");
let mut head_only = member_view(&community);
head_only.channels[0] = epoch_view(&base, k2, 2);
let only = fetch_channel_messages(&relay, &head_only, &head_only.channels[0]).await.unwrap();
assert_eq!(only.iter().map(|m| m.content.as_str()).collect::<Vec<_>>(), vec!["epoch2"],
"head-only reader sees only the current epoch (single-epoch fallback)");
}
#[tokio::test]
async fn page_spans_the_epoch_boundary() {
let relay = MemoryRelay::new();
let community = community();
let base = community.channels[0].clone();
let alice = Keys::generate();
let k0 = base.key.clone();
let k1 = ChannelKey([0x33u8; 32]);
publish_message(&relay, &community, &epoch_view(&base, k0.clone(), 0), &alice, "old-a", 100).await.unwrap();
publish_message(&relay, &community, &epoch_view(&base, k0.clone(), 0), &alice, "old-b", 200).await.unwrap();
publish_message(&relay, &community, &epoch_view(&base, k1.clone(), 1), &alice, "new-c", 300).await.unwrap();
let mut reader = member_view(&community);
reader.channels[0] = Channel {
id: base.id, key: k1.clone(), epoch: Epoch(1), name: base.name.clone(),
banned: Vec::new(), protected: Vec::new(), roster: Default::default(),
epoch_keys: vec![(Epoch(0), k0), (Epoch(1), k1)],
dissolved: false,
};
let page = fetch_channel_page(&relay, &reader, &reader.channels[0], None, None, 20).await.unwrap();
let opened: Vec<String> = page.iter()
.filter_map(|e| open_message_multi(e, &reader.channels[0].id, &reader.channels[0].read_epoch_keys()).ok())
.map(|m| m.content)
.collect();
assert!(opened.contains(&"new-c".to_string()), "head-epoch message in the page");
assert!(opened.contains(&"old-a".to_string()) && opened.contains(&"old-b".to_string()),
"older-epoch messages in the SAME page (across the epoch boundary)");
}
#[tokio::test]
async fn three_clients_one_broadcast() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
publish_message(&relay, &community, &channel, &alice, "hello all", 1)
.await
.unwrap();
for reader in [member_view(&community), member_view(&community)] {
let msgs = fetch_channel_messages(&relay, &reader, &reader.channels[0])
.await
.unwrap();
assert_eq!(msgs.len(), 1);
assert_eq!(msgs[0].content, "hello all");
}
}
#[tokio::test]
async fn non_member_with_wrong_key_cannot_read() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
publish_message(&relay, &community, &channel, &alice, "secret", 1)
.await
.unwrap();
let mut outsider = member_view(&community);
outsider.channels[0].key = ChannelKey([0xeeu8; 32]);
let msgs = fetch_channel_messages(&relay, &outsider, &outsider.channels[0])
.await
.unwrap();
assert!(msgs.is_empty(), "wrong key derives a different pseudonym → no hits");
let raw = relay
.fetch(
&Query { kinds: vec![event_kind::COMMUNITY_MESSAGE], ..Default::default() },
&community.relays,
)
.await
.unwrap();
assert_eq!(raw.len(), 1);
assert!(open_message(&raw[0], &outsider.channels[0].key, &channel.id, channel.epoch).is_err());
}
#[tokio::test]
async fn messages_return_in_ms_order() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
publish_message(&relay, &community, &channel, &alice, "third", 300).await.unwrap();
publish_message(&relay, &community, &channel, &alice, "first", 100).await.unwrap();
publish_message(&relay, &community, &channel, &alice, "second", 200).await.unwrap();
let reader = member_view(&community);
let msgs = fetch_channel_messages(&relay, &reader, &reader.channels[0]).await.unwrap();
let contents: Vec<&str> = msgs.iter().map(|m| m.content.as_str()).collect();
assert_eq!(contents, vec!["first", "second", "third"]);
}
#[tokio::test]
async fn other_channel_traffic_is_not_returned() {
let relay = MemoryRelay::new();
let community = community();
let chan_a = community.channels[0].clone();
let chan_b = Channel {
id: super::super::ChannelId([0x77u8; 32]),
key: ChannelKey([0x88u8; 32]),
epoch: Epoch(0),
name: "other".into(),
banned: Vec::new(),
protected: Vec::new(), roster: Default::default(),
epoch_keys: Vec::new(),
dissolved: false,
};
let alice = Keys::generate();
publish_message(&relay, &community, &chan_a, &alice, "in A", 1).await.unwrap();
publish_message(&relay, &community, &chan_b, &alice, "in B", 1).await.unwrap();
let reader = member_view(&community);
let msgs = fetch_channel_messages(&relay, &reader, &reader.channels[0]).await.unwrap();
assert_eq!(msgs.len(), 1);
assert_eq!(msgs[0].content, "in A");
}
#[tokio::test]
async fn two_distinct_messages_are_not_collapsed() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
publish_message(&relay, &community, &channel, &alice, "one", 1).await.unwrap();
publish_message(&relay, &community, &channel, &alice, "two", 2).await.unwrap();
let reader = member_view(&community);
let msgs = fetch_channel_messages(&relay, &reader, &reader.channels[0]).await.unwrap();
assert_eq!(msgs.len(), 2);
let contents: Vec<&str> = msgs.iter().map(|m| m.content.as_str()).collect();
assert!(contents.contains(&"one") && contents.contains(&"two"));
}
#[tokio::test]
async fn backfill_fetches_and_applies_messages_then_reactions() {
use super::super::inbound::{process_channel_batch, IncomingEvent};
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
let bob = Keys::generate();
let (m1_outer, _) = publish_message(&relay, &community, &channel, &alice, "hello", 1).await.unwrap();
publish_message(&relay, &community, &channel, &alice, "world", 2).await.unwrap();
let m1_inner = open_message(&m1_outer, &channel.key, &channel.id, channel.epoch).unwrap().message_id.to_hex();
let react_inner = super::super::envelope::build_inner_typed(
bob.public_key(), &channel.id, channel.epoch,
event_kind::COMMUNITY_REACTION, "🔥", 3, Some(&m1_inner), &[],
).sign_with_keys(&bob).unwrap();
let react_outer = seal_with_signed_inner(&Keys::generate(), &react_inner, &channel.key, &channel.id, channel.epoch).unwrap();
relay.publish(&react_outer, &community.relays).await.unwrap();
let events = fetch_channel_events(&relay, &community, &channel).await.unwrap();
assert_eq!(events.len(), 3, "two messages + one reaction fetched");
let mut state = crate::state::ChatState::new();
let applied = process_channel_batch(&mut state, &events, &channel, &bob.public_key());
let new_msgs = applied.iter().filter(|e| matches!(e, IncomingEvent::NewMessage(_))).count();
let updates: Vec<&String> = applied.iter().filter_map(|e| match e {
IncomingEvent::Updated { target_id, .. } => Some(target_id),
_ => None,
}).collect();
assert_eq!(new_msgs, 2, "both messages backfilled");
assert_eq!(updates.len(), 1, "reaction applied during backfill");
assert_eq!(updates[0], &m1_inner, "reaction landed on its target message");
}
#[tokio::test]
async fn same_inner_message_in_two_wrappers_collapses() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
let inner = super::super::envelope::build_inner_event(alice.public_key(), &channel.id, channel.epoch, "dup me", 1, None)
.sign_with_keys(&alice)
.unwrap();
let outer_a = seal_with_signed_inner(&Keys::generate(), &inner, &channel.key, &channel.id, channel.epoch).unwrap();
let outer_b = seal_with_signed_inner(&Keys::generate(), &inner, &channel.key, &channel.id, channel.epoch).unwrap();
assert_ne!(outer_a.id, outer_b.id, "distinct outer wrappers");
relay.publish(&outer_a, &community.relays).await.unwrap();
relay.publish(&outer_b, &community.relays).await.unwrap();
let reader = member_view(&community);
let msgs = fetch_channel_messages(&relay, &reader, &reader.channels[0]).await.unwrap();
assert_eq!(msgs.len(), 1, "same inner id collapses across wrappers");
assert_eq!(msgs[0].content, "dup me");
}
#[tokio::test]
async fn bad_event_dropped_good_event_kept_in_same_batch() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
publish_message(&relay, &community, &channel, &alice, "valid", 1).await.unwrap();
let pseudonym =
super::super::derive::channel_pseudonym(&channel.key, &channel.id, channel.epoch);
let garbage = EventBuilder::new(Kind::Custom(event_kind::COMMUNITY_MESSAGE), "not-base64-or-cipher!!")
.tags([
Tag::custom(
TagKind::SingleLetter(SingleLetterTag::lowercase(Alphabet::Z)),
[pseudonym.to_hex()],
),
Tag::custom(TagKind::Custom("v".into()), ["1".to_string()]),
])
.sign_with_keys(&Keys::generate())
.unwrap();
relay.publish(&garbage, &community.relays).await.unwrap();
let reader = member_view(&community);
let msgs = fetch_channel_messages(&relay, &reader, &reader.channels[0]).await.unwrap();
assert_eq!(msgs.len(), 1);
assert_eq!(msgs[0].content, "valid");
}
#[tokio::test]
async fn publish_retains_key_and_owner_can_delete() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
let (outer, ephemeral) =
publish_message(&relay, &community, &channel, &alice, "deletable", 1).await.unwrap();
assert_eq!(ephemeral.public_key(), outer.pubkey);
let before = fetch_channel_messages(&relay, &community, &channel).await.unwrap();
assert_eq!(before.len(), 1);
delete_own_message(&relay, &community.relays, &ephemeral, outer.id).await.unwrap();
let after = fetch_channel_messages(&relay, &community, &channel).await.unwrap();
assert!(after.is_empty(), "owner's deletion should remove the message");
}
#[tokio::test]
async fn deletion_by_a_different_key_is_ignored() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
let (outer, ephemeral) =
publish_message(&relay, &community, &channel, &alice, "mine", 1).await.unwrap();
let attacker = Keys::generate();
delete_own_message(&relay, &community.relays, &attacker, outer.id).await.unwrap();
let after = fetch_channel_messages(&relay, &community, &channel).await.unwrap();
assert_eq!(after.len(), 1, "a foreign key must not delete someone else's message");
delete_own_message(&relay, &community.relays, &ephemeral, outer.id).await.unwrap();
let gone = fetch_channel_messages(&relay, &community, &channel).await.unwrap();
assert!(gone.is_empty(), "the original signer's key must delete it");
}
#[tokio::test]
async fn deletion_must_reach_all_relays_to_take_effect() {
let relay = MemoryRelay::new();
let community = community(); let channel = community.channels[0].clone();
let alice = Keys::generate();
let (outer, ephemeral) =
publish_message(&relay, &community, &channel, &alice, "sticky", 1).await.unwrap();
delete_own_message(&relay, &["r1".to_string()], &ephemeral, outer.id).await.unwrap();
let still = fetch_channel_messages(&relay, &community, &channel).await.unwrap();
assert_eq!(still.len(), 1, "deletion on a subset must not remove it everywhere");
delete_own_message(&relay, &community.relays, &ephemeral, outer.id).await.unwrap();
let gone = fetch_channel_messages(&relay, &community, &channel).await.unwrap();
assert!(gone.is_empty());
}
#[tokio::test]
#[ignore]
async fn live_relay_roundtrip_and_cleanup() {
use super::super::transport::LiveTransport;
use crate::community::derive::channel_pseudonym;
let _ = rustls::crypto::ring::default_provider().install_default();
let relays = vec!["wss://jskitty.com/nostr".to_string()];
let community = Community::create("LiveTest", "general", relays.clone());
let channel = community.channels[0].clone();
let alice = Keys::generate();
let transport = LiveTransport::with_timeout(std::time::Duration::from_secs(12));
let pseudonym = channel_pseudonym(&channel.key, &channel.id, channel.epoch).to_hex();
eprintln!("[live] channel pseudonym (z tag) = {pseudonym}");
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as u64;
let mut published: Vec<(Keys, Event)> = Vec::new();
let mut publish_errs: Vec<String> = Vec::new();
for (i, body) in ["live hello one", "live hello two"].iter().enumerate() {
let ephemeral = Keys::generate();
let outer = seal_message_with_ephemeral(
&ephemeral, &alice, &channel.key, &channel.id, channel.epoch, body, now_ms + i as u64,
)
.expect("seal");
match transport.publish(&outer, &relays).await {
Ok(()) => {
eprintln!("[live] published event {}", outer.id.to_hex());
published.push((ephemeral, outer));
}
Err(e) => publish_errs.push(e),
}
}
let fetched = fetch_channel_messages(&transport, &community, &channel).await;
let mut cleanup_ok = true;
for (ephemeral, outer) in &published {
let del = EventBuilder::delete(EventDeletionRequest::new().ids([outer.id]))
.sign_with_keys(ephemeral)
.expect("build deletion");
if let Err(e) = transport.publish(&del, &relays).await {
cleanup_ok = false;
eprintln!("[live] CLEANUP FAILED for {}: {e} — manual delete may be needed", outer.id.to_hex());
} else {
eprintln!("[live] deleted event {}", outer.id.to_hex());
}
}
assert!(publish_errs.is_empty(), "publish errors: {publish_errs:?}");
let msgs = fetched.expect("fetch failed");
let contents: Vec<&str> = msgs.iter().map(|m| m.content.as_str()).collect();
assert!(contents.contains(&"live hello one"), "got {contents:?}");
assert!(contents.contains(&"live hello two"), "got {contents:?}");
for m in &msgs {
assert_eq!(m.author, alice.public_key(), "author recovered over the wire");
}
assert!(cleanup_ok, "cleanup deletion failed — check relay for leftover events");
eprintln!("[live] OK: {} messages round-tripped and cleaned up", msgs.len());
}
#[tokio::test]
async fn redundancy_a_dropped_relay_still_delivers() {
let relay = MemoryRelay::new();
let community = community();
let channel = community.channels[0].clone();
let alice = Keys::generate();
let outer = seal_message_with_ephemeral(
&Keys::generate(), &alice, &channel.key, &channel.id, channel.epoch, "survives", 1,
)
.unwrap();
relay.inject(&outer, &["r2".to_string()]);
let reader = member_view(&community);
let msgs = fetch_channel_messages(&relay, &reader, &reader.channels[0]).await.unwrap();
assert_eq!(msgs.len(), 1);
assert_eq!(msgs[0].content, "survives");
}
}