update community backend
This commit is contained in:
@@ -3,14 +3,19 @@ use std::path::PathBuf;
|
||||
|
||||
use anyhow::Result;
|
||||
use concord::cord02::{ControlFold, ImageRef};
|
||||
use concord::store::{ChannelKeyRef, CommunityState};
|
||||
use concord::cord03::{self, ChatMessage, ReplyRef};
|
||||
use concord::cord04::roles::{Permissions, citation_ok};
|
||||
use concord::derive::channel_group_key;
|
||||
use concord::store::{self, ChannelKeyRef, CommunityState};
|
||||
use concord::{ChannelId, CommunityId, Epoch};
|
||||
use gpui::{AppContext, Context, EventEmitter, Task};
|
||||
use gpui::{App, AppContext, Context, EventEmitter, Task};
|
||||
use nostr_sdk::prelude::*;
|
||||
use state::NostrRegistry;
|
||||
|
||||
use crate::sync::{self, Snapshot};
|
||||
|
||||
const MESSAGE_LIMIT: usize = 200;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct SubscriptionKey {
|
||||
control_pks: BTreeMap<u64, PublicKey>,
|
||||
@@ -39,6 +44,7 @@ impl SubscriptionKey {
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum CommunityEvent {
|
||||
Updated(CommunityId),
|
||||
Open(CommunityId),
|
||||
Error(String),
|
||||
}
|
||||
|
||||
@@ -109,6 +115,132 @@ impl Community {
|
||||
SubscriptionKey::of(&self.state)
|
||||
}
|
||||
|
||||
/// A public channel derives its plane from the community root; a private one uses its granted key.
|
||||
fn channel_secret(&self, channel: &ChannelId) -> Option<(Epoch, [u8; 32])> {
|
||||
let held = self
|
||||
.state
|
||||
.channels
|
||||
.iter()
|
||||
.find(|held| held.id == *channel)?;
|
||||
|
||||
if held.private {
|
||||
return held.key.map(|key| (held.epoch, key));
|
||||
}
|
||||
|
||||
Some((held.epoch, self.state.community_root))
|
||||
}
|
||||
|
||||
/// Page a channel's history into the local cache, once per channel.
|
||||
pub fn backfill(&self, channel: &ChannelId, cx: &App) -> Task<Result<()>> {
|
||||
let Some((epoch, secret)) = self.channel_secret(channel) else {
|
||||
return Task::ready(Ok(()));
|
||||
};
|
||||
|
||||
let client = NostrRegistry::global(cx).read(cx).client();
|
||||
let channel = *channel;
|
||||
|
||||
cx.background_spawn(async move {
|
||||
if !store::query_rumors(&client, &channel, None, 1)
|
||||
.await?
|
||||
.is_empty()
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
store::backfill(&client, &channel, &[(epoch, secret)], None, MESSAGE_LIMIT).await?;
|
||||
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
/// The channel's timeline, folded from the local cache, oldest first.
|
||||
pub fn messages(&self, channel: &ChannelId, cx: &App) -> Task<Result<Vec<ChatMessage>>> {
|
||||
let client = NostrRegistry::global(cx).read(cx).client();
|
||||
let channel = *channel;
|
||||
let owner = self.state.owner;
|
||||
let community_id = self.state.id;
|
||||
let floors = self.state.floors();
|
||||
let roles = self.control.roles.clone();
|
||||
|
||||
cx.background_spawn(async move {
|
||||
let cached = store::query_rumors(&client, &channel, None, MESSAGE_LIMIT).await?;
|
||||
let mut rumors = Vec::with_capacity(cached.len());
|
||||
|
||||
for rumor in &cached {
|
||||
match cord03::parse_rumor(rumor) {
|
||||
Ok(chat) => rumors.push(chat),
|
||||
Err(error) => {
|
||||
log::warn!("community: skipping an unreadable cached rumor: {error}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut messages =
|
||||
cord03::fold(&rumors, Timestamp::now(), |actor, citation, author| {
|
||||
citation_ok(&owner, &community_id, actor, citation, &floors)
|
||||
&& roles.can_act_on_member(
|
||||
actor,
|
||||
&owner,
|
||||
author,
|
||||
Permissions::MANAGE_MESSAGES,
|
||||
)
|
||||
});
|
||||
|
||||
messages.reverse();
|
||||
|
||||
Ok(messages)
|
||||
})
|
||||
}
|
||||
|
||||
/// Seal a message to the channel plane, cache it, then publish it to the relays.
|
||||
pub fn send(
|
||||
&self,
|
||||
channel: &ChannelId,
|
||||
content: &str,
|
||||
reply_to: Option<ReplyRef>,
|
||||
cx: &App,
|
||||
) -> Option<Task<Result<EventId>>> {
|
||||
let (epoch, secret) = self.channel_secret(channel)?;
|
||||
|
||||
let nostr = NostrRegistry::global(cx);
|
||||
let client = nostr.read(cx).client();
|
||||
let signer = nostr.read(cx).signer();
|
||||
let author = nostr.read(cx).current_user()?;
|
||||
|
||||
let channel = *channel;
|
||||
let relays = self.state.relays.clone();
|
||||
let timer = self
|
||||
.control
|
||||
.community
|
||||
.as_ref()
|
||||
.and_then(|metadata| metadata.message_expiration);
|
||||
let content = content.to_owned();
|
||||
|
||||
Some(cx.background_spawn(async move {
|
||||
let group = channel_group_key(&secret, &channel, epoch)?;
|
||||
let at_ms = Timestamp::now().as_secs().saturating_mul(1000);
|
||||
|
||||
let rumor = cord03::build_message(
|
||||
author,
|
||||
&channel,
|
||||
epoch,
|
||||
&content,
|
||||
reply_to.as_ref(),
|
||||
at_ms,
|
||||
timer,
|
||||
);
|
||||
let (wrap, _) = cord03::seal_rumor(&rumor, &group, &signer, false).await?;
|
||||
|
||||
let (opened, _) = cord03::open(&wrap, &group, &channel, epoch)?;
|
||||
store::cache_rumor(&client, &channel, &opened).await?;
|
||||
|
||||
sync::connect_relays(&client, &relays).await;
|
||||
sync::publish_wrap(&client, &wrap, &relays).await;
|
||||
|
||||
Ok(opened.rumor_id)
|
||||
}))
|
||||
}
|
||||
|
||||
/// Rebuilds the community from the wraps in the local database.
|
||||
pub fn refresh(&mut self, cx: &mut Context<Self>) {
|
||||
if self.refresh_task.is_some() {
|
||||
|
||||
@@ -5,7 +5,7 @@ use concord::CommunityId;
|
||||
use concord::cord01::KIND_WRAP;
|
||||
pub use concord::cord02::CommunityMetadata;
|
||||
use concord::store::CommunityState;
|
||||
use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task};
|
||||
use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, Window};
|
||||
use nostr_sdk::prelude::*;
|
||||
use smallvec::{SmallVec, smallvec};
|
||||
use state::NostrRegistry;
|
||||
@@ -107,6 +107,20 @@ impl CommunityRegistry {
|
||||
self.index.get(id).cloned()
|
||||
}
|
||||
|
||||
/// Ask the workspace to open a community's panel.
|
||||
pub fn emit_community(
|
||||
&mut self,
|
||||
community: &Entity<Community>,
|
||||
window: &mut Window,
|
||||
cx: &mut Context<Self>,
|
||||
) {
|
||||
let id = community.read(cx).id();
|
||||
|
||||
cx.defer_in(window, move |_this, _window, cx| {
|
||||
cx.emit(CommunityEvent::Open(id));
|
||||
});
|
||||
}
|
||||
|
||||
/// 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);
|
||||
|
||||
@@ -142,13 +142,24 @@ where
|
||||
|
||||
/// Best-effort publication of the genesis wraps to the community's relays.
|
||||
async fn publish_wraps(client: &Client, wraps: &[Event], relays: &[RelayUrl]) {
|
||||
for url in relays {
|
||||
if let Err(error) = client.add_relay(url).and_connect().await {
|
||||
log::warn!("community genesis: failed to add relay {url}: {error}");
|
||||
}
|
||||
}
|
||||
connect_relays(client, relays).await;
|
||||
|
||||
for wrap in wraps {
|
||||
publish_wrap(client, wrap, relays).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Bring the community's relays into the pool before anything is sent through them.
|
||||
pub(crate) async fn connect_relays(client: &Client, relays: &[RelayUrl]) {
|
||||
for url in relays {
|
||||
if let Err(error) = client.add_relay(url).and_connect().await {
|
||||
log::warn!("community: failed to add relay {url}: {error}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Best-effort publication of a single wrap to the community's relays.
|
||||
pub(crate) async fn publish_wrap(client: &Client, wrap: &Event, relays: &[RelayUrl]) {
|
||||
let sent = if relays.is_empty() {
|
||||
client.send_event(wrap).broadcast().await
|
||||
} else {
|
||||
@@ -158,12 +169,11 @@ async fn publish_wraps(client: &Client, wraps: &[Event], relays: &[RelayUrl]) {
|
||||
match sent {
|
||||
Ok(output) if output.failed.is_empty() => {}
|
||||
Ok(output) => log::warn!(
|
||||
"community genesis: {} relay(s) rejected {}",
|
||||
"community: {} relay(s) rejected {}",
|
||||
output.failed.len(),
|
||||
wrap.id
|
||||
),
|
||||
Err(error) => log::warn!("community genesis: publishing {} failed: {error}", wrap.id),
|
||||
}
|
||||
Err(error) => log::warn!("community: publishing {} failed: {error}", wrap.id),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -104,12 +104,12 @@ pub fn send(
|
||||
where `timer` is `control.community.message_expiration` and `at_ms` is now in ms;
|
||||
`cord03::seal_rumor(&rumor, &plane, &signer, false)`; then — the order matters —
|
||||
`cord03::open(&wrap, &plane, channel, epoch)` and `store::cache_rumor` *before*
|
||||
`client.send_event(&wrap).to(&state.relays)`, so the author's own row exists
|
||||
whether or not a relay answers. Add the community's relays with `add_relay(..)
|
||||
.and_connect()` first, the way `sync::publish_wraps` does — lifting that loop into
|
||||
a `pub(crate) sync::publish_wrap(client, &wrap, &relays)` keeps one copy. Publish
|
||||
failures only `log::warn!`. `seal_rumor` needs the channel's `GroupKey` from
|
||||
`derive::channel_group_key(secret, channel, epoch)`, and the epoch from
|
||||
publishing, so the author's own row exists whether or not a relay answers. The
|
||||
publish is `pub(crate) sync::connect_relays(client, &relays)` (the
|
||||
`add_relay(..).and_connect()` loop the genesis path already ran) followed by
|
||||
`pub(crate) sync::publish_wrap(client, &wrap, &relays)`, so one copy serves both
|
||||
paths; failures only `log::warn!`. `seal_rumor` needs the channel's `GroupKey`
|
||||
from `derive::channel_group_key(secret, channel, epoch)`, and the epoch from
|
||||
`channel_secret`. Returns `None` without a signer or a held secret, and the rumor
|
||||
id so the panel can reload.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user