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

wip: refactor

Browse Source
This commit is contained in:
reya
2024-12-19 09:46:22 +07:00
parent 407db7a7d5
commit f800a27aef
11 changed files with 328 additions and 239 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/app/Cargo.toml
View File

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

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

@@ -10,13 +10,17 @@ use std::{
sync::{Arc, OnceLock},
time::Duration,
};
use tokio::{sync::mpsc, time::sleep};
use tokio::{
sync::{mpsc, Mutex},
time::sleep,
};
use constants::{ALL_MESSAGES_SUB_ID, APP_NAME, FAKE_SIG, METADATA_DELAY, NEW_MESSAGE_SUB_ID};
use states::{
account::AccountRegistry,
chat::ChatRegistry,
metadata::{MetadataRegistry, Signal},
metadata::MetadataRegistry,
signal::{Signal, SignalRegistry},
};
use views::app::AppView;
@@ -27,6 +31,7 @@ pub mod utils;
pub mod views;
actions!(main_menu, [Quit]);
actions!(app, [ReloadMetadata]);
static CLIENT: OnceLock<Client> = OnceLock::new();
@@ -75,21 +80,15 @@ async fn main() {
// Connect to all relays
_ = client.connect().await;
// Channel for EOSE
// When receive EOSE from relay(s) -> Load all rooms and push it into UI.
let (eose_tx, mut eose_rx) = mpsc::channel::<SubscriptionId>(200);
// Signal
let (signal_tx, mut signal_rx) = mpsc::channel::<Signal>(10000);
let (mta_tx, mut mta_rx) = mpsc::unbounded_channel::<PublicKey>();
// Channel for new message
// Push new message to chat panel or create new chat room if not exist.
let (message_tx, message_rx) = flume::unbounded::<Event>();
let message_rx_clone = message_rx.clone();
// Channel for signal
// Merge all metadata requests into single one.
// Notify to reload element if receive new metadata.
let (signal_tx, mut signal_rx) = mpsc::channel::<Signal>(5000);
let signal_tx_clone = signal_tx.clone();
// Re use sender
let mta_tx_clone = mta_tx.clone();
// Handle notification from Relays
// Send notfiy back to GPUI
tokio::spawn(async move {
let sig = Signature::from_str(FAKE_SIG).unwrap();
let new_message = SubscriptionId::new(NEW_MESSAGE_SUB_ID);
@@ -129,19 +128,19 @@ async fn main() {
// Send event back to channel
if subscription_id == new_message {
if let Err(e) = message_tx.send_async(ev).await {
if let Err(e) = signal_tx.send(Signal::RecvEvent(ev)).await {
println!("Error: {}", e)
}
}
}
}
} else if event.kind == Kind::Metadata {
if let Err(e) = signal_tx.send(Signal::DONE(event.pubkey)).await {
if let Err(e) = signal_tx.send(Signal::RecvMetadata(event.pubkey)).await {
println!("Error: {}", e)
}
}
} else if let RelayMessage::EndOfStoredEvents(subscription_id) = message {
if let Err(e) = eose_tx.send(subscription_id).await {
if let Err(e) = signal_tx.send(Signal::RecvEose(subscription_id)).await {
println!("Error: {}", e)
}
}
@@ -149,6 +148,44 @@ async fn main() {
}
});
// Handle metadata request
// Merge all requests into single subscription
tokio::spawn(async move {
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);
let opts = SubscribeAutoCloseOptions::default()
.filter(FilterOptions::WaitDurationAfterEOSE(Duration::from_secs(2)));
if let Err(e) = client.subscribe(vec![filter], Some(opts)).await {
println!("Error: {}", e);
}
}
}
});
});
App::new()
.with_assets(Assets)
.with_http_client(Arc::new(reqwest_client::ReqwestClient::new()))
@@ -156,9 +193,11 @@ async fn main() {
// Account state
AccountRegistry::set_global(cx);
// Metadata state
MetadataRegistry::set_global(cx, signal_tx_clone);
MetadataRegistry::set_global(cx);
// Chat state
ChatRegistry::set_global(cx, message_rx);
ChatRegistry::set_global(cx);
// Signal state
SignalRegistry::set_global(cx, mta_tx_clone);
// Initialize components
coop_ui::init(cx);
@@ -166,71 +205,56 @@ async fn main() {
// Set quit action
cx.on_action(quit);
/*
cx.spawn(|async_cx| async move {
let mut queue: HashSet<PublicKey> = HashSet::new();
let accounts = get_all_accounts_from_keyring();
while let Some(signal) = signal_rx.recv().await {
match signal {
Signal::REQ(public_key) => {
queue.insert(public_key);
// Automatically Login if only habe 1 account
if let Some(account) = accounts.into_iter().next() {
if let Ok(keys) = get_keys_by_account(account) {
get_client().set_signer(keys).await;
// Wait for METADATA_DELAY
sleep(Duration::from_millis(METADATA_DELAY)).await;
if !queue.is_empty() {
let authors: Vec<PublicKey> = queue.iter().copied().collect();
let total = authors.len();
let filter = Filter::new()
.authors(authors)
.kind(Kind::Metadata)
.limit(total);
let opts = SubscribeAutoCloseOptions::default().filter(
FilterOptions::WaitDurationAfterEOSE(Duration::from_secs(2)),
);
queue.clear();
async_cx
.background_executor()
.spawn(async move {
if let Err(e) =
client.subscribe(vec![filter], Some(opts)).await
{
println!("Error: {}", e);
}
})
.await;
}
}
Signal::DONE(public_key) => {
_ = async_cx.update_global::<MetadataRegistry, _>(|state, _| {
state.seen(public_key);
});
}
_ = async_cx.update_global::<AccountRegistry, _>(|state, _| {
state.set_user(Some(account));
});
}
}
})
.detach();
cx.spawn(|async_cx| async move {
while let Ok(event) = message_rx_clone.recv_async().await {
_ = async_cx.update_global::<ChatRegistry, _>(|state, cx| {
state.push(event, cx);
});
}
})
.detach();
*/
cx.spawn(|async_cx| async move {
let all_messages = SubscriptionId::new(ALL_MESSAGES_SUB_ID);
let mut is_initialized = false;
while let Some(subscription_id) = eose_rx.recv().await {
if subscription_id == all_messages {
_ = async_cx.update_global::<ChatRegistry, _>(|state, cx| {
state.load(cx);
});
while let Some(signal) = signal_rx.recv().await {
match signal {
Signal::RecvEose(id) => {
if id == all_messages {
if !is_initialized {
_ = async_cx.update_global::<ChatRegistry, _>(|state, _| {
state.set_init();
});
is_initialized = true;
} else {
_ = async_cx.update_global::<ChatRegistry, _>(|state, _| {
state.set_reload();
});
}
}
}
Signal::RecvMetadata(public_key) => {
_ = async_cx.update_global::<MetadataRegistry, _>(|state, _cx| {
state.seen(public_key);
})
}
Signal::RecvEvent(event) => {
_ = async_cx.update_global::<ChatRegistry, _>(|state, _| {
state.push(event);
});
}
_ => {}
}
}
})

84
crates/app/src/states/chat.rs
View File

@@ -1,91 +1,35 @@
use flume::Receiver;
use gpui::*;
use itertools::Itertools;
use nostr_sdk::prelude::*;
use crate::get_client;
pub struct ChatRegistry {
chats: Model<Option<Vec<Event>>>,
is_initialized: bool,
// Use for receive new message
pub(crate) receiver: Receiver<Event>,
pub new_messages: Vec<Event>,
pub reload: bool,
pub is_initialized: bool,
}
impl Global for ChatRegistry {}
impl ChatRegistry {
pub fn set_global(cx: &mut AppContext, receiver: Receiver<Event>) {
let chats = cx.new_model(|_| None);
cx.set_global(Self::new(chats, receiver));
pub fn set_global(cx: &mut AppContext) {
cx.set_global(Self::new());
}
pub fn load(&mut self, cx: &mut AppContext) {
let mut async_cx = cx.to_async();
let async_chats = self.chats.clone();
if !self.is_initialized {
self.is_initialized = true;
cx.foreground_executor()
.spawn(async move {
let client = get_client();
let signer = client.signer().await.unwrap();
let public_key = signer.get_public_key().await.unwrap();
let filter = Filter::new()
.kind(Kind::PrivateDirectMessage)
.pubkey(public_key);
let events = async_cx
.background_executor()
.spawn(async move {
if let Ok(events) = client.database().query(vec![filter]).await {
events
.into_iter()
.filter(|ev| ev.pubkey != public_key) // Filter all messages from current user
.unique_by(|ev| ev.pubkey) // Get unique list
.collect::<Vec<_>>()
} else {
Vec::new()
}
})
.await;
async_cx.update_model(&async_chats, |a, b| {
*a = Some(events);
b.notify();
})
})
.detach();
}
pub fn set_init(&mut self) {
self.is_initialized = true;
}
pub fn push(&self, event: Event, cx: &mut AppContext) {
cx.update_model(&self.chats, |a, b| {
if let Some(chats) = a {
if let Some(index) = chats.iter().position(|c| c.pubkey == event.pubkey) {
chats.swap_remove(index);
chats.push(event);
b.notify();
} else {
chats.push(event);
b.notify();
}
}
})
pub fn set_reload(&mut self) {
self.reload = true;
}
pub fn get(&self, cx: &WindowContext) -> Option<Vec<Event>> {
self.chats.read(cx).clone()
pub fn push(&mut self, event: Event) {
self.new_messages.push(event);
}
fn new(chats: Model<Option<Vec<Event>>>, receiver: Receiver<Event>) -> Self {
fn new() -> Self {
Self {
chats,
receiver,
new_messages: Vec::new(),
reload: false,
is_initialized: false,
}
}

21
crates/app/src/states/metadata.rs
View File

@@ -1,25 +1,15 @@
use gpui::*;
use nostr_sdk::prelude::*;
use tokio::sync::mpsc::Sender;
#[derive(Clone)]
pub enum Signal {
/// Send
DONE(PublicKey),
/// Receive
REQ(PublicKey),
}
pub struct MetadataRegistry {
seens: Vec<PublicKey>,
pub reqs: Sender<Signal>,
}
impl Global for MetadataRegistry {}
impl MetadataRegistry {
pub fn set_global(cx: &mut AppContext, reqs: Sender<Signal>) {
cx.set_global(Self::new(reqs));
pub fn set_global(cx: &mut AppContext) {
cx.set_global(Self::new());
}
pub fn contains(&self, public_key: PublicKey) -> bool {
@@ -32,10 +22,7 @@ impl MetadataRegistry {
}
}
fn new(reqs: Sender<Signal>) -> Self {
Self {
seens: Vec::new(),
reqs,
}
fn new() -> Self {
Self { seens: Vec::new() }
}
}

1
crates/app/src/states/mod.rs
View File

@@ -1,3 +1,4 @@
pub mod account;
pub mod chat;
pub mod metadata;
pub mod signal;

32
crates/app/src/states/signal.rs Normal file
View File

@@ -0,0 +1,32 @@
use gpui::*;
use nostr_sdk::prelude::*;
use std::sync::Arc;
use tokio::sync::mpsc::UnboundedSender;
#[derive(Clone)]
pub enum Signal {
/// Request metadata
ReqMetadata(PublicKey),
/// Receive metadata
RecvMetadata(PublicKey),
/// Receive EOSE
RecvEose(SubscriptionId),
/// Receive event
RecvEvent(Event),
}
pub struct SignalRegistry {
pub tx: Arc<UnboundedSender<PublicKey>>,
}
impl Global for SignalRegistry {}
impl SignalRegistry {
pub fn set_global(cx: &mut AppContext, tx: UnboundedSender<PublicKey>) {
cx.set_global(Self::new(tx));
}
fn new(tx: UnboundedSender<PublicKey>) -> Self {
Self { tx: Arc::new(tx) }
}
}

38
crates/app/src/utils.rs
View File

@@ -1,7 +1,10 @@
use chrono::{Local, TimeZone};
use chrono::{Duration, Local, TimeZone};
use keyring::Entry;
use keyring_search::{Limit, List, Search};
use nostr_sdk::prelude::*;
use crate::constants::KEYRING_SERVICE;
pub fn get_all_accounts_from_keyring() -> Vec<PublicKey> {
let search = Search::new().expect("Keyring not working.");
let results = search.by_service("Coop Safe Storage");
@@ -15,6 +18,15 @@ pub fn get_all_accounts_from_keyring() -> Vec<PublicKey> {
accounts
}
pub fn get_keys_by_account(public_key: PublicKey) -> Result<Keys, anyhow::Error> {
let bech32 = public_key.to_bech32()?;
let entry = Entry::new(KEYRING_SERVICE, &bech32)?;
let password = entry.get_password()?;
let keys = Keys::parse(&password)?;
Ok(keys)
}
pub fn show_npub(public_key: PublicKey, len: usize) -> String {
let bech32 = public_key.to_bech32().unwrap_or_default();
let separator = " ... ";
@@ -37,12 +49,28 @@ pub fn ago(time: u64) -> String {
let input_time = Local.timestamp_opt(time as i64, 0).unwrap();
let diff = (now - input_time).num_hours();
if diff == 0 {
"now".to_owned()
} else if diff < 24 {
if diff < 24 {
let duration = now.signed_duration_since(input_time);
format!("{} hours ago", duration.num_hours())
format_duration(duration)
} else {
input_time.format("%b %d").to_string()
}
}
pub fn format_duration(duration: Duration) -> String {
if duration.num_seconds() < 60 {
"now".to_string()
} else if duration.num_minutes() == 1 {
"1m".to_string()
} else if duration.num_minutes() < 60 {
format!("{}m", duration.num_minutes())
} else if duration.num_hours() == 1 {
"1h".to_string()
} else if duration.num_hours() < 24 {
format!("{}h", duration.num_hours())
} else if duration.num_days() == 1 {
"1d".to_string()
} else {
format!("{}d", duration.num_days())
}
}

4
crates/app/src/views/dock/chat/list.rs
View File

@@ -1,7 +1,7 @@
use gpui::*;
use nostr_sdk::prelude::*;
use crate::{get_client, states::chat::ChatRegistry};
use crate::get_client;
pub struct MessageList {
member: PublicKey,
@@ -56,6 +56,7 @@ impl MessageList {
}
pub fn subscribe(&self, cx: &mut ViewContext<Self>) {
/*
let receiver = cx.global::<ChatRegistry>().receiver.clone();
cx.foreground_executor()
@@ -65,6 +66,7 @@ impl MessageList {
}
})
.detach();
*/
}
}

57
crates/app/src/views/dock/inbox/chat.rs
View File

@@ -5,7 +5,7 @@ use prelude::FluentBuilder;
use crate::{
get_client,
states::metadata::{MetadataRegistry, Signal},
states::{metadata::MetadataRegistry, signal::SignalRegistry},
utils::{ago, show_npub},
views::app::AddPanel,
};
@@ -138,66 +138,41 @@ impl RenderOnce for ChatItem {
pub struct Chat {
title: Option<String>,
public_key: PublicKey,
metadata: Model<Option<Metadata>>,
last_seen: Timestamp,
pub(crate) public_key: PublicKey,
}
impl Chat {
pub fn new(event: Event, cx: &mut ViewContext<'_, Self>) -> Self {
let public_key = event.pubkey;
let last_seen = event.created_at;
let title = if let Some(tag) = event.tags.find(TagKind::Title) {
tag.content().map(|s| s.to_string())
} else {
None
};
let metadata = cx.new_model(|_| None);
let async_metadata = metadata.clone();
let mut async_cx = cx.to_async();
// Request metadata
_ = cx.global::<SignalRegistry>().tx.send(public_key);
let client = get_client();
let signal = cx.global::<MetadataRegistry>();
if !signal.contains(public_key) {
cx.foreground_executor()
.spawn(async move {
let query = async_cx
.background_executor()
.spawn(async move { client.database().metadata(public_key).await })
.await;
if let Ok(metadata) = query {
_ = async_cx.update_model(&async_metadata, |a, b| {
*a = metadata;
b.notify();
});
};
})
.detach();
} else {
let reqs = signal.reqs.clone();
cx.foreground_executor()
.spawn(async move {
if let Err(e) = reqs.send(Signal::REQ(public_key)).await {
println!("Error: {}", e)
}
})
.detach();
cx.observe_global::<MetadataRegistry>(|view, cx| {
view.profile(cx);
})
.detach();
};
// Reload when received metadata
cx.observe_global::<MetadataRegistry>(|chat, cx| {
chat.load_metadata(cx);
})
.detach();
Self {
public_key,
last_seen,
metadata,
title: None,
title,
}
}
fn profile(&self, cx: &mut ViewContext<Self>) {
pub fn load_metadata(&mut self, cx: &mut ViewContext<Self>) {
let public_key = self.public_key;
let async_metadata = self.metadata.clone();
let mut async_cx = cx.to_async();

156
crates/app/src/views/dock/inbox/mod.rs
View File

@@ -1,67 +1,148 @@
use chat::Chat;
use coop_ui::{theme::ActiveTheme, v_flex, Collapsible, Icon, IconName, StyledExt};
use coop_ui::{
skeleton::Skeleton, theme::ActiveTheme, v_flex, Collapsible, Icon, IconName, StyledExt,
};
use gpui::*;
use itertools::Itertools;
use nostr_sdk::prelude::*;
use prelude::FluentBuilder;
use std::cmp::Reverse;
use crate::states::{account::AccountRegistry, chat::ChatRegistry};
use crate::{get_client, states::chat::ChatRegistry};
pub mod chat;
pub struct Inbox {
label: SharedString,
chats: Model<Option<Vec<View<Chat>>>>,
events: Model<Option<Vec<Event>>>,
chats: Model<Vec<View<Chat>>>,
is_loading: bool,
is_fetching: bool,
is_collapsed: bool,
}
impl Inbox {
pub fn new(cx: &mut ViewContext<'_, Self>) -> Self {
let chats = cx.new_model(|_| None);
let chats = cx.new_model(|_| Vec::new());
let events = cx.new_model(|_| None);
cx.observe_global::<ChatRegistry>(|inbox, cx| {
inbox.load(cx);
let state = cx.global::<ChatRegistry>();
if state.reload || (state.is_initialized && state.new_messages.is_empty()) {
inbox.load(cx);
} else {
let new_messages = state.new_messages.clone();
for message in new_messages.into_iter() {
cx.update_model(&inbox.events, |model, b| {
if let Some(events) = model {
if !events.iter().any(|ev| ev.pubkey == message.pubkey) {
events.push(message);
b.notify();
}
}
});
}
}
})
.detach();
cx.observe(&events, |inbox, model, cx| {
// Show fetching indicator
inbox.set_fetching(cx);
let events: Option<Vec<Event>> = model.read(cx).clone();
if let Some(events) = events {
let views = inbox.chats.read(cx);
let public_keys: Vec<PublicKey> =
views.iter().map(|v| v.read(cx).public_key).collect();
for event in events
.into_iter()
.sorted_by_key(|ev| Reverse(ev.created_at))
{
if !public_keys.contains(&event.pubkey) {
let view = cx.new_view(|cx| Chat::new(event, cx));
cx.update_model(&inbox.chats, |a, b| {
a.push(view);
b.notify();
});
}
}
// Hide fetching indicator
inbox.set_fetching(cx);
}
})
.detach();
cx.observe_new_views::<Chat>(|chat, cx| {
chat.load_metadata(cx);
})
.detach();
Self {
events,
chats,
label: "Inbox".into(),
is_loading: true,
is_fetching: false,
is_collapsed: false,
}
}
fn load(&mut self, cx: &mut ViewContext<Self>) {
// Stop loading indicator;
pub fn load(&mut self, cx: &mut ViewContext<Self>) {
// Hide loading indicator
self.set_loading(cx);
// Read global chat registry
let events = cx.global::<ChatRegistry>().get(cx);
let current_user = cx.global::<AccountRegistry>().get();
let async_events = self.events.clone();
let mut async_cx = cx.to_async();
if let Some(public_key) = current_user {
if let Some(events) = events {
let chats: Vec<View<Chat>> = events
.into_iter()
.filter(|ev| ev.pubkey != public_key)
.sorted_by_key(|ev| Reverse(ev.created_at))
.map(|ev| cx.new_view(|cx| Chat::new(ev, cx)))
.collect();
cx.foreground_executor()
.spawn(async move {
let client = get_client();
let signer = client.signer().await.unwrap();
let public_key = signer.get_public_key().await.unwrap();
cx.update_model(&self.chats, |a, b| {
*a = Some(chats);
let filter = Filter::new()
.kind(Kind::PrivateDirectMessage)
.pubkey(public_key);
let events = async_cx
.background_executor()
.spawn(async move {
if let Ok(events) = client.database().query(vec![filter]).await {
events
.into_iter()
.filter(|ev| ev.pubkey != public_key) // Filter all messages from current user
.unique_by(|ev| ev.pubkey)
.collect::<Vec<_>>()
} else {
Vec::new()
}
})
.await;
async_cx.update_model(&async_events, |a, b| {
*a = Some(events);
b.notify();
});
}
}
})
})
.detach();
}
fn set_loading(&mut self, cx: &mut ViewContext<Self>) {
self.is_loading = false;
cx.notify();
}
fn set_fetching(&mut self, cx: &mut ViewContext<Self>) {
self.is_fetching = !self.is_fetching;
cx.notify();
}
}
impl Collapsible for Inbox {
@@ -78,14 +159,31 @@ impl Collapsible for Inbox {
impl Render for Inbox {
fn render(&mut self, cx: &mut ViewContext<Self>) -> impl IntoElement {
let mut content = div();
let chats = self.chats.read(cx);
if let Some(chats) = self.chats.read(cx).as_ref() {
content = content.children(chats.clone())
if self.is_loading {
content = content.children((0..5).map(|_| {
div()
.h_8()
.px_1()
.flex()
.items_center()
.gap_2()
.child(Skeleton::new().flex_shrink_0().size_6().rounded_full())
.child(Skeleton::new().w_20().h_3().rounded_sm())
}))
} else {
match self.is_loading {
true => content = content.child("Loading..."),
false => content = content.child("Empty"),
}
content = content
.children(chats.clone())
.when(self.is_fetching, |this| {
this.h_8()
.px_1()
.flex()
.items_center()
.gap_2()
.child(Skeleton::new().flex_shrink_0().size_6().rounded_full())
.child(Skeleton::new().w_20().h_3().rounded_sm())
});
}
v_flex()