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

wip

Browse Source
This commit is contained in:
reya
2026-01-09 15:15:35 +07:00
parent 68015ead1c
commit c7c10941be
3 changed files with 232 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
crates/chat/src/lib.rs
View File

@@ -10,7 +10,9 @@ use common::EventUtils;
use flume::Sender;
use fuzzy_matcher::skim::SkimMatcherV2;
use fuzzy_matcher::FuzzyMatcher;
use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Task, WeakEntity};
use gpui::{
App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, WeakEntity,
};
use nostr_sdk::prelude::*;
use settings::AppSettings;
use smallvec::{smallvec, SmallVec};
@@ -61,8 +63,14 @@ pub struct ChatRegistry {
/// Loading status of the registry
loading: bool,
/// Handle notifications asynchronous task
notifications: Option<Task<Result<(), Error>>>,
/// Tasks for asynchronous operations
_tasks: SmallVec<[Task<()>; 3]>,
tasks: SmallVec<[Task<()>; 3]>,
/// Subscriptions
_subscriptions: SmallVec<[Subscription; 1]>,
}
impl EventEmitter<ChatEvent> for ChatRegistry {}
@@ -82,6 +90,7 @@ impl ChatRegistry {
fn new(cx: &mut Context<Self>) -> Self {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let device_signer = nostr.read(cx).device_signer();
// A flag to indicate if the registry is loading
let status = Arc::new(AtomicBool::new(true));
@@ -90,15 +99,46 @@ impl ChatRegistry {
let (tx, rx) = flume::bounded::<NostrEvent>(2048);
let mut tasks = smallvec![];
let mut subscriptions = smallvec![];
tasks.push(
// Handle nostr notifications
let notifications =
Some(
cx.background_spawn({
let client = client.clone();
let status = Arc::clone(&status);
let device_signer = device_signer.read(cx).clone();
let loading = Arc::clone(&status);
let tx = tx.clone();
async move { Self::handle_notifications(&client, &status, &tx).await }
async move {
Self::handle_notifications(&client, &device_signer, &loading, &tx).await
}
}),
);
subscriptions.push(
// Observe the device signer state
cx.observe(&device_signer, {
let loading = Arc::clone(&status);
let tx = tx.clone();
move |this, state, cx| {
if state.read(cx).is_some() {
this.notifications = Some(cx.background_spawn({
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let device_signer = state.read(cx).clone();
let loading = Arc::clone(&loading);
let tx = tx.clone();
async move {
Self::handle_notifications(&client, &device_signer, &loading, &tx)
.await
}
}))
}
}
}),
);
@@ -141,15 +181,18 @@ impl ChatRegistry {
Self {
rooms: vec![],
loading: true,
_tasks: tasks,
notifications,
tasks,
_subscriptions: subscriptions,
}
}
async fn handle_notifications(
client: &Client,
device_signer: &Option<Arc<dyn NostrSigner>>,
loading: &Arc<AtomicBool>,
tx: &Sender<NostrEvent>,
) {
) -> Result<(), Error> {
let initialized_at = Timestamp::now();
let subscription_id = SubscriptionId::new(GIFTWRAP_SUBSCRIPTION);
@@ -175,21 +218,21 @@ impl ChatRegistry {
}
// Extract the rumor from the gift wrap event
match Self::extract_rumor(client, event.as_ref()).await {
match Self::extract_rumor(client, device_signer, event.as_ref()).await {
Ok(rumor) => match rumor.created_at >= initialized_at {
true => {
// Check if the event is sent by coop
let sent_by_coop = {
let tracker = tracker().read().await;
tracker.is_sent_by_coop(&event.id)
};
// No need to emit if sent by coop
// the event is already emitted
if !sent_by_coop {
let new_message = NewMessage::new(event.id, rumor);
let signal = NostrEvent::Message(new_message);
if let Err(e) = tx.send_async(signal).await {
log::error!("Failed to send signal: {}", e);
}
tx.send_async(signal).await.ok();
}
}
false => {
@@ -197,20 +240,20 @@ impl ChatRegistry {
}
},
Err(e) => {
log::warn!("Failed to unwrap gift wrap event: {}", e);
log::warn!("Failed to unwrap: {e}");
}
}
}
RelayMessage::EndOfStoredEvents(id) => {
if id.as_ref() == &subscription_id {
if let Err(e) = tx.send_async(NostrEvent::Eose).await {
log::error!("Failed to send signal: {}", e);
}
tx.send_async(NostrEvent::Eose).await.ok();
}
}
_ => {}
}
}
Ok(())
}
async fn unwrapping_status(client: &Client, status: &Arc<AtomicBool>, tx: &Sender<NostrEvent>) {
@@ -381,7 +424,7 @@ impl ChatRegistry {
pub fn get_rooms(&mut self, cx: &mut Context<Self>) {
let task = self.create_get_rooms_task(cx);
self._tasks.push(
self.tasks.push(
// Run and finished in the background
cx.spawn(async move |this, cx| {
match task.await {
@@ -542,14 +585,18 @@ impl ChatRegistry {
}
// Unwraps a gift-wrapped event and processes its contents.
async fn extract_rumor(client: &Client, gift_wrap: &Event) -> Result<UnsignedEvent, Error> {
async fn extract_rumor(
client: &Client,
device_signer: &Option<Arc<dyn NostrSigner>>,
gift_wrap: &Event,
) -> Result<UnsignedEvent, Error> {
// Try to get cached rumor first
if let Ok(event) = Self::get_rumor(client, gift_wrap.id).await {
return Ok(event);
}
// Try to unwrap with the available signer
let unwrapped = Self::try_unwrap(client, gift_wrap).await?;
let unwrapped = Self::try_unwrap(client, device_signer, gift_wrap).await?;
let mut rumor_unsigned = unwrapped.rumor;
// Generate event id for the rumor if it doesn't have one
@@ -562,7 +609,28 @@ impl ChatRegistry {
}
// Helper method to try unwrapping with different signers
async fn try_unwrap(client: &Client, gift_wrap: &Event) -> Result<UnwrappedGift, Error> {
async fn try_unwrap(
client: &Client,
device_signer: &Option<Arc<dyn NostrSigner>>,
gift_wrap: &Event,
) -> Result<UnwrappedGift, Error> {
if let Some(signer) = device_signer.as_ref() {
let seal = signer
.nip44_decrypt(&gift_wrap.pubkey, &gift_wrap.content)
.await?;
let seal: Event = Event::from_json(seal)?;
seal.verify_with_ctx(&SECP256K1)?;
let rumor = signer.nip44_decrypt(&seal.pubkey, &seal.content).await?;
let rumor = UnsignedEvent::from_json(rumor)?;
return Ok(UnwrappedGift {
sender: seal.pubkey,
rumor,
});
}
let signer = client.signer().await?;
let unwrapped = UnwrappedGift::from_gift_wrap(&signer, gift_wrap).await?;

22
crates/state/src/identity.rs
View File

@@ -1,5 +1,3 @@
use std::sync::Arc;
use nostr_sdk::prelude::*;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
@@ -16,16 +14,12 @@ impl RelayState {
}
}
/// Identity
#[derive(Debug, Clone, Default)]
pub struct Identity {
/// The public key of the account
pub public_key: Option<PublicKey>,
/// Decoupled encryption key
///
/// NIP-4e: https://github.com/nostr-protocol/nips/blob/per-device-keys/4e.md
dekey: Option<Arc<dyn NostrSigner>>,
/// Status of the current user NIP-65 relays
relay_list: RelayState,
@@ -43,7 +37,6 @@ impl Identity {
pub fn new() -> Self {
Self {
public_key: None,
dekey: None,
relay_list: RelayState::default(),
messaging_relays: RelayState::default(),
}
@@ -69,19 +62,6 @@ impl Identity {
self.messaging_relays
}
/// Returns the decoupled encryption key.
pub fn dekey(&self) -> Option<Arc<dyn NostrSigner>> {
self.dekey.clone()
}
/// Sets the decoupled encryption key.
pub fn set_dekey<S>(&mut self, dekey: S)
where
S: NostrSigner + 'static,
{
self.dekey = Some(Arc::new(dekey));
}
/// Force getting the public key of the identity.
///
/// Panics if the public key is not set.

175
crates/state/src/lib.rs
View File

@@ -1,4 +1,5 @@
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use anyhow::{anyhow, Context as AnyhowContext, Error};
@@ -52,6 +53,11 @@ pub struct NostrRegistry {
/// Gossip implementation
gossip: Entity<Gossip>,
/// Device signer
///
/// NIP-4e: https://github.com/nostr-protocol/nips/blob/per-device-keys/4e.md
device_signer: Entity<Option<Arc<dyn NostrSigner>>>,
/// Device state
///
/// NIP-4e: https://github.com/nostr-protocol/nips/blob/per-device-keys/4e.md
@@ -113,6 +119,9 @@ impl NostrRegistry {
// Construct the identity entity
let identity = cx.new(|_| Identity::default());
// Construct the device signer entity
let device_signer = cx.new(|_| None);
let device_state = cx.new(|_| DeviceState::default());
// Channel for communication between nostr and gpui
@@ -133,10 +142,11 @@ impl NostrRegistry {
match state.read(cx).messaging_relays_state() {
RelayState::Initial => {
this.get_profile(cx);
this.get_announcement(cx);
this.get_messaging_relays(cx);
}
RelayState::Set => {
this.get_messages(state.read(cx).dekey(), cx);
this.get_messages(cx);
}
_ => {}
};
@@ -158,7 +168,7 @@ impl NostrRegistry {
tasks.push(
// Update GPUI states
cx.spawn(async move |this, cx| {
cx.spawn(async move |_this, cx| {
while let Ok(event) = rx.recv_async().await {
match event.kind {
Kind::RelayList => {
@@ -173,11 +183,6 @@ impl NostrRegistry {
cx.notify();
})?;
}
Kind::Custom(10044) => {
this.update(cx, |this, cx| {
this.init_dekey(&event, cx);
})?;
}
_ => {}
}
}
@@ -188,10 +193,11 @@ impl NostrRegistry {
Self {
client,
identity,
device_state,
gossip,
app_keys,
identity,
gossip,
device_signer,
device_state,
_subscriptions: subscriptions,
tasks,
}
@@ -240,16 +246,6 @@ impl NostrRegistry {
Kind::InboxRelays => {
tx.send_async(event.into_owned()).await?;
}
Kind::Custom(10044) => {
if let Ok(signer) = client.signer().await {
if let Ok(public_key) = signer.get_public_key().await {
// Only send if the event is from the current user
if public_key == event.pubkey {
tx.send_async(event.into_owned()).await?;
}
}
}
}
_ => {}
}
}
@@ -356,6 +352,11 @@ impl NostrRegistry {
self.identity.clone()
}
/// Get current device signer
pub fn device_signer(&self) -> Entity<Option<Arc<dyn NostrSigner>>> {
self.device_signer.clone()
}
/// Get a relay hint (messaging relay) for a given public key
pub fn relay_hint(&self, public_key: &PublicKey, cx: &App) -> Option<RelayUrl> {
self.gossip
@@ -567,6 +568,58 @@ impl NostrRegistry {
task.detach();
}
/// Get device announcement for current user
fn get_announcement(&mut self, cx: &mut Context<Self>) {
let client = self.client();
let public_key = self.identity().read(cx).public_key();
let write_relays = self.write_relays(&public_key, cx);
let task: Task<Result<Event, Error>> = cx.background_spawn(async move {
let urls = write_relays.await;
// Construct the filter for the device announcement event
let filter = Filter::new()
.kind(Kind::Custom(10044))
.author(public_key)
.limit(1);
let mut stream = client
.stream_events_from(&urls, vec![filter], Duration::from_secs(TIMEOUT))
.await?;
while let Some((_url, res)) = stream.next().await {
match res {
Ok(event) => {
log::info!("Received device announcement event: {event:?}");
return Ok(event);
}
Err(e) => {
log::error!("Failed to receive device announcement event: {e}");
}
}
}
Err(anyhow!("Device announcement not found"))
});
self.tasks.push(cx.spawn(async move |this, cx| {
match task.await {
Ok(event) => {
this.update(cx, |this, cx| {
this.init_device_signer(&event, cx);
})?;
}
Err(_) => {
this.update(cx, |this, cx| {
this.announce_device(cx);
})?;
}
}
Ok(())
}));
}
/// Get messaging relays for current user
fn get_messaging_relays(&mut self, cx: &mut Context<Self>) {
let client = self.client();
@@ -623,11 +676,9 @@ impl NostrRegistry {
}
/// Continuously get gift wrap events for the current user in their messaging relays
fn get_messages<T>(&mut self, dekey: Option<T>, cx: &mut Context<Self>)
where
T: NostrSigner + 'static,
{
fn get_messages(&mut self, cx: &mut Context<Self>) {
let client = self.client();
let device_signer = self.device_signer().read(cx).clone();
let public_key = self.identity().read(cx).public_key();
let messaging_relays = self.messaging_relays(&public_key, cx);
@@ -640,7 +691,7 @@ impl NostrRegistry {
filters.push(Filter::new().kind(Kind::GiftWrap).pubkey(public_key));
// Construct a filter to get dekey messages if available
if let Some(signer) = dekey {
if let Some(signer) = device_signer.as_ref() {
if let Ok(pubkey) = signer.get_public_key().await {
filters.push(Filter::new().kind(Kind::GiftWrap).pubkey(pubkey));
}
@@ -654,12 +705,12 @@ impl NostrRegistry {
}
/// Set the decoupled encryption key for the current user
fn set_dekey<T>(&mut self, dekey: T, cx: &mut Context<Self>)
fn set_device_signer<S>(&mut self, signer: S, cx: &mut Context<Self>)
where
T: NostrSigner + 'static,
S: NostrSigner + 'static,
{
self.identity.update(cx, |this, cx| {
this.set_dekey(dekey);
self.device_signer.update(cx, |this, cx| {
*this = Some(Arc::new(signer));
cx.notify();
});
self.device_state.update(cx, |this, cx| {
@@ -668,11 +719,65 @@ impl NostrRegistry {
});
}
/// Initialize dekey (decoupled encryption key) for the current user
fn init_dekey(&mut self, event: &Event, cx: &mut Context<Self>) {
/// Create a new device signer and announce it
fn announce_device(&mut self, cx: &mut Context<Self>) {
let client = self.client();
let public_key = self.identity().read(cx).public_key();
let write_relays = self.write_relays(&public_key, cx);
let keys = Keys::generate();
let secret = keys.secret_key().to_secret_hex();
let n = keys.public_key();
let task: Task<Result<(), Error>> = cx.background_spawn(async move {
let signer = client.signer().await?;
let urls = write_relays.await;
// Construct an announcement event
let event = EventBuilder::new(Kind::Custom(10044), "")
.tags(vec![
Tag::custom(TagKind::custom("n"), vec![n]),
Tag::client(app_name()),
])
.sign(&signer)
.await?;
// Publish announcement
client.send_event_to(&urls, &event).await?;
// Encrypt the secret key
let encrypted = signer.nip44_encrypt(&public_key, &secret).await?;
// Construct a storage event
let event = EventBuilder::new(Kind::ApplicationSpecificData, encrypted)
.tag(Tag::identifier("coop:device"))
.sign(&signer)
.await?;
// Save storage event to database
//
// Note: never publish to any relays
client.database().save_event(&event).await?;
Ok(())
});
cx.spawn(async move |this, cx| {
if task.await.is_ok() {
this.update(cx, |this, cx| {
this.set_device_signer(keys, cx);
})
.ok();
}
})
.detach();
}
/// Initialize device signer (decoupled encryption key) for the current user
fn init_device_signer(&mut self, event: &Event, cx: &mut Context<Self>) {
let client = self.client();
let announcement = Announcement::from(event);
let dekey = announcement.public_key();
let device_pubkey = announcement.public_key();
let task: Task<Result<Keys, Error>> = cx.background_spawn(async move {
let signer = client.signer().await?;
@@ -689,7 +794,7 @@ impl NostrRegistry {
let secret = SecretKey::parse(&content)?;
let keys = Keys::new(secret);
if keys.public_key() == dekey {
if keys.public_key() == device_pubkey {
Ok(keys)
} else {
Err(anyhow!("Key mismatch"))
@@ -703,7 +808,7 @@ impl NostrRegistry {
match task.await {
Ok(keys) => {
this.update(cx, |this, cx| {
this.set_dekey(keys, cx);
this.set_device_signer(keys, cx);
})
.ok();
}
@@ -789,7 +894,7 @@ impl NostrRegistry {
match task.await {
Ok(Some(keys)) => {
this.update(cx, |this, cx| {
this.set_dekey(keys, cx);
this.set_device_signer(keys, cx);
})
.ok();
}