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

update

Browse Source
This commit is contained in:
reya
2026-10-25 18:23:18 +07:00
parent 4b4a99c464
commit 29aa7f05aa
3 changed files with 194 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/chat/src/lib.rs
+156 -67
View File
@@ -11,12 +11,12 @@ use gpui::{
App, AppContext, Context, Entity, EventEmitter, Global, SharedString, Subscription, Task,
WeakEntity,
};
use instant::Duration;
use instant::{Duration, Instant};
use nip4e::{DeviceEvent, DeviceRegistry};
use nostr_sdk::prelude::*;
use person::PersonRegistry;
use smallvec::{SmallVec, smallvec};
use state::{DEVICE_GIFTWRAP, NostrRegistry, USER_GIFTWRAP};
use state::{DEVICE_GIFTWRAP, NostrRegistry, USER_GIFTWRAP, UniversalSigner};
mod message;
mod room;
@@ -29,7 +29,10 @@ pub use state::FileAttachment;
use crate::search::MessageSearchIndex;
/// A static keypair used only for signing locally-cached rumor events.
/// Upper bound on signals applied to the UI in a single update
const MAX_SIGNAL_BATCH: usize = 256;
/// Keypair used only for identifying locally-cached rumor events.
static LOCAL_KEYS: LazyLock<Keys> = LazyLock::new(Keys::generate);
pub fn init(cx: &mut App) {
@@ -74,6 +77,13 @@ impl Signal {
}
}
/// Which signer to attempt when unwrapping a gift wrap
enum UnwrapRoute {
User,
Device,
Both,
}
/// Chat Registry
#[derive(Debug)]
pub struct ChatRegistry {
@@ -216,32 +226,34 @@ impl ChatRegistry {
let tx = self.signal_tx.clone();
let rx = self.signal_rx.clone();
let executor = cx.background_executor().clone();
self.notification_listener = Some(cx.background_spawn(async move {
let mut notifications = client.notifications();
let mut processed_events = HashSet::new();
const MAX_PROCESSED: usize = 10_000;
let mut notifications = client.notifications();
let mut processed_events = HashSet::new();
let mut user_key = signer.get_public_key_async().await.ok();
let device_key = match device_signer.as_ref() {
Some(device) => device.get_public_key_async().await.ok(),
None => None,
};
let mut last_yield = Instant::now();
while let Some(notification) = notifications.next().await {
if last_yield.elapsed() >= Duration::from_millis(8) {
executor.timer(Duration::from_millis(0)).await;
last_yield = Instant::now();
}
let ClientNotification::Message { message, relay_url } = notification else {
continue;
};
match *message {
RelayMessage::Event {
subscription_id,
event,
..
} => {
let chat_sub = subscription_id.as_str() != sub_id1.as_str();
let device_sub = subscription_id.as_str() != sub_id2.as_str();
// Concord wraps are also kind 1059.
//
// Only the two gift wrap subscriptions carry NIP-59 wraps for this account.
if event.kind == Kind::GiftWrap && chat_sub && device_sub {
continue;
}
RelayMessage::Event { event, .. } => {
// Prune the dedup set before it grows unbounded
if processed_events.len() >= MAX_PROCESSED {
processed_events.clear();
@@ -253,8 +265,10 @@ impl ChatRegistry {
// Handle msg relays event to determine when the app is ready to subscribe
if event.kind == Kind::InboxRelays {
let current_user = signer.get_public_key_async().await?;
if event.pubkey == current_user {
if user_key.is_none() {
user_key = signer.get_public_key_async().await.ok();
}
if Some(event.pubkey) == user_key {
tx.send_async(Signal::InboxReady).await?;
}
}
@@ -270,15 +284,39 @@ impl ChatRegistry {
seen.entry(event.id).or_default().insert(relay_url);
}
// Route to the signer matching the recipient tag
// so a wrap is never decrypted twice.
let recipient = event.tags.public_keys().next();
let route = match recipient {
Some(key) if Some(key) == device_key => UnwrapRoute::Device,
Some(key) if Some(key) == user_key => UnwrapRoute::User,
Some(_) if user_key.is_some() && device_key.is_some() => continue,
_ => UnwrapRoute::Both,
};
// Extract the rumor from the gift wrap event
match extract_rumor(
let result = match route {
UnwrapRoute::User => {
extract_rumor(&client, &signer, None, event.as_ref()).await
}
UnwrapRoute::Device => match device_signer.as_ref() {
Some(device) => {
extract_rumor(&client, device, None, event.as_ref()).await
}
None => continue,
},
UnwrapRoute::Both => {
extract_rumor(
&client,
&signer,
device_signer.as_ref(),
event.as_ref(),
)
.await
{
}
};
match result {
Ok(rumor) => {
let Some(rumor_id) = rumor.id else {
log::error!("Rumor missing id after ensure_id");
@@ -317,35 +355,56 @@ impl ChatRegistry {
Ok(())
}));
let tracking = self.tracking.clone();
self.signal_consumer = Some(cx.spawn(async move |this, cx| {
while let Ok(message) = rx.recv_async().await {
match message {
Signal::Message(message) => {
this.update(cx, |this, cx| {
this.new_message(message, cx);
})?;
while let Ok(first) = rx.recv_async().await {
// Batch whatever has already arrived so the UI updates once per burst
let mut messages = Vec::new();
let mut errors = Vec::new();
let mut inbox_ready = false;
let mut eose = false;
let mut pending = Some(first);
while let Some(signal) = pending.take() {
match signal {
Signal::Message(message) => messages.push(message),
Signal::InboxReady => inbox_ready = true,
Signal::Eose => eose = true,
Signal::Error(failed) => errors.push(failed),
}
Signal::InboxReady => {
if messages.len() + errors.len() >= MAX_SIGNAL_BATCH {
break;
}
pending = rx.try_recv().ok();
}
if inbox_ready {
this.update(cx, |this, cx| {
this.get_messages(cx);
})?;
}
Signal::Eose => {
this.update(cx, |this, _cx| {
this.tracking.store(false, Ordering::Release);
})?;
if !messages.is_empty() {
this.update(cx, |this, cx| {
this.new_messages(messages, cx);
})?;
}
if !errors.is_empty() {
trash.update(cx, |trash, cx| {
trash.extend(errors);
cx.notify();
})?;
}
if eose && tracking.swap(false, Ordering::AcqRel) {
this.update(cx, |this, cx| {
this.get_rooms(cx);
})?;
}
Signal::Error(failed) => {
trash.update(cx, |this, cx| {
this.insert(failed);
cx.notify();
})?;
}
};
}
Ok(())
@@ -669,6 +728,7 @@ impl ChatRegistry {
self.rooms.clear();
self.room_index.clear();
self.search.write().unwrap().clear();
self.tracking.store(true, Ordering::Release);
self.trash.update(cx, |this, cx| {
this.clear();
cx.notify();
@@ -748,6 +808,7 @@ impl ChatRegistry {
let client = nostr.read(cx).client();
let signer = nostr.read(cx).signer();
let search = self.search.clone();
let executor = cx.background_executor().clone();
cx.background_spawn(async move {
let public_key = signer.get_public_key_async().await?;
@@ -775,7 +836,15 @@ impl ChatRegistry {
let events = client.database().query(filter).await?;
let mut grouped: HashMap<u64, Vec<UnsignedEvent>> = HashMap::new();
// Keep the single-threaded web event loop responsive during large rescans
let mut last_yield = Instant::now();
for raw in events.into_iter() {
if last_yield.elapsed() >= Duration::from_millis(8) {
executor.timer(Duration::from_millis(0)).await;
last_yield = Instant::now();
}
if let Ok(rumor) = UnsignedEvent::from_json(&raw.content)
&& rumor.tags.public_keys().next().is_some()
{
@@ -811,12 +880,17 @@ impl ChatRegistry {
})
}
/// Parse a nostr event into a message and push it to the belonging room
/// Parse nostr events into messages and push them to the belonging rooms
///
/// - If the room doesn't exist, it will be created.
/// - Updates room ordering based on the most recent messages.
pub(crate) fn new_message(&mut self, message: NewMessage, cx: &mut Context<Self>) {
self.search.write().unwrap().insert(&message.rumor);
/// - If a room doesn't exist, it will be created.
/// - Room ordering is updated once after the whole batch.
pub(crate) fn new_messages(&mut self, messages: Vec<NewMessage>, cx: &mut Context<Self>) {
{
let mut search = self.search.write().unwrap();
for message in &messages {
search.insert(&message.rumor);
}
}
let nostr = NostrRegistry::global(cx);
@@ -824,17 +898,18 @@ impl ChatRegistry {
return;
};
for message in messages {
match self.room_index.get(&message.room).cloned() {
Some(room) => {
room.update(cx, |this, cx| {
if this.kind == RoomKind::Request
&& (message.rumor.pubkey == public_key || this.auto_trusted(public_key, cx))
&& (message.rumor.pubkey == public_key
|| this.auto_trusted(public_key, cx))
{
this.set_ongoing(cx);
}
this.push_message(message, cx);
});
self.sort(cx);
}
None => {
// Push the new room to the front of the list
@@ -842,19 +917,23 @@ impl ChatRegistry {
}
}
}
self.sort(cx);
}
/// Parse a nostr event into a message and push it to the belonging room
pub(crate) fn new_message(&mut self, message: NewMessage, cx: &mut Context<Self>) {
self.new_messages(vec![message], cx);
}
}
/// Unwraps a gift-wrapped event and processes its contents.
async fn extract_rumor<S, D>(
async fn extract_rumor(
client: &Client,
signer: &S,
device_signer: Option<&D>,
signer: &UniversalSigner,
device_signer: Option<&UniversalSigner>,
gift_wrap: &Event,
) -> Result<UnsignedEvent, Error>
where
S: AsyncNip44 + ?Sized,
D: AsyncNip44 + ?Sized,
{
) -> Result<UnsignedEvent, Error> {
// Try to get cached rumor first
if let Ok(rumor) = get_rumor(client, gift_wrap.id).await {
return Ok(rumor);
@@ -888,10 +967,10 @@ where
}
/// Attempts to unwrap a gift wrap event with a given signer.
async fn try_unwrap_with<S>(signer: &S, gift_wrap: &Event) -> Result<UnwrappedGift, Error>
where
S: AsyncNip44 + ?Sized,
{
async fn try_unwrap_with(
signer: &UniversalSigner,
gift_wrap: &Event,
) -> Result<UnwrappedGift, Error> {
// Get the sealed event
let seal = signer
.nip44_decrypt_async(&gift_wrap.pubkey, &gift_wrap.content)
@@ -926,17 +1005,27 @@ where
async fn set_rumor(client: &Client, id: EventId, rumor: &UnsignedEvent) -> Result<(), Error> {
let room_id = rumor.uniq_id().to_string();
let tags = vec![
let tags = Tags::from_list(vec![
Tag::identifier(id),
Tag::public_key(rumor.pubkey),
Tag::custom("r", [room_id]),
Tag::custom("k", [rumor.kind.to_string()]),
];
]);
let event = EventBuilder::new(Kind::ApplicationSpecificData, rumor.as_json())
.tags(tags)
.finalize_async(&*LOCAL_KEYS)
.await?;
let public_key = LOCAL_KEYS.public_key();
let created_at = Timestamp::now();
let kind = Kind::ApplicationSpecificData;
let content = rumor.as_json();
let event = Event::new(
EventId::compute(&public_key, &created_at, &kind, &tags, &content),
public_key,
created_at,
kind,
tags,
content,
Signature::from_byte_array([0u8; 64]),
);
client.database().save_event(&event).await?;
crates/state/src/lib.rs
+2 -8
View File
@@ -477,16 +477,13 @@ impl NostrRegistry {
cx.notify();
}
/// Connect the browser extension signer (NIP-07), remembering the choice so
/// reloads reconnect automatically
/// Connect the browser extension signer (NIP-07).
#[cfg(target_arch = "wasm32")]
pub fn connect_extension(&mut self, cx: &mut Context<Self>) {
let signer = match BrowserSigner::new() {
Ok(signer) => signer,
Err(e) => {
cx.emit(StateEvent::error(format!(
"No Nostr extension found: {e}. Install a NIP-07 extension like Alby or nos2x."
)));
cx.emit(StateEvent::error(format!("No Nostr extension found: {e}.")));
cx.emit(StateEvent::NoSigner);
return;
}
@@ -495,7 +492,6 @@ impl NostrRegistry {
self.tasks.push(cx.spawn(async move |this, cx| {
match signer.get_public_key_async().await {
Ok(public_key) => {
// Remember the choice so reloads reconnect automatically
this.update(cx, |this, cx| {
credentials::write(cx, "extension", b"extension").detach_and_log_err(cx);
this.signer.swap_inner(signer);
@@ -507,8 +503,6 @@ impl NostrRegistry {
})?;
}
Err(e) => {
// Drop the remembered choice so the next visit does not
// retry a signer that is no longer answering
this.update(cx, |_, cx| {
credentials::delete(cx).detach_and_log_err(cx);
cx.emit(StateEvent::error(e.to_string()));
web/src/lib.rs
+9 -17
View File
@@ -1,5 +1,8 @@
#![cfg(target_arch = "wasm32")]
use std::cell::RefCell;
use std::rc::Rc;
use std::sync::Arc;
use gpui::*;
use universal_time::{Instant, MonotonicClock, SystemTime, WallClock, define_time_provider};
@@ -42,36 +45,29 @@ pub fn run() -> Result<(), JsValue> {
// Sets the panic hook and GPUI's console logger
gpui_platform::web_init();
// A single-threaded platform that draws emoji and CJK graphemes missing
// from the bundled fonts with the browser's own fonts, plus a fetch-based
// HTTP client so remote images and content work like on desktop
let platform = std::rc::Rc::new(gpui_web::WebPlatform::new_with_backend_and_font_fallback(
false,
gpui_web::WebBackendPreference::Auto,
gpui_web::CanvasFontFallback::EmojiAndCjk,
));
let http_client = std::sync::Arc::new(platform.fetch_http_client());
let http_client = Arc::new(platform.fetch_http_client());
let app = Application::with_platform(platform)
.with_http_client(http_client)
.with_assets(assets::CombinedAssets);
APPLICATION.with(|application| {
*application.borrow_mut() = Some(app.run_embedded(|cx| {
// Initialize the framework before anything renders
gpui_kit::init(cx);
// The web text system starts with no fonts, and GPUI panics when
// it lays out text in a family that does not exist. Inter is both
// bundled and named on the theme (including both mode configs, so
// theme reloads keep it) before any text is laid out
load_embedded_fonts(cx);
apply_web_fonts(cx);
// Initialize the framework before anything renders
gpui_kit::init(cx);
// Initialize settings
settings::init(cx);
// There is no native theme on the web; apply the persisted
// appearance directly, then the user's theme tuning
apply_appearance(settings::persisted_appearance_mode(), cx);
settings::AppSettings::apply_theme_style(cx);
@@ -121,10 +117,6 @@ fn load_embedded_fonts(cx: &mut App) {
.expect("Failed to load bundled fonts");
}
/// Name Inter on the theme and both mode configs so every text run renders in
/// it and theme reloads keep it. Without an explicit family the theme falls
/// back to `.SystemUIFont`, which the web platform resolves to a family that
/// is not bundled
fn apply_web_fonts(cx: &mut App) {
use gpui_kit::component::Theme;