reya/coop
1
0
Fork 1
Code Issues Pull Requests 1 Releases 3 Activity

remove community and community_ui crates

Browse Source
This commit is contained in:
reya
2026-09-30 17:39:10 +07:00
parent 8c7c672675
commit eda8eb543d
16 changed files with 0 additions and 7058 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-39
View File
@@ -1276,41 +1276,6 @@ dependencies = [
"regex", "regex",
] ]
[[package]]
name = "community"
version = "1.0.2"
dependencies = [
"anyhow",
"concord",
"flume 0.11.1",
"futures",
"gpui-pre",
"log",
"nostr-memory",
"nostr-sdk",
"serde_json",
"smallvec",
"smol",
"state",
]
[[package]]
name = "community_ui"
version = "1.0.2"
dependencies = [
"anyhow",
"common",
"community",
"gpui-pre",
"nostr-sdk",
"person",
"settings",
"smallvec",
"state",
"theme",
"ui",
]
[[package]] [[package]]
name = "compression-codecs" name = "compression-codecs"
version = "0.4.43" version = "0.4.43"
@@ -1449,7 +1414,6 @@ dependencies = [
"auto_update", "auto_update",
"chat", "chat",
"common", "common",
"community",
"device", "device",
"gpui-pre", "gpui-pre",
"gpui-pre-linux", "gpui-pre-linux",
@@ -1476,7 +1440,6 @@ dependencies = [
"assets", "assets",
"chat", "chat",
"common", "common",
"community",
"console_error_panic_hook", "console_error_panic_hook",
"console_log", "console_log",
"device", "device",
@@ -9518,8 +9481,6 @@ dependencies = [
"chat", "chat",
"chat_ui", "chat_ui",
"common", "common",
"community",
"community_ui",
"device", "device",
"gpui-pre", "gpui-pre",
"instant", "instant",
crates/community/Cargo.toml
-23
View File
@@ -1,23 +0,0 @@
[package]
name = "community"
version.workspace = true
edition.workspace = true
publish.workspace = true
[dependencies]
concord = { path = "../concord" }
state = { path = "../state" }
gpui.workspace = true
nostr-sdk.workspace = true
anyhow.workspace = true
flume.workspace = true
futures.workspace = true
log.workspace = true
serde_json.workspace = true
smallvec.workspace = true
smol.workspace = true
[dev-dependencies]
nostr-memory.workspace = true
crates/community/src/cache.rs
-331
View File
@@ -1,331 +0,0 @@
use std::collections::BTreeMap;
use std::sync::LazyLock;
use anyhow::{Result, anyhow};
use concord::cord01::OpenedStream;
use concord::state::{CommunityState, STATE_PREFIX, state_identifier};
use concord::{ChannelId, CommunityId, cord03};
use nostr_sdk::prelude::*;
static LOCAL_KEYS: LazyLock<Keys> = LazyLock::new(|| {
Keys::new(SecretKey::from_slice(&[0x43; 32]).expect("a fixed 32-byte scalar is a valid key"))
});
const CHANNEL_TAG: SingleLetterTag = SingleLetterTag::LOWERCASE_C;
const MARK_TAG: SingleLetterTag = SingleLetterTag::LOWERCASE_T;
const MARK_VALUE: &str = "concord";
const WRAP_TAG: &str = "e";
const KIND_TAG: SingleLetterTag = SingleLetterTag::LOWERCASE_K;
/// An already-expired rumor is refused at ingest. Returns whether it was kept.
pub async fn cache_rumor(
client: &Client,
channel: &ChannelId,
opened: &OpenedStream,
) -> Result<bool> {
let at = Timestamp::from_secs(opened.at_ms / 1000);
if cord03::expiration_of(&opened.rumor)?
.is_some_and(|expiration| expiration <= Timestamp::now())
{
return Ok(false);
}
let tags = vec![
Tag::identifier(opened.rumor_id),
Tag::custom(KIND_TAG.as_str(), [opened.rumor.kind.to_string()]),
Tag::custom(WRAP_TAG, [opened.wrapper_id.to_string()]),
Tag::custom(MARK_TAG.as_str(), [MARK_VALUE]),
Tag::custom(CHANNEL_TAG.as_str(), [channel.to_hex()]),
Tag::public_key(opened.author),
];
let event = EventBuilder::new(Kind::ApplicationSpecificData, opened.rumor.as_json())
.tags(tags)
.custom_created_at(at)
.finalize_async(&*LOCAL_KEYS)
.await?;
client.database().save_event(&event).await?;
Ok(true)
}
pub async fn purge_expired(client: &Client, channel: &ChannelId, now: Timestamp) -> Result<usize> {
let filter = Filter::new()
.kind(Kind::ApplicationSpecificData)
.custom_tag(MARK_TAG, MARK_VALUE)
.custom_tag(CHANNEL_TAG, channel.to_hex());
let mut expired = Vec::new();
for event in client.database().query(filter).await? {
let Ok(rumor) = UnsignedEvent::from_json(&event.content) else {
continue;
};
let Ok(Some(expiration)) = cord03::expiration_of(&rumor) else {
continue;
};
if expiration <= now {
expired.push(event.id);
}
}
let purged = expired.len();
if purged > 0 {
client.database().delete(Filter::new().ids(expired)).await?;
}
Ok(purged)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Observed {
pub author: PublicKey,
pub at_ms: u64,
}
/// The cached rumors of `channel`, keyed by the wrap they were opened from.
pub async fn wrapper_index(
client: &Client,
channel: &ChannelId,
) -> Result<BTreeMap<EventId, Observed>> {
let filter = Filter::new()
.kind(Kind::ApplicationSpecificData)
.custom_tag(MARK_TAG, MARK_VALUE)
.custom_tag(CHANNEL_TAG, channel.to_hex());
let mut index = BTreeMap::new();
for event in client.database().query(filter).await? {
let (Some(wrapper_id), Some(author)) = (
event.tags.event_ids().next(),
event.tags.public_keys().next(),
) else {
continue;
};
index.insert(
wrapper_id,
Observed {
author,
at_ms: event.created_at.as_secs().saturating_mul(1000),
},
);
}
Ok(index)
}
/// Cached rumors for `channel`, newest first, deduplicated by rumor id.
pub async fn query_rumors(
client: &Client,
channel: &ChannelId,
until: Option<Timestamp>,
limit: usize,
kinds: Option<&[u16]>,
) -> Result<Vec<UnsignedEvent>> {
let mut filter = Filter::new()
.kind(Kind::ApplicationSpecificData)
.custom_tag(MARK_TAG, MARK_VALUE)
.custom_tag(CHANNEL_TAG, channel.to_hex());
if let Some(kinds) = kinds {
filter = filter.custom_tags(KIND_TAG, kinds.iter().map(u16::to_string));
}
if let Some(until) = until {
filter = filter.until(until);
}
let mut newest: BTreeMap<String, Event> = BTreeMap::new();
for event in client.database().query(filter).await? {
let Some(rumor_id) = event.tags.identifier() else {
continue;
};
match newest.get(&rumor_id) {
Some(existing) if existing.created_at >= event.created_at => {}
_ => {
newest.insert(rumor_id, event);
}
}
}
let mut events: Vec<Event> = newest.into_values().collect();
events.sort_by_key(|event| std::cmp::Reverse(event.created_at));
events.truncate(limit);
let mut rumors = Vec::with_capacity(events.len());
for event in events {
let rumor = UnsignedEvent::from_json(event.content)
.map_err(|error| anyhow!("cached rumor is not a valid event: {error}"))?;
rumors.push(rumor);
}
Ok(rumors)
}
pub async fn save_state(client: &Client, state: &CommunityState) -> Result<()> {
let event = EventBuilder::new(Kind::ApplicationSpecificData, serde_json::to_string(state)?)
.tags([Tag::identifier(state.identifier())])
.finalize_async(&*LOCAL_KEYS)
.await?;
client.database().save_event(&event).await?;
Ok(())
}
pub async fn load_state(client: &Client, id: &CommunityId) -> Result<Option<CommunityState>> {
let filter = Filter::new()
.kind(Kind::ApplicationSpecificData)
.identifier(state_identifier(id))
.limit(1);
match client.database().query(filter).await?.into_iter().next() {
Some(event) => Ok(Some(serde_json::from_str(&event.content)?)),
None => Ok(None),
}
}
/// The newest state document per community carried in the local database.
pub async fn load_states(client: &Client) -> Result<Vec<CommunityState>> {
let filter = Filter::new().kind(Kind::ApplicationSpecificData);
let mut newest: BTreeMap<CommunityId, Event> = BTreeMap::new();
for event in client.database().query(filter).await? {
let Some(id) = state_document_of(&event) else {
continue;
};
match newest.get(&id) {
Some(existing) if existing.created_at >= event.created_at => {}
_ => {
newest.insert(id, event);
}
}
}
let mut states = Vec::with_capacity(newest.len());
for event in newest.into_values() {
match serde_json::from_str::<CommunityState>(&event.content) {
Ok(state) => states.push(state),
Err(error) => log::warn!("ignoring malformed community state {}: {error}", event.id),
}
}
Ok(states)
}
fn state_document_of(event: &Event) -> Option<CommunityId> {
let identifier = event.tags.identifier()?;
let hex = identifier.strip_prefix(STATE_PREFIX)?;
hex.parse().ok()
}
#[cfg(test)]
mod tests {
use std::collections::BTreeSet;
use concord::Epoch;
use nostr_memory::MemoryDatabase;
use super::*;
fn client() -> Client {
ClientBuilder::default()
.database(MemoryDatabase::unbounded())
.build()
}
#[test]
fn load_states_reads_one_document_per_community_and_ignores_other_documents() {
smol::block_on(async {
let client = client();
let state = CommunityState {
id: CommunityId::from_bytes([0x42; 32]),
name: Some("Anime and Manga".to_owned()),
owner: Keys::generate().public_key(),
owner_salt: [0x01; 32],
community_root: [0x02; 32],
root_epoch: Epoch(0),
control_root: None,
control_pks: BTreeMap::new(),
channels: Vec::new(),
relays: Vec::new(),
heads: Vec::new(),
banned: BTreeSet::new(),
cursors: BTreeMap::new(),
held_roots: Vec::new(),
channel_cuts: BTreeMap::new(),
refounders: BTreeSet::new(),
removed_at: None,
stranded: false,
dissolved: false,
added_at_ms: 7,
};
save_state(&client, &state).await.expect("saves");
// A cached rumor is also an application-specific document, but not a
// state document, so the prefix keeps it out of the state scan.
let other = EventBuilder::new(Kind::ApplicationSpecificData, "{}")
.tags([Tag::identifier("deadbeef")])
.finalize(&*LOCAL_KEYS)
.expect("builds");
client.database().save_event(&other).await.expect("saves");
let loaded = load_states(&client).await.expect("loads");
assert_eq!(loaded, vec![state]);
});
}
#[test]
fn caching_the_same_rumor_twice_leaves_one_row() {
smol::block_on(async {
let client = client();
let channel = ChannelId::from_bytes([0x9c; 32]);
let keys = Keys::generate();
let group = concord::derive::channel_group_key(&[0x07; 32], &channel, Epoch(0))
.expect("a group key");
let rumor = concord::cord03::build_message(
keys.public_key(),
&channel,
Epoch(0),
"twice",
None,
1_700_000_000_000,
None,
);
let (wrap, _) = concord::cord03::seal_rumor(&rumor, &group, &keys, false)
.await
.expect("seals");
let (opened, _) =
concord::cord03::open(&wrap, &group, &channel, Epoch(0)).expect("opens");
assert!(
cache_rumor(&client, &channel, &opened)
.await
.expect("caches")
);
assert!(
cache_rumor(&client, &channel, &opened)
.await
.expect("caches")
);
let cached = query_rumors(&client, &channel, None, 10, None)
.await
.expect("reads");
assert_eq!(cached.len(), 1);
});
}
}
crates/community/src/community.rs
-1263
View File
File diff suppressed because it is too large Load Diff
crates/community/src/history.rs
-764
View File
@@ -1,764 +0,0 @@
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::future::Future;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::{Duration, Instant};
use anyhow::{Result, bail};
use concord::cord01::{KIND_WRAP_EPHEMERAL, OpenedStream};
use concord::cord03::{self, ChatRumor};
use concord::derive::channel_group_key;
use concord::state::{ChannelCursor, HeldKey};
use concord::{ChannelId, GroupKey};
use futures::future::{Either, select};
use nostr_sdk::prelude::*;
use crate::cache::cache_rumor;
use crate::sync::connect_relays;
/// How long one relay is given to answer one page of history.
const PAGE_TIMEOUT: Duration = Duration::from_secs(10);
/// How far below a cursor a warm window reaches back.
pub const CURSOR_OVERLAP: Duration = Duration::from_secs(60);
/// The region of history to read.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Window {
pub until: Option<Timestamp>,
pub since: Option<Timestamp>,
}
impl Window {
/// The newest wraps, with no older bound.
pub fn newest() -> Self {
Self::default()
}
/// The wraps strictly older than `oldest`.
pub fn older_than(oldest: Timestamp) -> Self {
Self {
until: Some(oldest - 1u64),
since: None,
}
}
/// The region between `since` and `oldest`, both inclusive.
pub fn between(since: Timestamp, oldest: Timestamp) -> Self {
Self {
until: Some(oldest - 1u64),
since: Some(since),
}
}
/// The window a channel is opened with.
pub fn opening(cursor: ChannelCursor) -> Self {
match cursor.newest {
Some(newest) => Self {
since: Some(newest - CURSOR_OVERLAP),
until: None,
},
None => Self::default(),
}
}
}
/// What a relay said about one page subscription.
#[derive(Debug, Clone)]
pub enum Settled {
Replayed,
Refused(String),
}
/// One relay's verdict on one page subscription.
#[derive(Debug, Clone)]
pub struct PageReport {
pub relay: RelayUrl,
pub outcome: Settled,
}
/// The page subscriptions in flight, by subscription id.
pub struct PageRegistry {
pages: Arc<Mutex<HashMap<SubscriptionId, flume::Sender<PageReport>>>>,
}
impl Default for PageRegistry {
fn default() -> Self {
Self {
pages: Arc::new(Mutex::new(HashMap::new())),
}
}
}
impl Clone for PageRegistry {
fn clone(&self) -> Self {
Self {
pages: Arc::clone(&self.pages),
}
}
}
impl PageRegistry {
pub fn register(&self, id: SubscriptionId, sender: flume::Sender<PageReport>) {
self.lock().insert(id, sender);
}
pub fn unregister(&self, id: &SubscriptionId) {
self.lock().remove(id);
}
/// Forget every page still in flight, because its round is gone.
pub fn clear(&self) {
self.lock().clear();
}
/// Hand a relay's verdict to the page that owns `id`, when it is still waiting.
pub fn deliver(&self, id: &SubscriptionId, relay: RelayUrl, outcome: Settled) {
let sender = self.lock().get(id).cloned();
let Some(sender) = sender else {
return;
};
if let Err(error) = sender.try_send(PageReport { relay, outcome }) {
log::debug!("community: a page report was not delivered: {error}");
}
}
fn lock(&self) -> MutexGuard<'_, HashMap<SubscriptionId, flume::Sender<PageReport>>> {
match self.pages.lock() {
Ok(pages) => pages,
Err(poisoned) => poisoned.into_inner(),
}
}
}
/// What one paged fetch saw.
#[derive(Debug, Clone, Default)]
pub struct WrapPage {
pub opened: Vec<ChatRumor>,
pub raw: usize,
/// Wraps that reached us under a held plane but that no held key could open.
pub unreadable: usize,
pub newest: Option<Timestamp>,
pub oldest: Option<Timestamp>,
pub exhausted: bool,
pub failed: bool,
pub errors: usize,
}
/// Walks a channel's history back over the community's own relays.
#[allow(clippy::too_many_arguments)]
pub async fn page(
client: &Client,
pages: &PageRegistry,
channel: &ChannelId,
held: &[HeldKey],
relays: &[RelayUrl],
window: Window,
max_pages: usize,
limit: usize,
) -> Result<WrapPage> {
let planes: Vec<(HeldKey, GroupKey)> = held
.iter()
.map(|key| Ok((*key, channel_group_key(&key.key, channel, key.epoch)?)))
.collect::<Result<Vec<_>>>()?;
let authors: Vec<PublicKey> = planes.iter().map(|(_, group)| group.pk()).collect();
if authors.is_empty() || relays.is_empty() || limit == 0 {
return Ok(WrapPage {
failed: true,
..WrapPage::default()
});
}
// A REQ can only target a relay the pool already knows about.
connect_relays(client, relays).await;
let mut walk = Walk::new(relays, window);
let mut opened = Vec::new();
for _ in 0..max_pages {
if walk.is_done() {
break;
}
let filter = wrap_filter(&authors, walk.region(), limit);
let asked: Vec<(usize, RelayUrl)> = walk
.live()
.map(|index| (index, walk.url(index).clone()))
.collect();
let answers = ask_page(client, pages, &asked, &filter).await;
for (index, url) in &asked {
match answers.get(url) {
Some(Settled::Replayed) => {}
Some(Settled::Refused(reason)) => {
log::warn!("community: relay {url} refused a history page: {reason}");
walk.reject(*index);
}
None => {
log::warn!("community: relay {url} did not finish a history page");
walk.reject(*index);
}
}
}
let answered = client.database().query(filter).await?;
for wrap in walk.accept(answered, limit) {
let Some((held, group)) = planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
else {
continue;
};
let Ok((stream, rumor)) = read_under(&wrap, held, group, channel) else {
walk.unreadable += 1;
continue;
};
if cache_rumor(client, channel, &stream).await? {
opened.push(rumor);
}
}
}
Ok(walk.finish(opened))
}
/// Opens one wrap under a held key.
fn read_under(
wrap: &Event,
held: &HeldKey,
group: &GroupKey,
channel: &ChannelId,
) -> Result<(OpenedStream, ChatRumor)> {
if held
.retired_at
.is_some_and(|retired| wrap.created_at > retired)
{
bail!("sealed after the key that reads it was retired");
}
Ok(cord03::open(wrap, group, channel, held.epoch)?)
}
/// The one filter a page is asked for.
fn wrap_filter(authors: &[PublicKey], window: Window, limit: usize) -> Filter {
let mut filter = Filter::new()
.kinds([Kind::GiftWrap, Kind::Custom(KIND_WRAP_EPHEMERAL)])
.authors(authors.iter().copied())
.limit(limit);
if let Some(until) = window.until {
filter = filter.until(until);
}
if let Some(since) = window.since {
filter = filter.since(since);
}
filter
}
/// One page's verdicts, by relay.
type PageAnswers = BTreeMap<RelayUrl, Settled>;
async fn ask_page(
client: &Client,
pages: &PageRegistry,
asked: &[(usize, RelayUrl)],
filter: &Filter,
) -> PageAnswers {
let mut answers = PageAnswers::new();
let distinct: BTreeSet<RelayUrl> = asked.iter().map(|(_, url)| url.clone()).collect();
let mut targets: Vec<(RelayUrl, Vec<Filter>)> = Vec::with_capacity(distinct.len());
for url in &distinct {
match client.relay(url).await {
Ok(Some(_)) => targets.push((url.clone(), vec![filter.clone()])),
Ok(None) => {
log::warn!("community: relay {url} is not in the pool for a history page");
answers.insert(
url.clone(),
Settled::Refused("not in the relay pool".to_owned()),
);
}
Err(error) => {
log::warn!("community: relay {url} could not be looked up: {error}");
answers.insert(url.clone(), Settled::Refused(error.to_string()));
}
}
}
if targets.is_empty() {
return answers;
}
let id = history_subscription();
let (sender, receiver) = flume::bounded(distinct.len());
pages.register(id.clone(), sender);
let options = SubscribeAutoCloseOptions::default()
.exit_policy(ReqExitPolicy::ExitOnEOSE)
.timeout(Some(PAGE_TIMEOUT));
match client
.subscribe(ReqTarget::manual(targets))
.with_id(id.clone())
.close_on(options)
.await
{
Ok(output) => {
for (url, reason) in output.failed {
answers.insert(url, Settled::Refused(reason));
}
let deadline = Instant::now() + PAGE_TIMEOUT;
while answers.len() < distinct.len() {
let remaining = deadline.saturating_duration_since(Instant::now());
let Some(report) = within(remaining, receiver.recv_async()).await else {
break;
};
match report {
Ok(report) => {
answers.insert(report.relay, report.outcome);
}
Err(error) => {
log::warn!("community: a history page's reports were lost: {error}");
break;
}
}
}
}
Err(error) => {
log::warn!("community: a history page REQ was refused: {error}");
}
}
pages.unregister(&id);
answers
}
pub(crate) fn auth_required(reason: &str) -> bool {
matches!(
MachineReadablePrefix::parse(reason),
Some(MachineReadablePrefix::AuthRequired)
)
}
fn history_subscription() -> SubscriptionId {
static NEXT: AtomicU64 = AtomicU64::new(0);
SubscriptionId::new(format!("history-{}", NEXT.fetch_add(1, Ordering::Relaxed)))
}
async fn within<F>(limit: Duration, future: F) -> Option<F::Output>
where
F: Future,
{
let future = std::pin::pin!(future);
let deadline = std::pin::pin!(smol::Timer::after(limit));
match select(future, deadline).await {
Either::Left((output, _)) => Some(output),
Either::Right(_) => None,
}
}
/// One pass over a channel's history, page by page.
#[derive(Debug)]
struct Walk {
relays: Vec<Walker>,
since: Option<Timestamp>,
/// The inclusive upper bound of the next page.
cursor: Option<Timestamp>,
seen: BTreeSet<EventId>,
newest: Option<Timestamp>,
oldest: Option<Timestamp>,
raw: usize,
errors: usize,
/// Wraps the caller could not read under any held key.
unreadable: usize,
/// A short page ended the walk.
bottom: bool,
}
/// One relay's standing in a walk.
#[derive(Debug)]
struct Walker {
url: RelayUrl,
dead: bool,
}
impl Walk {
fn new(relays: &[RelayUrl], window: Window) -> Self {
Self {
relays: relays
.iter()
.cloned()
.map(|url| Walker { url, dead: false })
.collect(),
since: window.since,
cursor: window.until,
seen: BTreeSet::new(),
newest: None,
oldest: None,
raw: 0,
errors: 0,
unreadable: 0,
bottom: false,
}
}
fn is_done(&self) -> bool {
self.bottom || self.relays.iter().all(|walker| walker.dead)
}
/// The region the next page asks for.
fn region(&self) -> Window {
Window {
until: self.cursor,
since: self.since,
}
}
fn live(&self) -> impl Iterator<Item = usize> + '_ {
(0..self.relays.len()).filter(|&index| !self.relays[index].dead)
}
fn url(&self, index: usize) -> &RelayUrl {
&self.relays[index].url
}
fn reject(&mut self, index: usize) {
self.relays[index].dead = true;
self.errors += 1;
}
fn accept(&mut self, page: BTreeSet<Event>, limit: usize) -> Vec<Event> {
if page.len() < limit {
self.bottom = true;
}
let mut oldest: Option<Timestamp> = None;
let mut events = Vec::with_capacity(page.len());
for event in page {
let at = event.created_at;
self.newest = Some(self.newest.map_or(at, |newest| newest.max(at)));
oldest = Some(oldest.map_or(at, |oldest| oldest.min(at)));
if self.seen.insert(event.id) {
self.raw += 1;
events.push(event);
}
}
// The walk's floor, which the round resumes the older pass from.
if let Some(oldest) = oldest {
self.oldest = Some(self.oldest.map_or(oldest, |held| held.min(oldest)));
}
match oldest {
Some(oldest) if !oldest.is_zero() => self.cursor = Some(oldest - 1u64),
Some(_) => self.bottom = true,
None => {}
}
events
}
fn finish(self, opened: Vec<ChatRumor>) -> WrapPage {
let swept = self.bottom && self.errors == 0;
WrapPage {
opened,
raw: self.raw,
unreadable: self.unreadable,
newest: self.newest,
oldest: self.oldest,
exhausted: swept && self.raw > 0,
failed: self.errors > 0 || (self.bottom && self.raw == 0),
errors: self.errors,
}
}
}
#[cfg(test)]
mod tests {
use std::cmp::Reverse;
use concord::Epoch;
use concord::cord03::{build_message, seal_rumor};
use concord::derive::channel_group_key;
use super::*;
const SECRET: [u8; 32] = [0x07u8; 32];
const NEXT_SECRET: [u8; 32] = [0x11u8; 32];
fn serve_page(database: &BTreeSet<Event>, window: Window, limit: usize) -> BTreeSet<Event> {
let mut events: Vec<Event> = database
.iter()
.filter(|event| {
window.until.is_none_or(|until| event.created_at <= until)
&& window.since.is_none_or(|since| event.created_at >= since)
})
.cloned()
.collect();
events.sort_by_key(|event| Reverse(event.created_at));
events.truncate(limit);
events.into_iter().collect()
}
fn relay_url(host: &str) -> RelayUrl {
RelayUrl::parse(&format!("wss://{host}.example.com")).expect("parses")
}
#[test]
fn a_walk_pages_back_across_a_rekey() {
let channel = ChannelId::from_bytes([0x9cu8; 32]);
let author = Keys::generate();
let held = [
HeldKey {
epoch: Epoch(0),
key: SECRET,
retired_at: None,
},
HeldKey {
epoch: Epoch(1),
key: NEXT_SECRET,
retired_at: None,
},
];
let planes: Vec<(HeldKey, GroupKey)> = held
.iter()
.map(|key| {
(
*key,
channel_group_key(&key.key, &channel, key.epoch).expect("derives"),
)
})
.collect();
// Three messages a second apart: a page boundary falls between each.
let base = 1_700_000_000_000;
let mut relay: BTreeSet<Event> = 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,
None,
);
relay.insert(
smol::block_on(seal_rumor(&rumor, &group, &author, false))
.expect("seals")
.0,
);
}
let mut walk = Walk::new(&[relay_url("history")], Window::newest());
let mut found = Vec::new();
let mut pages = 0;
while !walk.is_done() && pages < 10 {
pages += 1;
let page = serve_page(&relay, walk.region(), 2);
for wrap in walk.accept(page, 2) {
let Some((held, group)) =
planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
else {
continue;
};
let (_, rumor) = cord03::open(&wrap, group, &channel, held.epoch).expect("opens");
found.push(rumor);
}
}
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"]
);
let page = walk.finish(Vec::new());
assert!(page.exhausted);
assert!(!page.failed);
assert_eq!(page.raw, 3);
assert_eq!(page.newest, Some(Timestamp::from_secs(1_700_000_002)));
assert_eq!(
page.oldest,
Some(Timestamp::from_secs(1_700_000_000)),
"the walk reports its floor, which the older pass resumes below"
);
}
/// A wrap that reaches us and still will not open is history we cannot read,
/// not history that does not exist.
#[test]
fn a_wrap_no_held_key_can_open_reads_as_unreadable() {
let channel = ChannelId::from_bytes([0x9cu8; 32]);
let other = ChannelId::from_bytes([0x9du8; 32]);
let author = Keys::generate();
let group = channel_group_key(&SECRET, &channel, Epoch(0)).expect("derives");
let held = HeldKey {
epoch: Epoch(0),
key: SECRET,
retired_at: Some(Timestamp::from_secs(1_000)),
};
let wrap_at = |channel: &ChannelId, at_ms: u64| {
let rumor = build_message(
author.public_key(),
channel,
Epoch(0),
"sealed",
None,
at_ms,
None,
);
smol::block_on(seal_rumor(&rumor, &group, &author, false))
.expect("seals")
.0
};
// Sealed before the rotation superseded this key, so it still reads.
let before = wrap_at(&channel, 999_000);
assert!(read_under(&before, &held, &group, &channel).is_ok());
// Sealed after the cutoff the rotation set on that key.
let after = wrap_at(&channel, 1_001_000);
assert!(read_under(&after, &held, &group, &channel).is_err());
// Sealed to this plane but bound to another channel.
let misbound = wrap_at(&other, 999_000);
assert!(read_under(&misbound, &held, &group, &channel).is_err());
}
#[test]
fn an_empty_answer_never_seals_the_channel() {
let database: BTreeSet<Event> = BTreeSet::new();
let mut walk = Walk::new(&[relay_url("history")], Window::newest());
let page = serve_page(&database, walk.region(), 50);
assert!(walk.accept(page, 50).is_empty());
let page = walk.finish(Vec::new());
assert!(page.failed);
assert!(!page.exhausted);
assert_eq!(page.raw, 0);
assert_eq!(page.oldest, None);
}
#[test]
fn a_silent_relay_blocks_the_bottom() {
let database: BTreeSet<Event> = BTreeSet::new();
let mut walk = Walk::new(
&[relay_url("history"), relay_url("archive")],
Window::newest(),
);
// One relay answered the empty page; the other never answered at all, so
// its share of the region was never read and the walk must not seal.
walk.reject(1);
let page = serve_page(&database, walk.region(), 50);
assert!(walk.accept(page, 50).is_empty());
let page = walk.finish(Vec::new());
assert!(page.failed);
assert!(!page.exhausted);
assert_eq!(page.errors, 1);
}
#[test]
fn a_page_boundary_is_exclusive() {
let database = BTreeSet::from([
event_at(Timestamp::from_secs(1_700_000_000)),
event_at(Timestamp::from_secs(1_700_000_001)),
]);
let mut walk = Walk::new(&[relay_url("history")], Window::newest());
let first = walk.accept(serve_page(&database, walk.region(), 1), 1);
let second = walk.accept(serve_page(&database, walk.region(), 1), 1);
assert_eq!(first.len(), 1);
assert_eq!(second.len(), 1);
assert_ne!(first[0].id, second[0].id);
let oldest = second
.iter()
.map(|event| event.created_at)
.min()
.expect("one wrap");
assert_eq!(oldest, Timestamp::from_secs(1_700_000_000));
}
fn event_at(at: Timestamp) -> Event {
let keys = Keys::generate();
EventBuilder::new(Kind::TextNote, "page")
.custom_created_at(at)
.finalize(&keys)
.expect("signs")
}
/// A cold channel asks wide; a warm one resumes at the overlap above its
/// cursor, and `older_than` never includes the boundary event itself.
#[test]
fn a_cold_window_is_open_and_a_warm_one_resumes_at_the_overlap() {
assert_eq!(Window::opening(ChannelCursor::default()), Window::default());
let warm = Window::opening(ChannelCursor {
newest: Some(Timestamp::from_secs(2_000_000)),
oldest: Some(Timestamp::from_secs(1_000)),
exhausted: false,
});
assert_eq!(
warm,
Window {
since: Some(Timestamp::from_secs(2_000_000) - CURSOR_OVERLAP),
until: None,
}
);
assert_eq!(
Window::older_than(Timestamp::from_secs(1_000)),
Window {
since: None,
until: Some(Timestamp::from_secs(999)),
}
);
}
/// A page whose round has moved on is simply a report nobody reads.
#[test]
fn a_page_that_moved_on_receives_nothing() {
let pages = PageRegistry::default();
let id = SubscriptionId::new("concord-history-9");
let (sender, receiver) = flume::bounded(1);
pages.register(id.clone(), sender);
pages.unregister(&id);
pages.deliver(&id, relay_url("history"), Settled::Replayed);
assert!(receiver.try_recv().is_err());
}
}
crates/community/src/lib.rs
-829
View File
@@ -1,829 +0,0 @@
use std::collections::{BTreeSet, HashMap};
use std::time::{Duration, Instant};
use anyhow::Result;
pub use concord::cord02::CommunityMetadata;
pub use concord::cord03::{ChatMessage, ReplyRef};
use concord::state::CommunityState;
pub use concord::{ChannelId, CommunityId, Epoch};
use futures::future::{Either, select};
use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task};
use nostr_sdk::prelude::*;
use smallvec::{SmallVec, smallvec};
use state::NostrRegistry;
use crate::history::{PageRegistry, Settled, auth_required};
use crate::rekey::WatchRegistry;
pub mod cache;
mod community;
pub mod history;
mod rekey;
mod sync;
pub use community::*;
pub use sync::*;
/// How long a burst of relay notifications is collected before it is folded.
const PUMP_WINDOW: Duration = Duration::from_millis(200);
/// How long a community's relays may deliver nothing before
/// its standing subscription is torn down and re-issued.
const LIVE_ROTATE: Duration = Duration::from_secs(90);
pub fn init(cx: &mut App) {
CommunityRegistry::set_global(cx.new(CommunityRegistry::new), cx);
}
struct GlobalCommunityRegistry(Entity<CommunityRegistry>);
impl Global for GlobalCommunityRegistry {}
#[derive(Debug, Clone, PartialEq, Eq)]
enum Signal {
Event(CommunityId),
List,
Rekey(CommunityId),
}
/// Which standing subscription an id belongs to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Route {
List,
Community(CommunityId),
}
fn route_of(id: &SubscriptionId) -> Option<Route> {
if sync::is_list_subscription(id) {
return Some(Route::List);
}
sync::community_of(id).map(Route::Community)
}
/// What a window of relay notifications saw, waiting to be folded once.
#[derive(Default)]
struct Batch {
list: bool,
communities: BTreeSet<CommunityId>,
rekeys: BTreeSet<CommunityId>,
}
/// Whether the pump should keep listening.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Flow {
Continue,
Stop,
}
/// Fold one notification into the window, or settle a page it belongs to.
fn route(
notification: ClientNotification,
pages: &PageRegistry,
watches: &WatchRegistry,
batch: &mut Batch,
) -> Flow {
match notification {
ClientNotification::Event {
subscription_id, ..
} => match route_of(&subscription_id) {
Some(Route::List) => batch.list = true,
Some(Route::Community(id)) => {
batch.communities.insert(id);
}
None => {
if let Some(id) = watches.community_of(&subscription_id) {
batch.rekeys.insert(id);
}
}
},
ClientNotification::Message { relay_url, message } => match *message {
RelayMessage::EndOfStoredEvents(id) => {
pages.deliver(&id, relay_url, Settled::Replayed);
}
RelayMessage::Closed {
subscription_id,
message,
} if !auth_required(&message) => {
pages.deliver(
&subscription_id,
relay_url,
Settled::Refused(message.into_owned()),
);
}
_ => {}
},
ClientNotification::Shutdown => return Flow::Stop,
}
Flow::Continue
}
/// Hand the window's signals to the foreground consumer, one per community.
async fn flush(tx: &flume::Sender<Signal>, batch: &mut Batch) -> Result<()> {
for id in std::mem::take(&mut batch.communities) {
tx.send_async(Signal::Event(id)).await?;
}
for id in std::mem::take(&mut batch.rekeys) {
tx.send_async(Signal::Rekey(id)).await?;
}
if std::mem::take(&mut batch.list) {
tx.send_async(Signal::List).await?;
}
Ok(())
}
impl EventEmitter<CommunityEvent> for CommunityRegistry {}
pub struct CommunityRegistry {
communities: Vec<Entity<Community>>,
index: HashMap<CommunityId, Entity<Community>>,
/// The plane set each community was last subscribed with
synced: HashMap<CommunityId, SubscriptionKey>,
/// When a relay last delivered something for a community,
/// which is the only evidence the standing subscription is alive.
last_event: HashMap<CommunityId, Instant>,
/// One observer per tracked community, dropped on reset
observers: HashMap<CommunityId, Subscription>,
signal_tx: flume::Sender<Signal>,
signal_rx: flume::Receiver<Signal>,
/// The page subscriptions in flight, shared with the notification pump.
pages: PageRegistry,
/// The rekey watch subscriptions, resolved to their community.
watches: WatchRegistry,
tasks: SmallVec<[Task<Result<()>>; 2]>,
/// Notification listener task (cancelled on signer change)
notification_listener: Option<Task<Result<()>>>,
/// Signal consumer task (cancelled on signer change)
signal_consumer: Option<Task<Result<()>>>,
/// The round scheduler (cancelled on signer change)
scheduler: Option<Task<Result<()>>>,
_subscriptions: SmallVec<[Subscription; 2]>,
}
impl CommunityRegistry {
pub fn global(cx: &App) -> Entity<Self> {
cx.global::<GlobalCommunityRegistry>().0.clone()
}
fn set_global(state: Entity<Self>, cx: &mut App) {
cx.set_global(GlobalCommunityRegistry(state));
}
fn new(cx: &mut Context<Self>) -> Self {
let entity = cx.entity().downgrade();
let nostr = NostrRegistry::global(cx);
let (tx, rx) = flume::bounded::<Signal>(256);
let mut subscriptions = smallvec![];
subscriptions.push(cx.subscribe(&nostr, |this, _nostr, event, cx| {
if event.signer_changed() {
this.reset(cx);
this.handle_notifications(cx);
this.subscribe_list(cx);
this.load(cx);
}
}));
cx.defer(move |cx| {
entity
.update(cx, |this, cx| {
this.handle_notifications(cx);
if nostr.read(cx).current_user().is_some() {
this.subscribe_list(cx);
this.load(cx);
}
})
.ok();
});
Self {
communities: Vec::new(),
index: HashMap::new(),
synced: HashMap::new(),
last_event: HashMap::new(),
observers: HashMap::new(),
signal_tx: tx,
signal_rx: rx,
pages: PageRegistry::default(),
watches: WatchRegistry::default(),
tasks: smallvec![],
notification_listener: None,
signal_consumer: None,
scheduler: None,
_subscriptions: subscriptions,
}
}
pub fn communities(&self) -> &[Entity<Community>] {
&self.communities
}
pub fn community(&self, id: &CommunityId) -> Option<Entity<Community>> {
self.index.get(id).cloned()
}
/// Create a community owned by the current account and begin tracking it.
pub fn create(&mut self, metadata: CommunityMetadata, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
let current_user = nostr.read(cx).current_user();
if current_user.is_none() {
cx.emit(CommunityEvent::Error(
"cannot create a community without an account".to_owned(),
));
return;
}
let signer = nostr.read(cx).signer();
let client = nostr.read(cx).client();
let task =
cx.background_spawn(async move { sync::create(&client, &signer, &metadata).await });
self.tasks.push(cx.spawn(async move |this, cx| {
match task.await {
Ok(_state) => this.update(cx, |this, cx| this.load(cx))?,
Err(error) => {
this.update(cx, |_this, cx| {
cx.emit(CommunityEvent::Error(error.to_string()));
})?;
}
}
Ok(())
}));
}
/// Forget the current account and cancel everything in flight.
pub fn reset(&mut self, cx: &mut Context<Self>) {
self.notification_listener = None;
self.signal_consumer = None;
self.scheduler = None;
self.tasks.clear();
self.observers.clear();
self.pages.clear();
self.watches.clear();
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let ids: Vec<CommunityId> = self.index.keys().copied().collect();
for id in ids {
let client = client.clone();
let subscription = sync::subscription_id(&id);
let rekey = rekey::subscription_id(&id);
self.tasks.push(cx.background_spawn(async move {
client.unsubscribe(&subscription).await?;
client.unsubscribe(&rekey).await?;
Ok(())
}));
}
self.communities.clear();
self.index.clear();
self.synced.clear();
self.last_event.clear();
cx.notify();
}
/// Subscribe to the account's community list.
fn subscribe_list(&mut self, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
let signer = nostr.read(cx).signer();
let client = nostr.read(cx).client();
self.tasks.push(cx.spawn(async move |this, cx| {
let self_pk = signer.get_public_key_async().await?;
if let Err(error) = sync::subscribe_list(&client, self_pk).await {
this.update(cx, |_this, cx| {
cx.emit(CommunityEvent::Error(error.to_string()));
})?;
}
Ok(())
}));
}
/// Discover the account's communities in the local database.
fn load(&mut self, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
let signer = nostr.read(cx).signer();
let client = nostr.read(cx).client();
let task = cx.background_spawn(async move {
let self_pk = signer.get_public_key_async().await?;
sync::load(&client, &signer, self_pk).await
});
self.tasks.push(cx.spawn(async move |this, cx| {
match task.await {
Ok(states) => {
this.update(cx, |this, cx| this.track(states, cx))?;
}
Err(error) => {
this.update(cx, |_this, cx| {
cx.emit(CommunityEvent::Error(error.to_string()));
})?;
}
}
Ok(())
}));
}
/// Replace the tracked communities with a freshly loaded set.
///
/// A community that survives the reload keeps its entity, so an open panel
/// and a browsing sidebar stay pointed at a live community.
fn track(&mut self, states: Vec<CommunityState>, cx: &mut Context<Self>) {
let mut communities = Vec::with_capacity(states.len());
for state in states {
let id = state.id;
let community = match self.index.remove(&id) {
Some(community) => {
// The list can carry plane material the store does not.
if community.read(cx).state() != &state {
community.update(cx, |community, _cx| community.adopt(state));
}
community
}
None => {
let community = cx.new(|_| Community::new(state, self.pages.clone()));
self.observers.insert(
id,
cx.observe(&community, |this, _community, cx| {
this.sync_subscriptions(cx);
cx.notify();
}),
);
community
}
};
communities.push((id, community));
}
// Whatever the index still holds is no longer in the list.
let dropped: Vec<CommunityId> = self.index.keys().copied().collect();
for id in dropped {
self.observers.remove(&id);
self.synced.remove(&id);
self.last_event.remove(&id);
}
self.communities = communities
.iter()
.map(|(_, community)| community.clone())
.collect();
self.index = communities.into_iter().collect();
self.sync_subscriptions(cx);
// A backlog already in the database produces no notification, so fold it once.
for community in self.communities.clone() {
community.update(cx, |community, cx| community.refresh(cx));
}
cx.notify();
}
fn refresh(&mut self, id: CommunityId, cx: &mut Context<Self>) {
let Some(community) = self.index.get(&id).cloned() else {
return;
};
community.update(cx, |community, cx| community.refresh(cx));
}
/// Adopt whatever a community's rekey watch has delivered.
fn rekey(&mut self, id: CommunityId, cx: &mut Context<Self>) {
let Some(community) = self.index.get(&id).cloned() else {
return;
};
community.update(cx, |community, cx| community.rekey(cx));
}
/// One scheduler pass: every community repairs itself if it has gone stale,
/// and any whose relays have gone quiet is re-subscribed.
fn tick(&mut self, cx: &mut Context<Self>) {
for community in self.communities.clone() {
community.update(cx, |community, cx| community.tick(cx));
}
self.rotate_quiet(cx);
}
/// Re-issue the standing subscription of every community that has been quiet for `LIVE_ROTATE`
fn rotate_quiet(&mut self, cx: &mut Context<Self>) {
let now = Instant::now();
let ids: Vec<CommunityId> = self.index.keys().copied().collect();
let mut rotated = false;
for id in ids {
let quiet = self
.last_event
.get(&id)
.is_none_or(|at| now.duration_since(*at) >= LIVE_ROTATE);
if !quiet {
continue;
}
self.synced.remove(&id);
rotated = true;
}
if rotated {
self.sync_subscriptions(cx);
}
}
/// Re-subscribe every community whose held planes moved.
fn sync_subscriptions(&mut self, cx: &mut Context<Self>) {
let nostr = NostrRegistry::global(cx);
for community in self.communities.clone() {
let client = nostr.read(cx).client();
let (id, key, state) = {
let community = community.read(cx);
(
community.id(),
community.subscription_key(),
community.state().clone(),
)
};
if self.synced.get(&id) == Some(&key) {
continue;
}
let planes = match sync::planes(&state) {
Ok(planes) => planes,
Err(error) => {
cx.emit(CommunityEvent::Error(error.to_string()));
continue;
}
};
let subscription = sync::subscription_id(&id);
let filter = sync::live_filter(&planes, sync::live_window(&state, Timestamp::now()));
let relays = key.relays().to_vec();
// A fresh REQ counts as evidence of life for this community.
self.last_event.insert(id, Instant::now());
// The rekey watch is a second standing REQ over the same relays.
let watch = match rekey::watches(&state) {
Ok(watches) => rekey::watch_filter(&watches),
Err(error) => {
cx.emit(CommunityEvent::Error(error.to_string()));
continue;
}
};
let rekey_subscription = rekey::subscription_id(&id);
self.watches.register(rekey_subscription.clone(), id);
self.synced.insert(id, key);
self.tasks.push(cx.spawn(async move |this, cx| {
if let Err(error) = subscribe(&client, &subscription, &relays, filter).await {
this.update(cx, |_this, cx| {
cx.emit(CommunityEvent::Error(error.to_string()));
})?;
}
if let Err(error) = subscribe(&client, &rekey_subscription, &relays, watch).await {
this.update(cx, |_this, cx| {
cx.emit(CommunityEvent::Error(error.to_string()));
})?;
}
Ok(())
}));
}
}
fn handle_notifications(&mut self, cx: &mut Context<Self>) {
self.notification_listener = None;
self.signal_consumer = None;
self.scheduler = None;
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let tx = self.signal_tx.clone();
let rx = self.signal_rx.clone();
let pages = self.pages.clone();
let watches = self.watches.clone();
let executor = cx.background_executor().clone();
self.notification_listener = Some(cx.background_spawn(async move {
let mut notifications = client.notifications();
let mut batch = Batch::default();
'outer: loop {
match notifications.next().await {
Some(notification) => {
if route(notification, &pages, &watches, &mut batch) == Flow::Stop {
flush(&tx, &mut batch).await?;
break 'outer;
}
}
None => break 'outer,
}
let deadline = Instant::now() + PUMP_WINDOW;
loop {
let now = Instant::now();
if now >= deadline {
break;
}
let timer = executor.timer(deadline - now);
let next = notifications.next();
futures::pin_mut!(timer);
futures::pin_mut!(next);
match select(next, timer).await {
Either::Left((Some(notification), _)) => {
if route(notification, &pages, &watches, &mut batch) == Flow::Stop {
flush(&tx, &mut batch).await?;
break 'outer;
}
}
Either::Left((None, _)) => break 'outer,
Either::Right(_) => break,
}
}
flush(&tx, &mut batch).await?;
}
Ok(())
}));
self.signal_consumer = Some(cx.spawn(async move |this, cx| {
while let Ok(signal) = rx.recv_async().await {
match signal {
Signal::Event(id) => this.update(cx, |this, cx| {
// The only proof a community's subscription is still
// delivering anything.
this.last_event.insert(id, Instant::now());
this.refresh(id, cx);
})?,
Signal::Rekey(id) => this.update(cx, |this, cx| this.rekey(id, cx))?,
Signal::List => this.update(cx, |this, cx| this.load(cx))?,
}
}
Ok(())
}));
self.scheduler = Some(cx.spawn(async move |this, cx| {
loop {
cx.background_executor().timer(MIN_ROUND_INTERVAL).await;
if let Some(registry) = this.upgrade() {
registry.update(cx, |this, cx| {
this.tick(cx);
});
} else {
break;
}
}
Ok(())
}));
}
}
async fn subscribe(
client: &Client,
id: &SubscriptionId,
relays: &[RelayUrl],
filter: Filter,
) -> Result<()> {
client.unsubscribe(id).await?;
if relays.is_empty() {
log::warn!("community {id}: no relay to subscribe to");
return Ok(());
}
let mut targets: Vec<(RelayUrl, Vec<Filter>)> = Vec::with_capacity(relays.len());
for url in relays {
if let Err(error) = client.add_relay(url).and_connect().await {
log::warn!("community {id}: failed to add relay {url}: {error}");
}
match client.relay(url).await {
Ok(Some(_)) => targets.push((url.clone(), vec![filter.clone()])),
Ok(None) => log::warn!("community {id}: relay {url} is not in the pool"),
Err(error) => log::warn!("community {id}: relay {url} could not be looked up: {error}"),
}
}
if targets.is_empty() {
log::warn!("community {id}: no relay accepted the standing subscription");
return Ok(());
}
let output = client
.subscribe(ReqTarget::manual(targets))
.with_id(id.clone())
.await?;
if !output.failed.is_empty() {
log::warn!(
"community {id}: {} relay(s) rejected the subscription",
output.failed.len()
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn relay() -> RelayUrl {
RelayUrl::parse("wss://relay.example").expect("a url")
}
fn event(subscription_id: SubscriptionId) -> ClientNotification {
let keys = Keys::generate();
let event = EventBuilder::new(Kind::TextNote, "hi")
.finalize(&keys)
.expect("signs");
ClientNotification::Event {
relay_url: relay(),
subscription_id,
event: Box::new(event),
}
}
fn message(message: RelayMessage<'static>) -> ClientNotification {
ClientNotification::Message {
relay_url: relay(),
message: Box::new(message),
}
}
/// A burst within one window folds each community once, and the list once,
/// however many events arrived.
#[test]
fn a_burst_collapses_to_one_signal_per_community() {
let pages = PageRegistry::default();
let watches = WatchRegistry::default();
let community = CommunityId::from_bytes([0x42; 32]);
let mut batch = Batch::default();
let plane = event(sync::subscription_id(&community));
for _ in 0..50 {
assert_eq!(
route(plane.clone(), &pages, &watches, &mut batch),
Flow::Continue
);
}
route(
event(sync::list_subscription_id()),
&pages,
&watches,
&mut batch,
);
// A page's event is folded from the database later, not routed here.
route(
event(SubscriptionId::new("concord-history-7")),
&pages,
&watches,
&mut batch,
);
assert!(batch.list);
assert_eq!(batch.communities, BTreeSet::from([community]));
assert!(batch.rekeys.is_empty());
}
/// A rekey watch's wraps put themselves in the database; the pump's only job
/// is to wake the adoption pass, and it resolves the community by id.
#[test]
fn a_rekey_watch_event_wakes_its_community() {
let pages = PageRegistry::default();
let watches = WatchRegistry::default();
let community = CommunityId::from_bytes([0x42; 32]);
let id = rekey::subscription_id(&community);
watches.register(id.clone(), community);
let mut batch = Batch::default();
route(event(id), &pages, &watches, &mut batch);
assert_eq!(batch.rekeys, BTreeSet::from([community]));
assert!(batch.communities.is_empty());
}
#[test]
fn an_eose_settles_only_the_page_that_owns_the_id() {
let pages = PageRegistry::default();
let mine = SubscriptionId::new("concord-history-1");
let other = SubscriptionId::new("concord-history-2");
let (mine_tx, mine_rx) = flume::bounded(1);
let (other_tx, other_rx) = flume::bounded(1);
pages.register(mine.clone(), mine_tx);
pages.register(other, other_tx);
let mut batch = Batch::default();
let flow = route(
message(RelayMessage::eose(mine)),
&pages,
&WatchRegistry::default(),
&mut batch,
);
assert_eq!(flow, Flow::Continue);
assert!(matches!(
mine_rx.try_recv().expect("a report").outcome,
Settled::Replayed
));
assert!(other_rx.try_recv().is_err());
}
#[test]
fn a_refused_page_settles_its_relay_as_refused() {
let pages = PageRegistry::default();
let id = SubscriptionId::new("concord-history-1");
let (sender, receiver) = flume::bounded(1);
pages.register(id.clone(), sender);
let mut batch = Batch::default();
route(
message(RelayMessage::closed(id, "blocked: not allowed")),
&pages,
&WatchRegistry::default(),
&mut batch,
);
match receiver.try_recv().expect("a report").outcome {
Settled::Refused(reason) => assert!(reason.contains("blocked")),
other => panic!("expected a refusal, got {other:?}"),
}
}
/// The SDK re-issues an `auth-required` REQ under the same id after AUTH, so
/// the page keeps waiting rather than writing the relay off.
#[test]
fn an_auth_required_close_settles_nothing() {
let pages = PageRegistry::default();
let id = SubscriptionId::new("concord-history-1");
let (sender, receiver) = flume::bounded(1);
pages.register(id.clone(), sender);
let mut batch = Batch::default();
route(
message(RelayMessage::closed(
id,
"auth-required: please authenticate",
)),
&pages,
&WatchRegistry::default(),
&mut batch,
);
assert!(receiver.try_recv().is_err());
}
#[test]
fn a_shutdown_stops_the_pump() {
let pages = PageRegistry::default();
let mut batch = Batch::default();
assert_eq!(
route(
ClientNotification::Shutdown,
&pages,
&WatchRegistry::default(),
&mut batch
),
Flow::Stop
);
}
}
crates/community/src/rekey.rs
-1211
View File
File diff suppressed because it is too large Load Diff
crates/community/src/sync.rs
-1660
View File
File diff suppressed because it is too large Load Diff
crates/community_ui/Cargo.toml
-19
View File
@@ -1,19 +0,0 @@
[package]
name = "community_ui"
version.workspace = true
edition.workspace = true
publish.workspace = true
[dependencies]
community = { path = "../community" }
state = { path = "../state" }
ui = { path = "../ui" }
theme = { path = "../theme" }
common = { path = "../common" }
person = { path = "../person" }
settings = { path = "../settings" }
gpui.workspace = true
nostr-sdk.workspace = true
anyhow.workspace = true
smallvec.workspace = true
crates/community_ui/src/lib.rs
-832
View File
@@ -1,832 +0,0 @@
use std::collections::{BTreeMap, HashMap};
use std::fmt;
use anyhow::Result;
use community::{
ChannelId, ChatMessage, Community, CommunityEvent, Epoch, Intent, LOAD_OLDER_PAGES,
TIMELINE_PAGE, Timeline,
};
use gpui::prelude::FluentBuilder;
use gpui::{
AnyElement, App, AppContext, Context, Entity, EventEmitter, FocusHandle, Focusable, FollowMode,
IntoElement, ListAlignment, ListScrollEvent, ListState, ParentElement, Render, SharedString,
Styled, Subscription, Task, WeakEntity, Window, div, list, px,
};
use nostr_sdk::prelude::EventId;
use smallvec::{SmallVec, smallvec};
use theme::ActiveTheme;
use ui::avatar::Avatar;
use ui::button::{Button, ButtonVariants};
use ui::dock::{Panel, PanelEvent};
use ui::input::{InputEvent, Textarea, TextareaState};
use ui::markdown::RenderedText;
use ui::message::WelcomeMessage;
use ui::notification::Notification;
use ui::scroll::Scrollbar;
use ui::{Disableable, IconName, Sizable, WindowExtension, h_flex, v_flex};
mod message;
/// How near the top row a scroll has to come before the panel splices older history in.
const LOAD_OLDER_THRESHOLD: usize = 20;
/// A repeat message within this window keeps its run, so it carries no avatar or name.
const RUN_WINDOW_MS: u64 = 300_000;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Notice {
Stranded,
Removed(Epoch),
ChannelRemoved(Epoch),
MissingKey(Epoch),
Unreachable,
Unreadable(usize),
}
impl fmt::Display for Notice {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Notice::Stranded => formatter.write_str(
"This invite is stale, the community has rotated past the epoch it names",
),
Notice::Removed(epoch) => write!(
formatter,
"You were removed from this community at epoch {}. Its history stays readable",
epoch.0
),
Notice::ChannelRemoved(epoch) => write!(
formatter,
"A rotation removed you from this channel at epoch {}",
epoch.0
),
Notice::MissingKey(epoch) => write!(
formatter,
"Messages here can't be read yet, this channel's key for epoch {} is missing",
epoch.0
),
Notice::Unreachable => formatter.write_str("Couldn't reach the community's relays"),
Notice::Unreadable(1) => {
formatter.write_str("1 message here can't be read yet, no key we hold opens it")
}
Notice::Unreadable(count) => write!(
formatter,
"{count} messages here can't be read yet, no key we hold opens them"
),
}
}
}
impl Notice {
fn writable(self) -> bool {
matches!(self, Notice::Unreachable | Notice::Unreadable(_))
}
}
pub fn init(
community: Entity<Community>,
window: &mut Window,
cx: &mut App,
) -> Entity<CommunityPanel> {
cx.new(|cx| CommunityPanel::new(community, window, cx))
}
/// Community Panel
pub struct CommunityPanel {
id: SharedString,
focus_handle: FocusHandle,
/// Community
community: WeakEntity<Community>,
/// The selected channel
channel: Option<ChannelId>,
/// The selected channel's timeline (oldest first)
rows: Vec<ChatMessage>,
/// Rendered markdown content, keyed by message id and dropped when a row changes
rendered_texts_by_id: BTreeMap<EventId, RenderedText>,
/// Whether the store holds rows older than `rows`
has_more: bool,
/// A round or a page read is in flight
loading: bool,
/// Message list state
list_state: ListState,
/// Message input state
input: Entity<TextareaState>,
/// Spawned reads and publishes, cancelled when the panel closes
tasks: SmallVec<[Task<Result<()>>; 4]>,
/// Event subscriptions
_subscriptions: SmallVec<[Subscription; 2]>,
}
impl CommunityPanel {
pub fn new(community: Entity<Community>, window: &mut Window, cx: &mut Context<Self>) -> Self {
let (id, name, channel) = {
let community = community.read(cx);
(
SharedString::from(format!("community-{}", community.id().to_hex())),
community.name(),
community.active_channel(),
)
};
let input = cx.new(|cx| {
TextareaState::new(window, cx)
.placeholder(format!("Message {name}"))
.auto_grow(1, 20)
.clean_on_escape()
});
let mut subscriptions = smallvec![];
subscriptions.push(
cx.subscribe_in(&input, window, |this, _input, event, window, cx| {
if let InputEvent::PressEnter { .. } = event {
this.send(window, cx);
}
}),
);
subscriptions.push(cx.subscribe_in(
&community,
window,
|_this, _community, event, window, cx| {
match event {
CommunityEvent::Channel(..) => {
cx.defer_in(window, |this, window, cx| {
this.load(window, cx);
});
}
CommunityEvent::Error(error) => {
window.push_notification(Notification::error(error.clone()), cx);
}
_ => {
cx.defer_in(window, |this, window, cx| {
this.reload(window, cx);
});
}
};
},
));
cx.defer_in(window, |this, window, cx| {
this.list_state.set_follow_mode(FollowMode::Tail);
this.list_state.set_scroll_handler(cx.listener(
|this, event: &ListScrollEvent, window, cx| {
if event.visible_range.start <= LOAD_OLDER_THRESHOLD {
this.load_older(window, cx);
}
},
));
this.load(window, cx);
});
Self {
id,
focus_handle: cx.focus_handle(),
community: community.downgrade(),
channel,
rows: Vec::new(),
rendered_texts_by_id: BTreeMap::new(),
has_more: false,
loading: false,
list_state: ListState::new(0, ListAlignment::Bottom, px(1024.)),
input,
tasks: smallvec![],
_subscriptions: subscriptions,
}
}
/// The channel to show, following the community's selection.
fn resolve_channel(&mut self, cx: &App) -> Option<ChannelId> {
let channel = self
.community
.read_with(cx, |community, _cx| community.active_channel())
.ok()
.flatten();
if channel != self.channel {
self.channel = channel;
self.rows.clear();
self.rendered_texts_by_id.clear();
self.has_more = false;
self.loading = false;
self.list_state.reset(2);
}
channel
}
/// The list's item count: the welcome row, the load-older row, then every message row.
fn item_count(&self) -> usize {
self.rows.len() + 2
}
/// A timeline read, `before_ms` exclusive, or `None` for the newest rows.
fn read(
&self,
channel: ChannelId,
before_ms: Option<u64>,
cx: &App,
) -> Option<Task<Result<Timeline>>> {
self.community
.read_with(cx, |community, cx| {
community.timeline(&channel, before_ms, TIMELINE_PAGE, cx)
})
.ok()
}
/// A channel round, or `None` once the community is gone.
fn sync(
&self,
channel: ChannelId,
intent: Intent,
cx: &mut App,
) -> Option<Task<Result<community::Progress>>> {
self.community
.update(cx, |community, cx| {
community.sync_channel(&channel, intent, cx)
})
.ok()
}
/// Paint the selected channel's cache, then catch it up from the relays.
fn load(&mut self, window: &mut Window, cx: &mut Context<Self>) {
let Some(channel) = self.resolve_channel(cx) else {
return;
};
self.reload(window, cx);
// Opening a channel twice in a breath asks the relays once.
if self.due(channel, cx) {
self.round(channel, Intent::CatchUp, window, cx);
}
}
/// Whether the community would actually round `channel`, or just serve it.
fn due(&self, channel: ChannelId, cx: &App) -> bool {
self.community
.read_with(cx, |community, _cx| community.due(&channel))
.unwrap_or(true)
}
/// Why the room is not showing messages, when it is not simply empty.
fn notice(&self, cx: &App) -> Option<Notice> {
let channel = self.channel?;
self.community
.read_with(cx, |community, _cx| {
if community.stranded() {
return Some(Notice::Stranded);
}
if let Some(epoch) = community.removed_at() {
return Some(Notice::Removed(epoch));
}
if let Some(epoch) = community.channel_removed_at(&channel) {
return Some(Notice::ChannelRemoved(epoch));
}
if let Some(epoch) = community.missing_key(&channel) {
return Some(Notice::MissingKey(epoch));
}
if community
.progress(&channel)
.is_some_and(|progress| progress.failed && progress.errors > 0)
{
return Some(Notice::Unreachable);
}
let unreadable = community.unreadable(&channel);
(unreadable > 0).then_some(Notice::Unreadable(unreadable))
})
.ok()
.flatten()
}
/// Run a round for `channel`, its completion re-reads the timeline.
fn round(
&mut self,
channel: ChannelId,
intent: Intent,
window: &mut Window,
cx: &mut Context<Self>,
) {
let Some(round) = self.sync(channel, intent, cx) else {
return;
};
self.loading = true;
cx.notify();
let task = cx.spawn_in::<_, Result<()>>(window, async move |this, cx| {
let result = round.await;
this.update_in(cx, |this, window, cx| {
this.loading = false;
if let Err(error) = result {
window.push_notification(
Notification::error(error.to_string()).autohide(false),
cx,
);
}
})?;
Ok(())
});
self.tasks.push(task);
}
/// Run the newest round again after a relay failure.
fn retry(&mut self, window: &mut Window, cx: &mut Context<Self>) {
let Some(channel) = self.channel else {
return;
};
self.round(channel, Intent::CatchUp, window, cx);
}
/// Read the newest page and fold it into what is on screen.
fn reload(&mut self, window: &mut Window, cx: &mut Context<Self>) {
let Some(channel) = self.resolve_channel(cx) else {
return;
};
let Some(timeline) = self.read(channel, None, cx) else {
return;
};
let task = cx.spawn_in::<_, Result<()>>(window, async move |this, cx| {
match timeline.await {
Ok(timeline) => this.update(cx, |this, cx| this.apply(channel, timeline, cx))?,
Err(error) => {
this.update_in(cx, |_this, window, cx| {
window.push_notification(
Notification::error(error.to_string()).autohide(false),
cx,
);
})?;
}
}
Ok(())
});
self.tasks.push(task);
}
/// Splice the page of history above the oldest row on screen.
fn load_older(&mut self, window: &mut Window, cx: &mut Context<Self>) {
if self.loading || !self.has_more {
return;
}
let Some(channel) = self.channel else {
return;
};
let Some(before_ms) = self
.rows
.first()
.map(|message| message.at_ms.saturating_sub(1))
else {
return;
};
let Some(page) = self.read(channel, Some(before_ms), cx) else {
return;
};
self.loading = true;
let task = cx.spawn_in::<_, Result<()>>(window, async move |this, cx| {
let timeline = match page.await {
Ok(timeline) => timeline,
Err(error) => {
this.update_in(cx, |this, window, cx| {
this.loading = false;
window.push_notification(
Notification::error(error.to_string()).autohide(false),
cx,
);
})?;
return Ok(());
}
};
let swept = this.update(cx, |this, cx| {
this.prepend(channel, timeline, cx);
!this.has_more
})?;
if !swept {
this.update(cx, |this, _cx| this.loading = false)?;
return Ok(());
}
let round = this.update(cx, |this, cx| {
this.sync(
channel,
Intent::Older {
pages: LOAD_OLDER_PAGES,
},
cx,
)
})?;
let Some(round) = round else {
this.update(cx, |this, _cx| this.loading = false)?;
return Ok(());
};
match round.await {
Ok(_) => {
let page =
this.update(cx, |this, cx| this.read(channel, Some(before_ms), cx))?;
if let Some(page) = page {
match page.await {
Ok(timeline) => {
this.update(cx, |this, cx| this.prepend(channel, timeline, cx))?
}
Err(error) => this.update_in(cx, |_this, window, cx| {
window.push_notification(
Notification::error(error.to_string()).autohide(false),
cx,
);
})?,
}
}
this.update(cx, |this, _cx| this.loading = false)?;
}
Err(error) => {
this.update_in(cx, |this, window, cx| {
this.loading = false;
window.push_notification(
Notification::error(error.to_string()).autohide(false),
cx,
);
})?;
}
}
Ok(())
});
self.tasks.push(task);
}
/// Fold a freshly read window into the rows on screen.
fn apply(&mut self, channel: ChannelId, timeline: Timeline, cx: &mut Context<Self>) {
if self.channel != Some(channel) {
return;
}
let Timeline { messages, has_more } = timeline;
let connected = self
.rows
.last()
.is_some_and(|last| messages.iter().any(|message| message.id == last.id));
if !connected {
self.rows = messages;
self.rendered_texts_by_id.clear();
self.has_more = has_more;
self.list_state.reset(self.item_count());
cx.notify();
return;
}
self.merge(messages);
cx.notify();
}
/// Splice older rows in above what is on screen.
fn prepend(&mut self, channel: ChannelId, timeline: Timeline, cx: &mut Context<Self>) {
if self.channel != Some(channel) {
return;
}
self.has_more = timeline.has_more;
self.merge(timeline.messages);
cx.notify();
}
/// Fold read rows into what is on screen, keeping the list in time order.
fn merge(&mut self, messages: Vec<ChatMessage>) {
let mut shown: HashMap<EventId, usize> = self
.rows
.iter()
.enumerate()
.map(|(ix, message)| (message.id, ix))
.collect();
let mut fresh = Vec::new();
for message in messages {
match shown.get(&message.id).copied() {
Some(ix) => {
// Drop the cached render when the text changes, so edits re-parse.
if self.rows[ix].content != message.content {
self.rendered_texts_by_id.remove(&message.id);
}
self.rows[ix] = message;
}
None => fresh.push(message),
}
}
for message in fresh {
if shown.contains_key(&message.id) {
continue;
}
let at = self
.rows
.partition_point(|row| (row.at_ms, row.id) <= (message.at_ms, message.id));
shown.insert(message.id, at);
self.rows.insert(at, message);
// The welcome and load-older rows sit above the messages.
self.list_state.splice(at + 2..at + 2, 1);
}
}
fn send(&mut self, window: &mut Window, cx: &mut Context<Self>) {
let content = self.input.read(cx).value().trim().to_owned();
if content.is_empty() {
window.push_notification("Cannot send an empty message", cx);
return;
}
let Some(channel) = self.resolve_channel(cx) else {
return;
};
if let Some(notice) = self.notice(cx).filter(|notice| !notice.writable()) {
window.push_notification(Notification::error(notice.to_string()).autohide(false), cx);
return;
}
let Ok(send) = self.community.read_with(cx, |community, cx| {
community.send(&channel, &content, None, cx)
}) else {
return;
};
let Some(send) = send else {
window.push_notification(Notification::error("Failed to send the message"), cx);
return;
};
self.input.update(cx, |input, cx| {
input.set_value("", window, cx);
});
let task = cx.spawn_in::<_, Result<()>>(window, async move |this, cx| {
match send.await {
Ok(_) => {
this.update_in(cx, |this, window, cx| this.reload(window, cx))?;
}
Err(error) => {
this.update_in(cx, |_this, window, cx| {
window.push_notification(
Notification::error(error.to_string()).autohide(false),
cx,
);
})?;
}
}
Ok(())
});
self.tasks.push(task);
}
/// The honest reason this room has nothing to show, with a way out of it.
fn render_notice(&self, notice: Notice, cx: &mut Context<Self>) -> AnyElement {
h_flex()
.w_full()
.justify_center()
.items_center()
.gap_2()
.px_3()
.py_2()
.text_sm()
.text_color(cx.theme().text_placeholder)
.child(notice.to_string())
.when(notice == Notice::Unreachable, |this| {
this.child(
Button::new("retry-round")
.label("Retry")
.ghost()
.small()
.loading(self.loading)
.on_click(cx.listener(|this, _event, window, cx| this.retry(window, cx))),
)
})
.into_any_element()
}
/// The row at index 0: the welcome message for the channel.
fn render_welcome(&self, cx: &Context<Self>) -> AnyElement {
let (name, avatar) = self
.community
.read_with(cx, |community, _cx| {
let seed = community.id().to_hex();
let avatar = Avatar::from_source(community.icon()).seed(seed).large();
(community.name(), avatar)
})
.unwrap_or_else(|_| (SharedString::from("this community"), Avatar::new(None)));
WelcomeMessage::new("welcome")
.icon(avatar)
.title(format!("Welcome to {}", name))
.message(format!("This is the start of the {} channel.", name))
.into_any_element()
}
/// The row at index 1: the affordance that pages older history in.
fn render_older(&self, cx: &mut Context<Self>) -> AnyElement {
if !self.has_more {
return div().into_any_element();
}
h_flex()
.w_full()
.justify_center()
.py_2()
.child(
Button::new("load-older")
.label(if self.loading {
"Loading earlier messages…"
} else {
"Load earlier messages"
})
.ghost()
.small()
.loading(self.loading)
.on_click(cx.listener(|this, _event, window, cx| this.load_older(window, cx))),
)
.into_any_element()
}
/// Whether the row at `index` opens a run from one author.
fn opens_run(&self, index: usize) -> bool {
let Some(current) = self.rows.get(index) else {
return true;
};
let Some(previous) = index.checked_sub(1).and_then(|index| self.rows.get(index)) else {
return true;
};
current.author != previous.author
|| current.at_ms.saturating_sub(previous.at_ms) > RUN_WINDOW_MS
}
fn render_message(
&mut self,
ix: usize,
window: &mut Window,
cx: &mut Context<Self>,
) -> AnyElement {
if ix == 0 {
return self.render_welcome(cx);
}
if ix == 1 {
return self.render_older(cx);
}
// The welcome and load-older rows sit above the messages.
let Some(message) = self.rows.get(ix - 2) else {
return div().into_any_element();
};
let show_author = self.opens_run(ix - 2);
let content = if message.deleted {
message::deleted(cx)
} else {
self.rendered_texts_by_id
.entry(message.id)
.or_insert_with(|| RenderedText::new(&message.content, &[], true))
.element(ix.into(), window, cx)
};
message::render(ix, message, content, show_author, cx)
}
fn render_composer(&self, cx: &mut Context<Self>) -> impl IntoElement {
let writable = self.notice(cx).is_none_or(|notice| notice.writable());
h_flex()
.flex_shrink_0()
.w_full()
.p_2()
.gap_1()
.items_end()
.child(Textarea::new(&self.input).appearance(false).flex_1())
.child(
Button::new("send")
.icon(IconName::PaperPlaneFill)
.tooltip("Send")
.ghost()
.large()
.disabled(!writable)
.on_click(cx.listener(|this, _event, window, cx| {
this.send(window, cx);
})),
)
}
}
impl Panel for CommunityPanel {
fn panel_id(&self) -> SharedString {
self.id.clone()
}
fn title(&self, cx: &App) -> AnyElement {
self.community
.read_with(cx, |community, _cx| {
let seed = community.id().to_hex();
let avatar = Avatar::from_source(community.icon()).seed(seed).xsmall();
h_flex()
.gap_1()
.text_xs()
.child(avatar)
.child(community.name())
.into_any_element()
})
.unwrap_or_else(|_| div().text_xs().child("Unknown").into_any_element())
}
fn toolbar_buttons(&self, _window: &Window, _cx: &App) -> Vec<Button> {
vec![]
}
}
impl EventEmitter<PanelEvent> for CommunityPanel {}
impl Focusable for CommunityPanel {
fn focus_handle(&self, _: &App) -> FocusHandle {
self.focus_handle.clone()
}
}
impl Render for CommunityPanel {
fn render(&mut self, _window: &mut Window, cx: &mut Context<Self>) -> impl IntoElement {
v_flex()
.size_full()
.min_w_0()
.child(
v_flex()
.flex_1()
.min_h_0()
.relative()
.map(|this| {
let notice = self.notice(cx);
if self.rows.is_empty() {
this.child(
v_flex().size_full().justify_center().child(match notice {
Some(notice) => self.render_notice(notice, cx),
None => h_flex()
.size_full()
.justify_center()
.text_sm()
.text_color(cx.theme().text_placeholder)
.child("No messages yet")
.into_any_element(),
}),
)
} else {
this.when_some(notice, |this, notice| {
this.child(self.render_notice(notice, cx))
})
.child(
list(
self.list_state.clone(),
cx.processor(move |this, ix, window, cx| {
this.render_message(ix, window, cx)
}),
)
.size_full(),
)
}
})
.child(Scrollbar::vertical(&self.list_state)),
)
.child(self.render_composer(cx))
}
}
crates/community_ui/src/message.rs
-77
View File
@@ -1,77 +0,0 @@
use std::collections::BTreeMap;
use common::TimestampExt;
use community::ChatMessage;
use gpui::prelude::FluentBuilder;
use gpui::{AnyElement, App, IntoElement, ParentElement, SharedString, Styled, div};
use nostr_sdk::prelude::*;
use person::PersonRegistry;
use settings::AppSettings;
use theme::ActiveTheme;
use ui::avatar::Avatar;
use ui::h_flex;
use ui::message::MessageRow;
pub(crate) fn render(
ix: usize,
message: &ChatMessage,
content: AnyElement,
show_author: bool,
cx: &App,
) -> AnyElement {
let persons = PersonRegistry::global(cx);
let author = persons.read(cx).get(&message.author, cx);
let hide_avatar = AppSettings::get_hide_avatar(cx);
MessageRow::new(ix)
.show_author(show_author)
.hide_avatar(hide_avatar)
.avatar(
Avatar::new(author.avatar())
.seed(author.avatar_seed())
.flex_shrink_0(),
)
.author(author.name())
.timestamp(Timestamp::from_secs(message.at_ms / 1000).to_human_time())
.when(message.edited_at.is_some(), |this| {
this.header_extra(div().child("(edited)"))
})
.child(content)
.when(!message.reactions.is_empty(), |this| {
this.child(reactions(message, cx))
})
.into_any_element()
}
/// The placeholder shown in place of the body of a deleted message.
pub(crate) fn deleted(cx: &App) -> AnyElement {
div()
.text_color(cx.theme().text_danger)
.child("Message deleted")
.into_any_element()
}
fn reactions(message: &ChatMessage, cx: &App) -> AnyElement {
let mut grouped: BTreeMap<&str, usize> = BTreeMap::new();
for emoji in message.reactions.values() {
*grouped.entry(emoji.as_str()).or_default() += 1;
}
h_flex()
.mt_1()
.gap_1()
.children(grouped.into_iter().map(|(emoji, count)| {
h_flex()
.gap_1()
.py_0p5()
.px_1()
.rounded(cx.theme().radius)
.border_1()
.border_color(cx.theme().border)
.text_xs()
.child(SharedString::from(emoji))
.child(SharedString::from(count.to_string()))
}))
.into_any_element()
}
crates/workspace/Cargo.toml
-2
View File
@@ -12,8 +12,6 @@ state = { path = "../state" }
device = { path = "../device" } device = { path = "../device" }
chat = { path = "../chat" } chat = { path = "../chat" }
chat_ui = { path = "../chat_ui" } chat_ui = { path = "../chat_ui" }
community = { path = "../community" }
community_ui = { path = "../community_ui" }
nip29 = { path = "../nip29" } nip29 = { path = "../nip29" }
nip29_ui = { path = "../nip29_ui" } nip29_ui = { path = "../nip29_ui" }
settings = { path = "../settings" } settings = { path = "../settings" }
desktop/Cargo.toml
-1
View File
@@ -35,7 +35,6 @@ common = { path = "../crates/common" }
state = { path = "../crates/state" } state = { path = "../crates/state" }
device = { path = "../crates/device" } device = { path = "../crates/device" }
chat = { path = "../crates/chat" } chat = { path = "../crates/chat" }
community = { path = "../crates/community" }
nip29 = { path = "../crates/nip29" } nip29 = { path = "../crates/nip29" }
settings = { path = "../crates/settings" } settings = { path = "../crates/settings" }
auto_update = { path = "../crates/auto_update" } auto_update = { path = "../crates/auto_update" }
desktop/src/main.rs
-3
View File
@@ -78,9 +78,6 @@ fn main() {
// Initialize app registry // Initialize app registry
chat::init(cx); chat::init(cx);
// Initialize community registry
community::init(cx);
// Initialize groups registry // Initialize groups registry
nip29::init(cx); nip29::init(cx);
web/Cargo.toml
-1
View File
@@ -16,7 +16,6 @@ common = { path = "../crates/common" }
state = { path = "../crates/state" } state = { path = "../crates/state" }
device = { path = "../crates/device" } device = { path = "../crates/device" }
chat = { path = "../crates/chat" } chat = { path = "../crates/chat" }
community = { path = "../crates/community" }
nip29 = { path = "../crates/nip29" } nip29 = { path = "../crates/nip29" }
settings = { path = "../crates/settings" } settings = { path = "../crates/settings" }
person = { path = "../crates/person" } person = { path = "../crates/person" }
web/src/lib.rs
-3
View File
@@ -70,9 +70,6 @@ pub fn run() -> Result<(), JsValue> {
// Initialize app registry // Initialize app registry
chat::init(cx); chat::init(cx);
// Initialize community registry
community::init(cx);
// Initialize groups registry // Initialize groups registry
nip29::init(cx); nip29::init(cx);