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

wip: refactor

Browse Source
This commit is contained in:
reya
2025-01-26 08:31:29 +07:00
parent 6a67a79c3f
commit 3c15e74e56
6 changed files with 266 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -1089,6 +1089,7 @@ dependencies = [
"dirs 5.0.1", "dirs 5.0.1",
"gpui", "gpui",
"itertools 0.13.0", "itertools 0.13.0",
"log",
"nostr-sdk", "nostr-sdk",
"registry", "registry",
"reqwest_client", "reqwest_client",

1
crates/app/Cargo.toml
View File

@@ -29,3 +29,4 @@ rust-embed.workspace = true
smol.workspace = true smol.workspace = true
tracing-subscriber = { version = "0.3.18", features = ["fmt"] } tracing-subscriber = { version = "0.3.18", features = ["fmt"] }
log = "0.4"

234
crates/app/src/main.rs
View File

@@ -9,6 +9,7 @@ use gpui::{
}; };
#[cfg(target_os = "linux")] #[cfg(target_os = "linux")]
use gpui::{WindowBackgroundAppearance, WindowDecorations}; use gpui::{WindowBackgroundAppearance, WindowDecorations};
use log::warn;
use nostr_sdk::prelude::*; use nostr_sdk::prelude::*;
use registry::{app::AppRegistry, chat::ChatRegistry, contact::Contact}; use registry::{app::AppRegistry, chat::ChatRegistry, contact::Contact};
use state::{get_client, initialize_client}; use state::{get_client, initialize_client};
@@ -42,7 +43,6 @@ async fn main() {
// Get client // Get client
let client = get_client(); let client = get_client();
let mut notifications = client.notifications();
// Add some bootstrap relays // Add some bootstrap relays
_ = client.add_relay("wss://relay.damus.io/").await; _ = client.add_relay("wss://relay.damus.io/").await;
@@ -57,120 +57,15 @@ async fn main() {
// Signal // Signal
let (signal_tx, mut signal_rx) = mpsc::channel::<Signal>(4096); let (signal_tx, mut signal_rx) = mpsc::channel::<Signal>(4096);
let (mta_tx, mut mta_rx) = mpsc::channel::<PublicKey>(4096); let (mta_tx, mta_rx) = mpsc::channel::<PublicKey>(4096);
// Handle notification from Relays // Handle notifications from relays
// Send notify back to GPUI // Send notify back to GPUI
tokio::spawn(async move { tokio::spawn(async move { handle_notifications(client, signal_tx, mta_tx).await });
let sig = Signature::from_str(FAKE_SIG).unwrap();
let new_message = SubscriptionId::new(NEW_MESSAGE_SUB_ID);
let all_messages = SubscriptionId::new(ALL_MESSAGES_SUB_ID);
while let Ok(notification) = notifications.recv().await {
if let RelayPoolNotification::Message { message, .. } = notification {
if let RelayMessage::Event {
event,
subscription_id,
} = message
{
match event.kind {
Kind::GiftWrap => {
match client.unwrap_gift_wrap(&event).await {
Ok(UnwrappedGift { mut rumor, sender }) => {
// Request metadata
if let Err(e) = mta_tx.send(sender).await {
println!("Send error: {}", e)
};
// Compute event id if not exist
rumor.ensure_id();
if let Some(id) = rumor.id {
let ev = Event::new(
id,
rumor.pubkey,
rumor.created_at,
rumor.kind,
rumor.tags,
rumor.content,
sig,
);
// Save rumor to database to further query
if let Err(e) = client.database().save_event(&ev).await {
println!("Save error: {}", e);
}
// Send event back to channel
if subscription_id == new_message {
if let Err(e) = signal_tx.send(Signal::Event(ev)).await
{
println!("Send error: {}", e)
}
}
}
}
Err(e) => println!("Unwrap error: {}", e),
}
}
Kind::ContactList => {
let public_keys: Vec<PublicKey> =
event.tags.public_keys().copied().collect();
for public_key in public_keys.into_iter() {
if let Err(e) = mta_tx.send(public_key).await {
println!("Send error: {}", e)
};
}
}
_ => {}
}
} else if let RelayMessage::EndOfStoredEvents(subscription_id) = message {
if subscription_id == all_messages {
if let Err(e) = signal_tx.send(Signal::Eose).await {
println!("Send error: {}", e)
}
}
}
}
}
});
// Handle metadata request // Handle metadata request
// Merge all requests into single subscription // Merge all requests into single subscription
tokio::spawn(async move { tokio::spawn(async move { handle_metadata(client, mta_rx).await });
let queue: Arc<Mutex<HashSet<PublicKey>>> = Arc::new(Mutex::new(HashSet::new()));
let queue_clone = queue.clone();
let (tx, mut rx) = mpsc::channel::<PublicKey>(200);
tokio::spawn(async move {
while let Some(public_key) = mta_rx.recv().await {
queue_clone.lock().await.insert(public_key);
_ = tx.send(public_key).await;
}
});
tokio::spawn(async move {
while rx.recv().await.is_some() {
sleep(Duration::from_millis(METADATA_DELAY)).await;
let authors: Vec<PublicKey> = queue.lock().await.drain().collect();
let total = authors.len();
if total > 0 {
let filter = Filter::new()
.authors(authors)
.kind(Kind::Metadata)
.limit(total);
if let Err(e) = client.sync(filter, &SyncOptions::default()).await {
println!("Error: {}", e);
}
}
}
});
});
App::new() App::new()
.with_assets(Assets) .with_assets(Assets)
@@ -189,14 +84,14 @@ async fn main() {
// Spawn a thread to handle Nostr events // Spawn a thread to handle Nostr events
cx.spawn(|async_cx| async move { cx.spawn(|async_cx| async move {
let (tx, rx) = smol::channel::unbounded::<Signal>(); let (tx, rx) = smol::channel::bounded::<Signal>(4096);
async_cx async_cx
.background_executor() .background_executor()
.spawn(async move { .spawn(async move {
while let Some(signal) = signal_rx.recv().await { while let Some(signal) = signal_rx.recv().await {
if let Err(e) = tx.send(signal).await { if let Err(e) = tx.send(signal).await {
println!("Send error: {}", e) warn!("Send error: {}", e)
} }
} }
}) })
@@ -206,7 +101,7 @@ async fn main() {
match signal { match signal {
Signal::Eose => { Signal::Eose => {
_ = async_cx.update_global::<ChatRegistry, _>(|chat, cx| { _ = async_cx.update_global::<ChatRegistry, _>(|chat, cx| {
chat.init(cx); chat.load(cx);
}); });
} }
Signal::Event(event) => { Signal::Event(event) => {
@@ -293,6 +188,119 @@ async fn main() {
}); });
} }
async fn handle_notifications(
client: &Client,
signal_tx: mpsc::Sender<Signal>,
mta_tx: mpsc::Sender<PublicKey>,
) {
let mut notifications = client.notifications();
let sig = Signature::from_str(FAKE_SIG).unwrap();
let new_message = SubscriptionId::new(NEW_MESSAGE_SUB_ID);
let all_messages = SubscriptionId::new(ALL_MESSAGES_SUB_ID);
while let Ok(notification) = notifications.recv().await {
if let RelayPoolNotification::Message { message, .. } = notification {
if let RelayMessage::Event {
event,
subscription_id,
} = message
{
match event.kind {
Kind::GiftWrap => {
match client.unwrap_gift_wrap(&event).await {
Ok(UnwrappedGift { mut rumor, sender }) => {
// Request metadata
if let Err(e) = mta_tx.send(sender).await {
warn!("Send error: {}", e)
};
// Compute event id if not exist
rumor.ensure_id();
if let Some(id) = rumor.id {
let ev = Event::new(
id,
rumor.pubkey,
rumor.created_at,
rumor.kind,
rumor.tags,
rumor.content,
sig,
);
// Save rumor to database to further query
if let Err(e) = client.database().save_event(&ev).await {
warn!("Save error: {}", e);
}
// Send event back to channel
if subscription_id == new_message {
if let Err(e) = signal_tx.send(Signal::Event(ev)).await {
warn!("Send error: {}", e)
}
}
}
}
Err(e) => warn!("Unwrap error: {}", e),
}
}
Kind::ContactList => {
let public_keys: Vec<PublicKey> =
event.tags.public_keys().copied().collect();
for public_key in public_keys.into_iter() {
if let Err(e) = mta_tx.send(public_key).await {
warn!("Send error: {}", e)
};
}
}
_ => {}
}
} else if let RelayMessage::EndOfStoredEvents(subscription_id) = message {
if subscription_id == all_messages {
if let Err(e) = signal_tx.send(Signal::Eose).await {
warn!("Send error: {}", e)
};
}
}
}
}
}
async fn handle_metadata(client: &'static Client, mut mta_rx: mpsc::Receiver<PublicKey>) {
let queue: Arc<Mutex<HashSet<PublicKey>>> = Arc::new(Mutex::new(HashSet::new()));
let queue_clone = Arc::clone(&queue);
let (tx, mut rx) = mpsc::channel::<PublicKey>(200);
tokio::spawn(async move {
while let Some(public_key) = mta_rx.recv().await {
queue_clone.lock().await.insert(public_key);
_ = tx.send(public_key).await;
}
});
tokio::spawn(async move {
while rx.recv().await.is_some() {
sleep(Duration::from_millis(METADATA_DELAY)).await;
let authors: Vec<PublicKey> = queue.lock().await.drain().collect();
let total = authors.len();
if total > 0 {
let filter = Filter::new()
.authors(authors)
.kind(Kind::Metadata)
.limit(total);
if let Err(e) = client.sync(filter, &SyncOptions::default()).await {
warn!("Error: {}", e);
}
}
}
});
}
fn quit(_: &Quit, cx: &mut AppContext) { fn quit(_: &Quit, cx: &mut AppContext) {
cx.quit(); cx.quit();
} }

118
crates/app/src/views/app.rs
View File

@@ -1,11 +1,12 @@
use super::{chat::ChatPanel, onboarding::Onboarding, sidebar::Sidebar, welcome::WelcomePanel}; use super::{chat::ChatPanel, onboarding::Onboarding, sidebar::Sidebar, welcome::WelcomePanel};
use gpui::{ use gpui::{
actions, div, img, impl_internal_actions, px, svg, Axis, Edges, InteractiveElement, actions, div, img, impl_internal_actions, px, svg, Axis, BorrowAppContext, Edges,
IntoElement, ObjectFit, ParentElement, Render, Styled, StyledImage, View, ViewContext, InteractiveElement, IntoElement, ObjectFit, ParentElement, Render, Styled, StyledImage, View,
VisualContext, WeakView, WindowContext, ViewContext, VisualContext, WeakView, WindowContext,
}; };
use registry::{app::AppRegistry, chat::ChatRegistry, contact::Contact}; use registry::{app::AppRegistry, chat::ChatRegistry, contact::Contact};
use serde::Deserialize; use serde::Deserialize;
use state::get_client;
use std::sync::Arc; use std::sync::Arc;
use ui::{ use ui::{
button::{Button, ButtonVariants}, button::{Button, ButtonVariants},
@@ -68,48 +69,6 @@ impl AppView {
AppView { onboarding, dock } AppView { onboarding, dock }
} }
fn on_panel_action(&mut self, action: &AddPanel, cx: &mut ViewContext<Self>) {
match &action.panel {
PanelKind::Room(id) => {
if let Some(weak_room) = cx.global::<ChatRegistry>().room(id, cx) {
if let Some(room) = weak_room.upgrade() {
let panel = Arc::new(ChatPanel::new(room, cx));
self.dock.update(cx, |dock_area, cx| {
dock_area.add_panel(panel, action.position, cx);
});
} else {
cx.push_notification((
NotificationType::Error,
"System error. Cannot open this chat room.",
));
}
}
}
};
}
fn render_account(&self, account: Contact) -> impl IntoElement {
Button::new("account")
.ghost()
.xsmall()
.reverse()
.icon(Icon::new(IconName::ChevronDownSmall))
.child(
img(account.avatar())
.size_5()
.rounded_full()
.object_fit(ObjectFit::Cover),
)
.popup_menu(move |this, _cx| {
this.menu("Profile", Box::new(OpenProfile))
.menu("Contacts", Box::new(OpenContacts))
.menu("Settings", Box::new(OpenSettings))
.separator()
.menu("Change account", Box::new(Logout))
})
}
fn render_dock(dock_area: WeakView<DockArea>, cx: &mut WindowContext) { fn render_dock(dock_area: WeakView<DockArea>, cx: &mut WindowContext) {
let left = DockItem::panel(Arc::new(Sidebar::new(cx))); let left = DockItem::panel(Arc::new(Sidebar::new(cx)));
let center = DockItem::split_with_sizes( let center = DockItem::split_with_sizes(
@@ -140,6 +99,70 @@ impl AppView {
// TODO: support bottom dock? // TODO: support bottom dock?
}); });
} }
fn render_account(&self, account: Contact) -> impl IntoElement {
Button::new("account")
.ghost()
.xsmall()
.reverse()
.icon(Icon::new(IconName::ChevronDownSmall))
.child(
img(account.avatar())
.size_5()
.rounded_full()
.object_fit(ObjectFit::Cover),
)
.popup_menu(move |this, _cx| {
this.menu("Profile", Box::new(OpenProfile))
.menu("Contacts", Box::new(OpenContacts))
.menu("Settings", Box::new(OpenSettings))
.separator()
.menu("Change account", Box::new(Logout))
})
}
fn on_panel_action(&mut self, action: &AddPanel, cx: &mut ViewContext<Self>) {
match &action.panel {
PanelKind::Room(id) => {
if let Some(weak_room) = cx.global::<ChatRegistry>().room(id, cx) {
if let Some(room) = weak_room.upgrade() {
let panel = Arc::new(ChatPanel::new(room, cx));
self.dock.update(cx, |dock_area, cx| {
dock_area.add_panel(panel, action.position, cx);
});
} else {
cx.push_notification((
NotificationType::Error,
"System error. Cannot open this chat room.",
));
}
}
}
};
}
fn on_profile_action(&mut self, _action: &OpenProfile, cx: &mut ViewContext<Self>) {
// TODO
}
fn on_contacts_action(&mut self, _action: &OpenContacts, cx: &mut ViewContext<Self>) {
// TODO
}
fn on_settings_action(&mut self, _action: &OpenSettings, cx: &mut ViewContext<Self>) {
// TODO
}
fn on_logout_action(&mut self, _action: &Logout, cx: &mut ViewContext<Self>) {
cx.update_global::<AppRegistry, _>(|this, cx| {
this.logout(cx);
// Reset nostr client
cx.background_executor()
.spawn(async move { get_client().reset().await })
.detach();
});
}
} }
impl Render for AppView { impl Render for AppView {
@@ -183,7 +206,12 @@ impl Render for AppView {
), ),
) )
.child(self.dock.clone()) .child(self.dock.clone())
// Listener
.on_action(cx.listener(Self::on_panel_action)) .on_action(cx.listener(Self::on_panel_action))
.on_action(cx.listener(Self::on_logout_action))
.on_action(cx.listener(Self::on_profile_action))
.on_action(cx.listener(Self::on_contacts_action))
.on_action(cx.listener(Self::on_settings_action))
} else { } else {
this.child(TitleBar::new()).child(self.onboarding.clone()) this.child(TitleBar::new()).child(self.onboarding.clone())
} }

13
crates/registry/src/app.rs
View File

@@ -15,8 +15,8 @@ impl Global for AppRegistry {}
impl AppRegistry { impl AppRegistry {
pub fn set_global(cx: &mut AppContext) { pub fn set_global(cx: &mut AppContext) {
let is_loading = true;
let user: Model<Option<Contact>> = cx.new_model(|_| None); let user: Model<Option<Contact>> = cx.new_model(|_| None);
let is_loading = true;
cx.observe(&user, |this, cx| { cx.observe(&user, |this, cx| {
if let Some(contact) = this.read(cx).as_ref() { if let Some(contact) = this.read(cx).as_ref() {
@@ -80,6 +80,10 @@ impl AppRegistry {
self.user.downgrade() self.user.downgrade()
} }
pub fn current_user(&self, cx: &WindowContext) -> Option<Contact> {
self.user.read(cx).clone()
}
pub fn set_user(&mut self, contact: Contact, cx: &mut AppContext) { pub fn set_user(&mut self, contact: Contact, cx: &mut AppContext) {
self.user.update(cx, |this, cx| { self.user.update(cx, |this, cx| {
*this = Some(contact); *this = Some(contact);
@@ -89,7 +93,10 @@ impl AppRegistry {
self.is_loading = false; self.is_loading = false;
} }
pub fn current_user(&self, cx: &WindowContext) -> Option<Contact> { pub fn logout(&mut self, cx: &mut AppContext) {
self.user.read(cx).clone() self.user.update(cx, |this, cx| {
*this = None;
cx.notify();
});
} }
} }

123
crates/registry/src/chat.rs
View File

@@ -1,12 +1,12 @@
use crate::room::Room;
use anyhow::Error;
use common::utils::{compare, room_hash}; use common::utils::{compare, room_hash};
use gpui::{AppContext, Context, Global, Model, WeakModel}; use gpui::{AppContext, AsyncAppContext, Context, Global, Model, ModelContext, Task, WeakModel};
use itertools::Itertools; use itertools::Itertools;
use nostr_sdk::prelude::*; use nostr_sdk::prelude::*;
use state::get_client; use state::get_client;
use std::cmp::Reverse; use std::cmp::Reverse;
use crate::room::Room;
pub struct Inbox { pub struct Inbox {
pub rooms: Vec<Model<Room>>, pub rooms: Vec<Model<Room>>,
pub is_loading: bool, pub is_loading: bool,
@@ -19,6 +19,37 @@ impl Inbox {
is_loading: true, is_loading: true,
} }
} }
pub fn current_rooms(&self, cx: &ModelContext<Self>) -> Vec<u64> {
self.rooms.iter().map(|room| room.read(cx).id).collect()
}
pub fn load(&mut self, cx: AsyncAppContext) -> Task<Result<Vec<Event>, Error>> {
cx.background_executor().spawn(async move {
let client = get_client();
let signer = client.signer().await?;
let public_key = signer.get_public_key().await?;
let filter = Filter::new()
.kind(Kind::PrivateDirectMessage)
.author(public_key);
// Get all DM events from database
let events = client.database().query(vec![filter]).await?;
// Filter result
// - Get unique rooms only
// - Sorted by created_at
let result = events
.into_iter()
.filter(|ev| ev.tags.public_keys().peekable().peek().is_some())
.unique_by(|ev| room_hash(&ev.tags))
.sorted_by_key(|ev| Reverse(ev.created_at))
.collect::<Vec<_>>();
Ok(result)
})
}
} }
impl Default for Inbox { impl Default for Inbox {
@@ -77,72 +108,38 @@ impl ChatRegistry {
cx.set_global(Self { inbox }); cx.set_global(Self { inbox });
} }
pub fn init(&mut self, cx: &mut AppContext) { pub fn load(&mut self, cx: &mut AppContext) {
let mut async_cx = cx.to_async(); self.inbox.update(cx, |this, cx| {
let async_inbox = self.inbox.clone(); let task = this.load(cx.to_async());
// Get all current room's id cx.spawn(|this, mut async_cx| async move {
let hashes: Vec<u64> = self if let Some(inbox) = this.upgrade() {
.inbox if let Ok(events) = task.await {
.read(cx) _ = async_cx.update_model(&inbox, |this, cx| {
.rooms let current_rooms = this.current_rooms(cx);
.iter() let items: Vec<Model<Room>> = events
.map(|room| room.read(cx).id) .into_iter()
.collect(); .filter_map(|ev| {
let id = room_hash(&ev.tags);
// Filter all seen events
if !current_rooms.iter().any(|h| h == &id) {
Some(cx.new_model(|_| Room::parse(&ev)))
} else {
None
}
})
.collect();
cx.foreground_executor() this.rooms.extend(items);
.spawn(async move { this.is_loading = false;
let client = get_client();
let query: anyhow::Result<Vec<Event>, anyhow::Error> = async_cx
.background_executor()
.spawn(async move {
let signer = client.signer().await?;
let public_key = signer.get_public_key().await?;
let filter = Filter::new() cx.notify();
.kind(Kind::PrivateDirectMessage) });
.author(public_key); }
// Get all DM events from database
let events = client.database().query(vec![filter]).await?;
// Filter result
// - Only unique rooms
// - Sorted by created_at
let result = events
.into_iter()
.filter(|ev| ev.tags.public_keys().peekable().peek().is_some())
.unique_by(|ev| room_hash(&ev.tags))
.sorted_by_key(|ev| Reverse(ev.created_at))
.collect::<Vec<_>>();
Ok(result)
})
.await;
if let Ok(events) = query {
_ = async_cx.update_model(&async_inbox, |model, cx| {
let items: Vec<Model<Room>> = events
.into_iter()
.filter_map(|ev| {
let id = room_hash(&ev.tags);
// Filter all seen events
if !hashes.iter().any(|h| h == &id) {
Some(cx.new_model(|_| Room::parse(&ev)))
} else {
None
}
})
.collect();
model.rooms.extend(items);
model.is_loading = false;
cx.notify();
});
} }
}) })
.detach(); .detach();
});
} }
pub fn inbox(&self) -> WeakModel<Inbox> { pub fn inbox(&self) -> WeakModel<Inbox> {