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

clean up and add community invite

Browse Source
This commit is contained in:
reya
2026-09-29 14:16:58 +07:00
parent 86ebf4dc62
commit 1a123f6d03
12 changed files with 366 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/community/Cargo.toml
+1
View File
@@ -12,6 +12,7 @@ gpui.workspace = true
nostr-sdk.workspace = true
anyhow.workspace = true
data-encoding.workspace = true
flume.workspace = true
futures.workspace = true
log.workspace = true
crates/community/src/community.rs
+18 -25
View File
@@ -112,6 +112,10 @@ pub enum CommunityEvent {
Unreadable(CommunityId),
/// The last round could not reach the community's relays.
Failed(CommunityId),
Invite {
id: CommunityId,
url: String,
},
Error(String),
}
@@ -336,8 +340,7 @@ impl Community {
cx.background_spawn(async move {
let progress = match receiver.recv_async().await {
Ok(progress) => progress,
Err(error) => {
log::warn!("community: channel round cancelled: {error}");
Err(_) => {
return Ok(Progress {
failed: true,
..Progress::default()
@@ -377,9 +380,8 @@ impl Community {
let finisher = cx.spawn(async move |this, cx| {
let result = round.await;
if let Err(error) = this.update(cx, |this, cx| this.finish_round(channel, result, cx)) {
log::warn!("community: a channel round outlived its community: {error}");
}
this.update(cx, |this, cx| this.finish_round(channel, result, cx))
.ok();
Ok(())
});
@@ -410,9 +412,7 @@ impl Community {
let waiters = std::mem::take(&mut round.waiters);
for waiter in waiters {
if let Err(error) = waiter.try_send(outcome.clone()) {
log::warn!("community: a channel round result was not delivered: {error}");
}
waiter.try_send(outcome.clone()).ok();
}
if let Some(intent) = queued {
@@ -620,7 +620,6 @@ impl Community {
let (opened, _) = cord03::open(&wrap, &group, &channel, epoch)?;
cache::cache_rumor(&client, &channel, &opened).await?;
sync::connect_relays(&client, &relays).await;
sync::publish_wrap(&client, &wrap, &relays).await;
Ok(opened.rumor_id)
@@ -967,14 +966,11 @@ impl Community {
};
self.icon_task = Some(cx.spawn(async move |this, cx| {
match sync::resolve_image(&icon, cx).await {
Ok(path) => {
this.update(cx, |this, cx| {
this.icon = Some(path);
cx.notify();
})?;
}
Err(error) => log::warn!("community icon: {error}"),
if let Ok(path) = sync::resolve_image(&icon, cx).await {
this.update(cx, |this, cx| {
this.icon = Some(path);
cx.notify();
})?;
}
Ok(())
}));
@@ -993,14 +989,11 @@ impl Community {
};
self.banner_task = Some(cx.spawn(async move |this, cx| {
match sync::resolve_image(&banner, cx).await {
Ok(path) => {
this.update(cx, |this, cx| {
this.banner = Some(path);
cx.notify();
})?;
}
Err(error) => log::warn!("community banner: {error}"),
if let Ok(path) = sync::resolve_image(&banner, cx).await {
this.update(cx, |this, cx| {
this.banner = Some(path);
cx.notify();
})?;
}
Ok(())
}));
crates/community/src/history.rs
+1 -4
View File
@@ -197,10 +197,7 @@ pub async fn page(
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);
}
None => walk.reject(*index),
}
}
crates/community/src/lib.rs
+55 -8
View File
@@ -257,6 +257,60 @@ impl CommunityRegistry {
}));
}
pub fn invite(&mut self, id: &CommunityId, cx: &mut Context<Self>) {
let Some(community) = self.index.get(id).cloned() else {
return;
};
let nostr = NostrRegistry::global(cx);
let signer = nostr.read(cx).signer();
let client = nostr.read(cx).client();
let state = community.read(cx).state().clone();
let id = *id;
let task =
cx.background_spawn(async move { sync::create_invite(&client, &signer, &state).await });
self.tasks.push(cx.spawn(async move |this, cx| {
match task.await {
Ok(url) => {
this.update(cx, |_this, cx| {
cx.emit(CommunityEvent::Invite { id, url });
})?;
}
Err(error) => {
this.update(cx, |_this, cx| {
cx.emit(CommunityEvent::Error(error.to_string()));
})?;
}
}
Ok(())
}));
}
pub fn join(&mut self, url: String, 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 { sync::join(&client, &signer, &url).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;
@@ -644,18 +698,11 @@ async fn subscribe(
return Ok(());
}
let output = client
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(())
}
crates/community/src/rekey.rs
-4
View File
@@ -447,10 +447,6 @@ pub async fn rotate(
if let RotationPlan::Base(refounding) = &plan {
let carried = carry_heads(client, state, refounding, at).await?;
if carried.is_empty() && !state.heads.is_empty() {
log::warn!("community: a refounding carried no settled head forward");
}
wraps.extend(carried);
}
crates/community/src/sync.rs
+201 -30
View File
@@ -1,5 +1,6 @@
use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use std::time::Duration;
use anyhow::{Context, Result};
use concord::cord01::KIND_WRAP_EPHEMERAL;
@@ -7,11 +8,16 @@ use concord::cord02::list::{CommunityList, JoinMaterial, KIND_COMMUNITY_LIST};
use concord::cord02::{self, ControlFold, ImageRef};
use concord::cord04::AuthorityCitation;
use concord::cord04::roles::{Permissions, citation_ok};
use concord::cord05::{
self, BundleState, ChannelGrant, CommunityInvite, InviteEntry, InviteList, ParsedInviteLink,
};
use concord::derive::{
channel_group_key, control_group_key, control_signer_group_key, guestbook_group_key,
TOKEN_LEN, channel_group_key, control_group_key, control_signer_group_key, guestbook_group_key,
invite_bundle_key,
};
use concord::state::{CommunityState, HeldKey, HeldRoot, list_entry};
use concord::{ChannelId, CommunityId, Epoch, GroupKey, decode_hex_32};
use data_encoding::HEXLOWER;
use gpui::AsyncApp;
use nostr_sdk::prelude::*;
use state::UniversalSigner;
@@ -21,6 +27,7 @@ use crate::history::{CURSOR_OVERLAP, Window};
/// How much of what a relay stores a cold subscription replays per relay.
const LIVE_REPLAY: usize = 500;
const INVITE_BASE: &str = "https://coopchat.xyz";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum PlaneKind {
@@ -213,10 +220,194 @@ where
Ok(state)
}
pub async fn create_invite<S>(client: &Client, signer: &S, state: &CommunityState) -> Result<String>
where
S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + ?Sized,
{
let link_signer = Keys::generate();
let mut token = [0u8; TOKEN_LEN];
token.copy_from_slice(&link_signer.secret_key().as_secret_bytes()[..TOKEN_LEN]);
let bundle_key = invite_bundle_key(&token);
let relays: Vec<String> = state.relays.iter().map(|relay| relay.to_string()).collect();
let channels = state
.channels
.iter()
.map(|channel| ChannelGrant {
id: channel.id,
key: channel.key.map(|key| HEXLOWER.encode(&key)),
epoch: channel.epoch,
name: channel.name.clone(),
extra: Default::default(),
})
.collect();
let invite = CommunityInvite {
community_id: state.id,
owner: state.owner,
owner_salt: HEXLOWER.encode(&state.owner_salt),
community_root: HEXLOWER.encode(&state.community_root),
root_epoch: state.root_epoch,
control_pk: state.control_pks.get(&state.root_epoch.0).copied(),
channels,
relays: relays.clone(),
name: state.name.clone().unwrap_or_default(),
icon: None,
expires_at: None,
creator_npub: None,
label: None,
extra: Default::default(),
};
let bundle = cord05::build_bundle_event(&link_signer, &invite, &bundle_key)?;
publish_wrap(client, &bundle, &state.relays).await;
let url = cord05::build_invite_url(INVITE_BASE, &link_signer.public_key(), &token, &relays)?;
let entry = InviteEntry {
token: HEXLOWER.encode(&token),
signer_sk: link_signer.secret_key().to_secret_hex(),
community_id: state.id,
url: url.clone(),
label: None,
created_at: Timestamp::now().as_secs().saturating_mul(1000),
expires_at: None,
extra: Default::default(),
};
record_invite(client, signer, entry).await?;
Ok(url)
}
pub async fn join<S>(client: &Client, signer: &S, url: &str) -> Result<CommunityState>
where
S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + ?Sized,
{
let link = cord05::parse_link(url)?;
let invite = resolve_invite(client, &link).await?;
let material = cord02::list::join_material(&invite, None);
let self_pk = signer.get_public_key_async().await?;
let at_ms = Timestamp::now().as_secs().saturating_mul(1000);
let state = CommunityState::from_join_material(&material, at_ms)?;
cache::save_state(client, &state).await?;
let group = guestbook_group_key(&state.community_root, &state.id, state.root_epoch)?;
let rumor = cord02::guestbook::build_join(self_pk, None, at_ms);
let (wrap, _) = cord02::guestbook::seal_rumor(&rumor, &group, signer).await?;
publish_wrap(client, &wrap, &state.relays).await;
if let Err(error) = record_membership(client, signer, &state, &invite.name).await {
log::warn!(
"community {}: recording the membership failed: {error}",
state.id.to_hex()
);
}
Ok(state)
}
async fn resolve_invite(client: &Client, link: &ParsedInviteLink) -> Result<CommunityInvite> {
let relays: Vec<RelayUrl> = link
.bootstrap_relays
.iter()
.filter_map(|relay| RelayUrl::parse(relay).ok())
.collect();
if relays.is_empty() {
anyhow::bail!("invite link names no relays");
}
connect_relays(client, &relays).await;
let filter = Filter::new()
.kind(Kind::Custom(cord05::KIND_BUNDLE))
.author(link.link_signer)
.identifier("");
let targets: Vec<(RelayUrl, Vec<Filter>)> = relays
.into_iter()
.map(|relay| (relay, vec![filter.clone()]))
.collect();
let events = client
.fetch_events(targets)
.timeout(Duration::from_secs(10))
.await?;
let Some(event) = events.into_iter().max_by_key(|event| event.created_at) else {
anyhow::bail!("invite link not found on its relays");
};
let bundle_key = invite_bundle_key(&link.token);
match cord05::parse_bundle_event(&event, &link.link_signer, &bundle_key)? {
BundleState::Live(invite) => Ok(*invite),
BundleState::Revoked => anyhow::bail!("invite link has been revoked"),
}
}
async fn record_invite<S>(client: &Client, signer: &S, entry: InviteEntry) -> Result<()>
where
S: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + ?Sized,
{
let self_pk = signer.get_public_key_async().await?;
let held = load_invite_list(client, signer, self_pk)
.await?
.unwrap_or_default();
let incoming = InviteList {
entries: vec![entry],
..Default::default()
};
let list = cord05::merge_invite_lists(held, incoming);
let event = cord05::build_invite_list(signer, &list).await?;
publish_list(client, &event).await;
Ok(())
}
async fn load_invite_list<S>(
client: &Client,
signer: &S,
self_pk: PublicKey,
) -> Result<Option<InviteList>>
where
S: AsyncGetPublicKey + AsyncNip44 + ?Sized,
{
let filter = Filter::new()
.kind(Kind::Custom(cord05::KIND_INVITE_LIST))
.author(self_pk);
let mut merged: Option<InviteList> = None;
for event in client.database().query(filter).await? {
match cord05::parse_invite_list(signer, &event).await {
Ok(list) => {
merged = Some(match merged {
Some(held) => cord05::merge_invite_lists(held, list),
None => list,
});
}
Err(error) => {
log::warn!("ignoring unreadable invite list {}: {error}", event.id);
}
}
}
Ok(merged)
}
/// Best-effort publication of the genesis wraps to the community's relays.
pub(crate) async fn publish_wraps(client: &Client, wraps: &[Event], relays: &[RelayUrl]) {
connect_relays(client, relays).await;
for wrap in wraps {
publish_wrap(client, wrap, relays).await;
}
@@ -233,20 +424,16 @@ pub(crate) async fn connect_relays(client: &Client, relays: &[RelayUrl]) {
/// Best-effort publication of a single wrap to the community's relays.
pub(crate) async fn publish_wrap(client: &Client, wrap: &Event, relays: &[RelayUrl]) {
connect_relays(client, relays).await;
let sent = if relays.is_empty() {
client.send_event(wrap).broadcast().await
} else {
client.send_event(wrap).to(relays.iter().cloned()).await
};
match sent {
Ok(output) if output.failed.is_empty() => {}
Ok(output) => log::warn!(
"community: {} relay(s) rejected {}",
output.failed.len(),
wrap.id
),
Err(error) => log::warn!("community: publishing {} failed: {error}", wrap.id),
if let Err(error) = sent {
log::warn!("community: publishing {} failed: {error}", wrap.id);
}
}
@@ -265,10 +452,6 @@ where
let frags = held.as_ref().map_or(1, |list| list.frags);
if frags > 1 {
log::warn!(
"community {}: the list spans {frags} fragments; deferring the membership write",
state.id.to_hex()
);
return Ok(());
}
@@ -289,13 +472,8 @@ where
}
async fn publish_list(client: &Client, event: &Event) {
match client.send_event(event).to_nip65().await {
Ok(output) if output.failed.is_empty() => {}
Ok(output) => log::warn!(
"community list: {} relay(s) rejected the publish",
output.failed.len()
),
Err(error) => log::warn!("community list: publish failed: {error}"),
if let Err(error) = client.send_event(event).to_nip65().await {
log::warn!("community list: publish failed: {error}");
}
}
@@ -328,18 +506,11 @@ pub async fn subscribe_list(client: &Client, self_pk: PublicKey) -> Result<()> {
.kind(Kind::Custom(KIND_COMMUNITY_LIST))
.author(self_pk);
let output = client
client
.subscribe(ReqTarget::auto(vec![filter]))
.with_id(id)
.await?;
if !output.failed.is_empty() {
log::warn!(
"community list: {} relay(s) rejected the subscription",
output.failed.len()
);
}
Ok(())
}