diff --git a/crates/signed_core/src/addr.rs b/crates/signed_core/src/addr.rs index 79df08b..0a7bc51 100644 --- a/crates/signed_core/src/addr.rs +++ b/crates/signed_core/src/addr.rs @@ -1,44 +1,13 @@ -use std::fmt; -use std::str::FromStr; - use nostr::prelude::*; /// Address of a NIP-34 repository announcement: `30617::`. -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -pub struct RepoAddr { - pub owner: PublicKey, - pub id: String, -} - -impl RepoAddr { - pub fn new(owner: PublicKey, id: impl Into) -> Self { - Self { - owner, - id: id.into(), - } - } - - /// The NIP-33 coordinate for the announcement event (`a` tag value). - pub fn coordinate(&self) -> Coordinate { - Coordinate::new(Kind::GitRepoAnnouncement, self.owner).identifier(self.id.clone()) - } -} - -impl fmt::Display for RepoAddr { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{}", self.coordinate()) - } -} - -impl FromStr for RepoAddr { - type Err = nostr::error::Error; - - /// Parse from `::`, `naddr1...` bech32 or `nostr:naddr1...` URI. - fn from_str(s: &str) -> Result { - let coordinate = Coordinate::parse(s)?; - Ok(Self { - owner: coordinate.public_key, - id: coordinate.identifier, - }) - } +/// +/// The Rust Nostr SDK's [`Coordinate`] already provides parsing, formatting +/// and hashing for this; the alias keeps the repository-specific vocabulary +/// while reusing the SDK type. +pub type RepoAddr = Coordinate; + +/// Build the address of a NIP-34 repository announcement. +pub fn repo_addr(owner: PublicKey, id: impl Into) -> RepoAddr { + Coordinate::new(Kind::GitRepoAnnouncement, owner).identifier(id) } diff --git a/crates/signed_core/src/builders.rs b/crates/signed_core/src/builders.rs deleted file mode 100644 index 5650448..0000000 --- a/crates/signed_core/src/builders.rs +++ /dev/null @@ -1,29 +0,0 @@ -use nostr::prelude::*; - -/// Build a NIP-34 user grasp list (kind `10317`). -pub fn grasp_list(grasp_servers: Vec) -> EventBuilder { - GitUserGraspList { grasp_servers }.into_event_builder() -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn grasp_list_tags() { - let servers = vec![ - RelayUrl::parse("wss://gitnostr.com").unwrap(), - RelayUrl::parse("wss://relay.ngit.dev").unwrap(), - ]; - - let builder = grasp_list(servers); - - let urls: Vec<&str> = builder - .tags - .iter() - .filter_map(|t| t.content()) - .collect(); - - assert_eq!(urls, vec!["wss://gitnostr.com", "wss://relay.ngit.dev"]); - } -} diff --git a/crates/signed_core/src/clone_url.rs b/crates/signed_core/src/clone_url.rs index 19772cc..ca5cc94 100644 --- a/crates/signed_core/src/clone_url.rs +++ b/crates/signed_core/src/clone_url.rs @@ -28,10 +28,7 @@ pub fn parse_clone_url(url: &str) -> Option { if first.starts_with("naddr1") { let coordinate = Nip19Coordinate::from_bech32(first).ok()?; - return Some(CloneTarget::Addr(RepoAddr::new( - coordinate.coordinate.public_key, - coordinate.coordinate.identifier, - ))); + return Some(CloneTarget::Addr(coordinate.coordinate)); } let (relay_hint, identifier) = match third { diff --git a/crates/signed_core/src/filters.rs b/crates/signed_core/src/filters.rs index 3277462..8bc26c3 100644 --- a/crates/signed_core/src/filters.rs +++ b/crates/signed_core/src/filters.rs @@ -1,4 +1,3 @@ -use nostr::filter::{Alphabet, SingleLetterTag}; use nostr::prelude::*; use crate::RepoAddr; @@ -19,16 +18,16 @@ pub const ACTIVITY_KINDS: [Kind; 8] = [ pub fn announcement(addr: &RepoAddr) -> Filter { Filter::new() .kind(Kind::GitRepoAnnouncement) - .author(addr.owner) - .identifier(addr.id.clone()) + .author(addr.public_key) + .identifier(addr.identifier.clone()) } /// Latest state event (refs / HEAD) for a repository. pub fn state(addr: &RepoAddr) -> Filter { Filter::new() .kind(Kind::RepoState) - .author(addr.owner) - .identifier(addr.id.clone()) + .author(addr.public_key) + .identifier(addr.identifier.clone()) } /// All NIP-34 activity addressed to a repository (`#a` tag). @@ -36,9 +35,7 @@ pub fn state(addr: &RepoAddr) -> Filter { /// Note: the `a` tag on status events is optional per NIP-34, so statuses /// published without it won't be matched here. pub fn activity(addr: &RepoAddr) -> Filter { - Filter::new() - .kinds(ACTIVITY_KINDS) - .custom_tag(SingleLetterTag::lowercase(Alphabet::A), addr.to_string()) + Filter::new().kinds(ACTIVITY_KINDS).coordinate(addr) } /// Status events (`1630..=1633`) referencing a specific root event (`#e` tag). diff --git a/crates/signed_core/src/lib.rs b/crates/signed_core/src/lib.rs index e103493..19adfec 100644 --- a/crates/signed_core/src/lib.rs +++ b/crates/signed_core/src/lib.rs @@ -1,11 +1,10 @@ pub mod addr; -pub mod builders; pub mod clone_url; pub mod filters; pub mod model; pub mod status; -pub use addr::RepoAddr; +pub use addr::{RepoAddr, repo_addr}; pub use clone_url::{CloneTarget, parse_clone_url}; pub use model::Announcement; pub use status::{RepoStatus, references_root, resolve_status}; diff --git a/crates/signed_core/src/model.rs b/crates/signed_core/src/model.rs index 199f09a..b1a6ceb 100644 --- a/crates/signed_core/src/model.rs +++ b/crates/signed_core/src/model.rs @@ -40,27 +40,24 @@ impl Announcement { let mut maintainers: Vec = Vec::new(); for tag in event.tags.iter() { - let values: &[String] = tag.as_slice(); - match tag.kind() { - "d" => id = tag.content().map(str::to_owned), - "name" => name = tag.content().map(str::to_owned), - "description" => description = tag.content().map(str::to_owned), - "web" => web.extend(values.iter().skip(1).cloned()), - "clone" => clone.extend(values.iter().skip(1).cloned()), - "relays" => relays.extend(values.iter().skip(1).cloned()), - "r" => { - if values.get(2).map(String::as_str) == Some("euc") { - euc = tag.content().map(str::to_owned); - } + // The `d` tag isn't part of the NIP-34 tag codec; parse it directly. + if tag.kind() == "d" { + id = tag.content().map(str::to_owned); + continue; + } + + match Nip34Tag::parse(tag.as_slice()) { + Ok(Nip34Tag::Name(value)) => name = Some(value), + Ok(Nip34Tag::Description(value)) => description = Some(value), + Ok(Nip34Tag::Web(urls)) => web.extend(urls.into_iter().map(|url| url.to_string())), + Ok(Nip34Tag::Clone(urls)) => { + clone.extend(urls.into_iter().map(|url| url.to_string())) } - "maintainers" => { - maintainers.extend( - values - .iter() - .skip(1) - .filter_map(|v| PublicKey::from_hex(v).ok()), - ); + Ok(Nip34Tag::Relays(urls)) => { + relays.extend(urls.into_iter().map(|url| url.to_string())) } + Ok(Nip34Tag::EarliestUniqueCommitId(commit)) => euc = Some(commit.to_string()), + Ok(Nip34Tag::Maintainers(keys)) => maintainers.extend(keys), _ => {} } } @@ -81,6 +78,6 @@ impl Announcement { /// The repository address of this announcement. pub fn addr(&self) -> crate::RepoAddr { - crate::RepoAddr::new(self.owner, self.id.clone()) + crate::repo_addr(self.owner, self.id.clone()) } } diff --git a/crates/signed_core/src/status.rs b/crates/signed_core/src/status.rs index 5c159e3..1cd7bc9 100644 --- a/crates/signed_core/src/status.rs +++ b/crates/signed_core/src/status.rs @@ -32,11 +32,7 @@ impl RepoStatus { /// Check whether a status event references the given root event via an `e` tag. pub fn references_root(event: &Event, root: &EventId) -> bool { - let root_hex: String = root.to_hex(); - event - .tags - .iter() - .any(|t| t.kind() == "e" && t.content() == Some(root_hex.as_str())) + event.tags.event_ids().any(|id| id == *root) } /// Resolve the status of a root event per NIP-34: diff --git a/crates/signed_git/src/lib.rs b/crates/signed_git/src/lib.rs index 0d5bb29..b208eb2 100644 --- a/crates/signed_git/src/lib.rs +++ b/crates/signed_git/src/lib.rs @@ -25,8 +25,8 @@ impl GitCache { /// Local path of the clone for a repository. pub fn repo_path(&self, addr: &RepoAddr) -> PathBuf { self.root - .join(addr.owner.to_hex()) - .join(sanitize_path_component(&addr.id)) + .join(addr.public_key.to_hex()) + .join(sanitize_path_component(&addr.identifier)) } /// Open an existing clone. @@ -116,8 +116,14 @@ fn clone(url: &str, path: &Path) -> Result { Ok(repo) } +/// Map an untrusted repository id to a safe single path component. +/// +/// Replaces everything outside `[A-Za-z0-9._-]` with `_`, and rejects the +/// special components `.` and `..` so the id can't escape the cache root +/// when joined onto the owner directory. fn sanitize_path_component(id: &str) -> String { - id.chars() + let sanitized: String = id + .chars() .map(|c| { if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.') { c @@ -125,5 +131,54 @@ fn sanitize_path_component(id: &str) -> String { '_' } }) - .collect() + .collect(); + + if sanitized == "." || sanitized == ".." { + return "_".to_owned(); + } + + sanitized +} + +#[cfg(test)] +mod tests { + use nostr::prelude::*; + use signed_core::repo_addr; + + use super::*; + + #[test] + fn keeps_plain_ids() { + assert_eq!(sanitize_path_component("my-repo"), "my-repo"); + assert_eq!(sanitize_path_component("repo.v2"), "repo.v2"); + assert_eq!(sanitize_path_component("a_b-c"), "a_b-c"); + } + + #[test] + fn replaces_unsafe_characters() { + assert_eq!(sanitize_path_component("a/b\\c:d"), "a_b_c_d"); + assert_eq!(sanitize_path_component(""), ""); + } + + #[test] + fn blocks_parent_components() { + assert_eq!(sanitize_path_component(".."), "_"); + assert_eq!(sanitize_path_component("."), "_"); + // Separators are neutralized before the check, so these stay safe. + assert_eq!(sanitize_path_component("../.."), ".._.."); + assert_eq!(sanitize_path_component("a/../b"), "a_.._b"); + } + + #[test] + fn repo_path_stays_inside_root() { + let cache = GitCache::new("/cache".into()); + let owner = Keys::generate().public_key(); + + let path = cache.repo_path(&repo_addr(owner, "..")); + assert!(path.starts_with("/cache")); + assert_eq!( + path.file_name().map(|n| n.to_string_lossy().into_owned()), + Some("_".into()) + ); + } } diff --git a/crates/signed_nostr/src/backend.rs b/crates/signed_nostr/src/backend.rs index 04bf083..6055a29 100644 --- a/crates/signed_nostr/src/backend.rs +++ b/crates/signed_nostr/src/backend.rs @@ -1,6 +1,6 @@ use std::time::Duration; -use anyhow::{Context, Result, anyhow}; +use anyhow::{Context, Result}; use nostr_gossip_memory::prelude::*; #[cfg(not(target_arch = "wasm32"))] use nostr_lmdb::prelude::*; @@ -10,112 +10,47 @@ use nostr_sdk::prelude::*; use crate::signer::UniversalSigner; -/// Owns the nostr client: relay pool, LMDB database and signer. +/// Open (or create) the LMDB database at `db_path` and build a client +/// configured for Signed, together with a fresh signer. /// -/// The SDK manages its own internal tokio runtime; every method here is a -/// plain async fn that can be driven by GPUI's executors. -#[derive(Clone)] -pub struct NostrBackend { - client: Client, - signer: UniversalSigner, +/// The SDK manages its own internal tokio runtime; the returned client can be +/// driven by GPUI's executors. +#[cfg(not(target_arch = "wasm32"))] +pub async fn new_backend( + db_path: impl AsRef, +) -> Result<(Client, UniversalSigner)> { + let signer = UniversalSigner::new(Keys::generate()); + let database = NostrLmdb::open(db_path) + .await + .context("failed to open nostr database")?; + Ok(with_database(signer, database)) } -impl NostrBackend { - /// Open (or create) the LMDB database at `db_path` and build the client. - #[cfg(not(target_arch = "wasm32"))] - pub async fn new(db_path: impl AsRef) -> Result { - let signer = UniversalSigner::new(Keys::generate()); - let database = NostrLmdb::open(db_path) - .await - .context("failed to open nostr database")?; - Ok(Self::with_database(signer, database)) - } - - /// In-memory database on wasm (no LMDB available). - #[cfg(target_arch = "wasm32")] - pub fn new() -> Result { - let signer = UniversalSigner::new(Keys::generate()); - Ok(Self::with_database(signer, MemoryDatabase::unbounded())) - } - - fn with_database(signer: UniversalSigner, database: D) -> Self - where - D: IntoNostrDatabase, - { - let authenticator = SignerAuthenticator::new(signer.clone()); - - let client = ClientBuilder::default() - .database(database) - .authenticator(authenticator) - .gossip(NostrGossipMemory::unbounded()) - .gossip_config(GossipConfig::default().no_background_refresh()) - .connect_timeout(Duration::from_secs(10)) - .verify_subscriptions(true) - .ban_relay_on_mismatch(true) - .sleep_when_idle(SleepWhenIdle::Enabled { - timeout: Duration::from_secs(600), - }) - .build(); - - Self { client, signer } - } - - pub fn client(&self) -> Client { - self.client.clone() - } - - pub fn signer(&self) -> UniversalSigner { - self.signer.clone() - } - - pub async fn add_relay(&self, url: &str) -> Result<()> { - self.client.add_relay(url).await?; - Ok(()) - } - - /// Add a relay used only for discovery (e.g. NIP-65 indexer relays). - /// No subscriptions or writes are routed through it. - pub async fn add_discovery_relay(&self, url: &str) -> Result<()> { - self.client - .add_relay(url) - .capabilities(RelayCapabilities::DISCOVERY) - .await?; - Ok(()) - } - - pub async fn connect(&self) { - self.client.connect().await; - } - - /// Start a persistent subscription. Received events are stored in the - /// database automatically by the relay pool. - pub async fn subscribe(&self, filter: Filter) -> Result { - let output = self.client.subscribe(filter).await?; - Ok(output.value) - } - - /// Query the local database (the single source of truth for the UI). - pub async fn query(&self, filter: Filter) -> Result> { - let events = self.client.database().query(filter).await?; - Ok(events.into_iter().collect()) - } - - /// Sign with the current signer, broadcast, and save locally so the - /// event is immediately visible to [`NostrBackend::query`]. - pub async fn send(&self, builder: EventBuilder) -> Result { - let event = builder.finalize_async(&self.signer).await?; - let output = self.client.send_event(&event).await?; - - if output.success.is_empty() && !output.failed.is_empty() { - let reasons = output - .failed - .values() - .cloned() - .collect::>() - .join(", "); - return Err(anyhow!("event not accepted by any relay: {reasons}")); - } - - Ok(event) - } +/// In-memory database on wasm (no LMDB available). +#[cfg(target_arch = "wasm32")] +pub fn new_backend() -> Result<(Client, UniversalSigner)> { + let signer = UniversalSigner::new(Keys::generate()); + Ok(with_database(signer, MemoryDatabase::unbounded())) +} + +fn with_database(signer: UniversalSigner, database: D) -> (Client, UniversalSigner) +where + D: IntoNostrDatabase, +{ + let authenticator = SignerAuthenticator::new(signer.clone()); + + let client = ClientBuilder::default() + .database(database) + .authenticator(authenticator) + .gossip(NostrGossipMemory::unbounded()) + .gossip_config(GossipConfig::default().no_background_refresh()) + .connect_timeout(Duration::from_secs(10)) + .verify_subscriptions(true) + .ban_relay_on_mismatch(true) + .sleep_when_idle(SleepWhenIdle::Enabled { + timeout: Duration::from_secs(600), + }) + .build(); + + (client, signer) } diff --git a/crates/signed_nostr/src/lib.rs b/crates/signed_nostr/src/lib.rs index 1328bf4..3c99075 100644 --- a/crates/signed_nostr/src/lib.rs +++ b/crates/signed_nostr/src/lib.rs @@ -2,6 +2,6 @@ mod backend; mod signer; mod update; -pub use backend::NostrBackend; +pub use backend::new_backend; pub use signer::{SignedAuthUrlHandler, UniversalSigner}; pub use update::Update; diff --git a/crates/signed_nostr/src/signer.rs b/crates/signed_nostr/src/signer.rs index 9f676d2..9f8e592 100644 --- a/crates/signed_nostr/src/signer.rs +++ b/crates/signed_nostr/src/signer.rs @@ -194,7 +194,7 @@ impl AuthUrlHandler for SignedAuthUrlHandler { auth_url: Url, ) -> Pin> + Send + '_>> { Box::pin(async move { - webbrowser::open(auth_url.as_str()).unwrap(); + webbrowser::open(auth_url.as_str()).map_err(nostr_connect::error::Error::other)?; Ok(()) }) } diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index f6121d7..4e43e34 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -6,8 +6,8 @@ use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Task}; use nostr_connect::prelude::*; use nostr_sdk::client::SyncSummary; use nostr_sdk::prelude::*; -use signed_core::{builders, filters}; -use signed_nostr::{NostrBackend, SignedAuthUrlHandler, UniversalSigner, Update}; +use signed_core::filters; +use signed_nostr::{SignedAuthUrlHandler, UniversalSigner, Update}; /// Keyring entry holding the user credential (`nsec1...` or `bunker://...` /// with an embedded `?master=` NIP-46 session key). @@ -71,7 +71,8 @@ impl BackendEvent { /// notification pump. Stores subscribe to [`BackendEvent`] and re-query the /// local database when relevant updates arrive. pub struct Backend { - inner: NostrBackend, + client: Client, + signer: UniversalSigner, current_user: Option, connected: bool, sync_progress: Option<(u64, u64)>, @@ -94,11 +95,11 @@ impl Backend { cx.set_global(GlobalBackend(entity)); } - pub(crate) fn new(inner: NostrBackend, cx: &mut Context) -> Self { - let client = inner.client(); + pub(crate) fn new(client: Client, signer: UniversalSigner, cx: &mut Context) -> Self { + let pump_client = client.clone(); let pump = cx.spawn(async move |this, cx| { - let mut notifications = client.notifications(); + let mut notifications = pump_client.notifications(); while let Some(notification) = notifications.next().await { let ClientNotification::Event { event, .. } = notification else { @@ -119,7 +120,8 @@ impl Backend { }); let mut this = Self { - inner, + client, + signer, current_user: None, connected: false, sync_progress: None, @@ -133,16 +135,19 @@ impl Backend { /// Bootstrap the client: connect to the default relays (indexers as /// discovery-only) and restore the saved session, if any. fn bootstrap(&mut self, cx: &mut Context) { - let backend = self.inner.clone(); + let client = self.client.clone(); let task = cx.background_spawn(async move { for url in BOOTSTRAP_RELAYS { - backend.add_relay(url).await?; + client.add_relay(url).await?; } for url in INDEXER_RELAYS { - backend.add_discovery_relay(url).await?; + client + .add_relay(url) + .capabilities(RelayCapabilities::DISCOVERY) + .await?; } - backend.connect().await; + client.connect().await; Ok::<(), Error>(()) }); @@ -276,7 +281,7 @@ impl Backend { this.update(cx, |this, cx| { // Become the new identity, so the publishes below are // signed with the new keys. - this.inner.signer().swap_inner(keys); + this.signer.swap_inner(keys); this.current_user = Some(public_key); this.bootstrap_user(public_key, cx); cx.emit(BackendEvent::SignerChanged); @@ -317,7 +322,7 @@ impl Backend { .map(|url| RelayUrl::parse(url).expect("valid relay URL")) .collect(); - this.send(builders::grasp_list(grasp_servers), cx); + this.send(GitUserGraspList { grasp_servers }.into_event_builder(), cx); })?; Ok(public_key) @@ -441,7 +446,7 @@ impl Backend { delete.await.ok(); this.update(cx, |this, cx| { - this.inner.signer().swap_inner(Keys::generate()); + this.signer.swap_inner(Keys::generate()); this.current_user = None; cx.emit(BackendEvent::SignerChanged); cx.emit(BackendEvent::SignerRequired); @@ -455,14 +460,11 @@ impl Backend { /// Fetch the user's grasp list (kind `10317`) and add the listed grasp /// servers as relays. fn bootstrap_user(&mut self, public_key: PublicKey, cx: &mut Context) { - let backend = self.inner.clone(); + let client = self.client.clone(); self.tasks.push(cx.spawn(async move |this, cx| { let result = async { - let events = backend - .client() - .fetch_events(filters::grasp_list(public_key)) - .await?; + let events = client.fetch_events(filters::grasp_list(public_key)).await?; let urls: Vec = events .into_iter() @@ -477,9 +479,9 @@ impl Backend { .unwrap_or_default(); for url in urls { - backend.add_relay(&url).await.ok(); + client.add_relay(&url).await.ok(); } - backend.connect().await; + client.connect().await; Ok::<_, Error>(()) } @@ -495,12 +497,12 @@ impl Backend { /// Get the nostr client. pub fn client(&self) -> Client { - self.inner.client() + self.client.clone() } /// Get the current signer. pub fn signer(&self) -> UniversalSigner { - self.inner.signer() + self.signer.clone() } /// Get the current user's public key. @@ -536,7 +538,7 @@ impl Backend { match new_signer.get_public_key_async().await { Ok(public_key) => { this.update(cx, |this, cx| { - this.inner.signer().swap_inner(new_signer); + this.signer.swap_inner(new_signer); this.current_user = Some(public_key); this.bootstrap_user(public_key, cx); cx.emit(BackendEvent::SignerChanged); @@ -557,13 +559,13 @@ impl Backend { /// Add relays and connect to them. pub fn add_relays(&mut self, urls: Vec, cx: &mut Context) { - let backend = self.inner.clone(); + let client = self.client.clone(); let task = cx.background_spawn(async move { for url in urls { - backend.add_relay(&url).await?; + client.add_relay(&url).await?; } - backend.connect().await; + client.connect().await; Ok::<(), Error>(()) }); @@ -587,13 +589,16 @@ impl Backend { /// Add relays used only for discovery (e.g. NIP-65 indexers) and /// connect to them. No subscriptions or writes are routed through them. pub fn add_discovery_relays(&mut self, urls: Vec, cx: &mut Context) { - let backend = self.inner.clone(); + let client = self.client.clone(); let task = cx.background_spawn(async move { for url in urls { - backend.add_discovery_relay(&url).await?; + client + .add_relay(&url) + .capabilities(RelayCapabilities::DISCOVERY) + .await?; } - backend.connect().await; + client.connect().await; Ok::<(), Error>(()) }); @@ -608,9 +613,9 @@ impl Backend { /// Start a persistent subscription. Matching events are stored in the /// database automatically and surface as [`BackendEvent::NostrUpdate`]. pub fn subscribe(&mut self, filter: Filter, cx: &mut Context) { - let backend = self.inner.clone(); + let client = self.client.clone(); - let task = cx.background_spawn(async move { backend.subscribe(filter).await.map(|_| ()) }); + let task = cx.background_spawn(async move { client.subscribe(filter).await.map(|_| ()) }); self.tasks.push(cx.spawn(async move |this, cx| { if let Err(e) = task.await { @@ -625,11 +630,10 @@ impl Backend { /// in the database and surface as [`BackendEvent::NostrUpdate`] while the /// subscription is open. pub fn subscribe_bootstrap(&mut self, filters: Vec, cx: &mut Context) { - let backend = self.inner.clone(); + let client = self.client.clone(); - let task = cx.background_spawn(async move { - subscribe_bootstrap_only(&backend.client(), filters).await - }); + let task = + cx.background_spawn(async move { subscribe_bootstrap_only(&client, filters).await }); self.tasks.push(cx.spawn(async move |this, cx| { if let Err(e) = task.await { @@ -644,7 +648,7 @@ impl Backend { /// Emits [`BackendEvent::SyncProgress`] while running (throttled to /// whole-percent changes) and [`BackendEvent::Synced`] on completion. pub fn sync_bootstrap(&mut self, filter: Filter, cx: &mut Context) { - let backend = self.inner.clone(); + let client = self.client.clone(); self.sync_progress = Some((0, 0)); cx.notify(); @@ -681,7 +685,7 @@ impl Backend { let task = cx.background_spawn(async move { let opts = SyncOptions::default().progress(tx); - sync_bootstrap_only(&backend.client(), filter, opts).await + sync_bootstrap_only(&client, filter, opts).await }); self.tasks.push(cx.spawn(async move |this, cx| { @@ -721,8 +725,27 @@ impl Backend { cx: &mut Context, ) -> flume::Receiver> { let (tx, rx) = flume::bounded(1); - let backend = self.inner.clone(); - let task = cx.background_spawn(async move { backend.send(builder).await }); + let client = self.client.clone(); + let signer = self.signer.clone(); + + let task = cx.background_spawn(async move { + // Sign with the current signer, broadcast, and save locally so + // the event is immediately visible to database queries. + let event = builder.finalize_async(&signer).await?; + let output = client.send_event(&event).await?; + + if output.success.is_empty() && !output.failed.is_empty() { + let reasons = output + .failed + .values() + .cloned() + .collect::>() + .join(", "); + return Err(anyhow!("event not accepted by any relay: {reasons}")); + } + + Ok(event) + }); self.tasks.push(cx.spawn(async move |this, cx| { let result = task.await; diff --git a/crates/signed_state/src/lib.rs b/crates/signed_state/src/lib.rs index 25fe3ec..8ccf4dc 100644 --- a/crates/signed_state/src/lib.rs +++ b/crates/signed_state/src/lib.rs @@ -10,7 +10,7 @@ use gpui::{App, AppContext, Entity}; pub use profile::{Profile, ProfileStore, shorten_pubkey}; pub use repo::RepoStore; pub use repo_list::RepoListStore; -use signed_nostr::NostrBackend; +use signed_nostr::new_backend; /// Initialize the backend and stores, and install them as globals. Call once /// at startup, before opening any window that uses the stores. @@ -22,13 +22,13 @@ pub fn init(db_path: impl AsRef, cx: &mut App) -> Entity { .ok(); let path = db_path.as_ref().to_path_buf(); - let inner = cx.foreground_executor().block_on(async move { - NostrBackend::new(path) + let (client, signer) = cx.foreground_executor().block_on(async move { + new_backend(path) .await .expect("failed to initialize nostr backend") }); - let entity = cx.new(|cx| Backend::new(inner, cx)); + let entity = cx.new(|cx| Backend::new(client, signer, cx)); Backend::set_global(entity.clone(), cx); ProfileStore::set_global(cx.new(ProfileStore::new), cx); @@ -39,9 +39,9 @@ pub fn init(db_path: impl AsRef, cx: &mut App) -> Entity { /// Initialize the backend with an in-memory database on wasm. #[cfg(target_arch = "wasm32")] pub fn init(cx: &mut App) -> Entity { - let inner = NostrBackend::new().expect("failed to initialize nostr backend"); + let (client, signer) = new_backend().expect("failed to initialize nostr backend"); - let entity = cx.new(|cx| Backend::new(inner, cx)); + let entity = cx.new(|cx| Backend::new(client, signer, cx)); Backend::set_global(entity.clone(), cx); ProfileStore::set_global(cx.new(ProfileStore::new), cx); diff --git a/crates/signed_state/src/profile.rs b/crates/signed_state/src/profile.rs index 9bd05c0..99eb183 100644 --- a/crates/signed_state/src/profile.rs +++ b/crates/signed_state/src/profile.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet}; use std::time::Duration; use anyhow::Error; -use gpui::{App, Context, Entity, Global, SharedString, Subscription, Task}; +use gpui::{App, AppContext, Context, Entity, Global, SharedString, Subscription, Task}; use nostr_sdk::prelude::*; use crate::backend::{Backend, BackendEvent, sync_bootstrap_only}; @@ -138,47 +138,68 @@ impl ProfileStore { fn load(&mut self, cx: &mut Context) { let client = Backend::global(cx).read(cx).client(); - let task = cx.spawn(async move |this, cx| { + let work = cx.background_spawn(async move { let filter = Filter::new().kind(Kind::Metadata).limit(200); let events = client.database().query(filter).await?; - this.update(cx, |this, cx| { - for event in events { + // Parse off the main thread; only plain profiles cross back. + let profiles: Vec = events + .into_iter() + .map(|event| { let metadata = Metadata::from_json(&event.content).unwrap_or_default(); - this.profiles - .insert(event.pubkey, Profile::new(event.pubkey, metadata)); + Profile::new(event.pubkey, metadata) + }) + .collect(); + + Ok::<_, Error>(profiles) + }); + + self.tasks.push(cx.spawn(async move |this, cx| { + let profiles = work.await?; + + this.update(cx, |this, cx| { + for profile in profiles { + this.profiles.insert(profile.public_key(), profile); } cx.notify(); })?; Ok(()) - }); - - self.tasks.push(task); + })); } /// Re-read the latest metadata of an author from the local database. fn apply_author(&mut self, public_key: PublicKey, cx: &mut Context) { let client = Backend::global(cx).read(cx).client(); - let task = cx.spawn(async move |this, cx| { + let work = cx.background_spawn(async move { let filter = Filter::new().kind(Kind::Metadata).author(public_key); let events = client.database().query(filter).await?; - if let Some(event) = events.into_iter().max_by_key(|e| e.created_at) { - let metadata = Metadata::from_json(event.content).unwrap_or_default(); + // Parse off the main thread; only the profile crosses back. + let profile = events + .into_iter() + .max_by_key(|e| e.created_at) + .map(|event| { + let metadata = Metadata::from_json(event.content).unwrap_or_default(); + Profile::new(event.pubkey, metadata) + }); - this.update(cx, |this, cx| { - this.profiles - .insert(public_key, Profile::new(public_key, metadata)); - cx.notify(); - })?; - } - - Ok(()) + Ok::<_, Error>(profile) }); - self.tasks.push(task); + self.tasks.push(cx.spawn(async move |this, cx| { + let profile = work.await?; + + this.update(cx, |this, cx| { + if let Some(profile) = profile { + this.profiles.insert(profile.public_key(), profile); + cx.notify(); + } + })?; + + Ok(()) + })); } /// Re-read the latest metadata of every requested author from the local @@ -191,10 +212,11 @@ impl ProfileStore { let client = Backend::global(cx).read(cx).client(); let authors: Vec = self.seen.iter().copied().collect(); - let task = cx.spawn(async move |this, cx| { + let work = cx.background_spawn(async move { let filter = Filter::new().kind(Kind::Metadata).authors(authors); let events = client.database().query(filter).await?; + // Pick the latest metadata per author off the main thread. let mut latest: HashMap = HashMap::new(); for event in events { match latest.get(&event.pubkey) { @@ -211,18 +233,26 @@ impl ProfileStore { } } + let profiles: Vec = latest + .into_iter() + .map(|(public_key, (_, metadata))| Profile::new(public_key, metadata)) + .collect(); + + Ok::<_, Error>(profiles) + }); + + self.tasks.push(cx.spawn(async move |this, cx| { + let profiles = work.await?; + this.update(cx, |this, cx| { - for (public_key, (_, metadata)) in latest { - this.profiles - .insert(public_key, Profile::new(public_key, metadata)); + for profile in profiles { + this.profiles.insert(profile.public_key(), profile); } cx.notify(); })?; Ok(()) - }); - - self.tasks.push(task); + })); } /// Drain the queue in a batched fetch, debounced to collect requests. diff --git a/crates/signed_state/src/repo.rs b/crates/signed_state/src/repo.rs index c98fce6..22cd35a 100644 --- a/crates/signed_state/src/repo.rs +++ b/crates/signed_state/src/repo.rs @@ -1,5 +1,5 @@ use anyhow::Error; -use gpui::{Context, Subscription, Task}; +use gpui::{AppContext, Context, Subscription, Task}; use nostr_sdk::prelude::*; use signed_core::{Announcement, RepoAddr, RepoStatus, filters}; @@ -33,20 +33,16 @@ impl RepoStore { let subscription = cx.subscribe(&backend, |this, _backend, event, cx| { let relevant = match event { BackendEvent::NostrUpdate(update) => { - let coordinate = update.coordinate.as_ref() == Some(&this.addr.coordinate()); - let author = update.author == this.addr.owner; + let coordinate = update.coordinate.as_ref() == Some(&this.addr); + let author = update.author == this.addr.public_key; let kind = update.kind == Kind::GitRepoAnnouncement; coordinate || (author && kind) } BackendEvent::Published(event) => { let kind = event.kind == Kind::GitRepoAnnouncement; - let author = event.pubkey == this.addr.owner; - let coordinate = event - .tags - .coordinates() - .into_iter() - .any(|c| c == this.addr.coordinate()); + let author = event.pubkey == this.addr.public_key; + let coordinate = event.tags.coordinates().into_iter().any(|c| c == this.addr); coordinate || (kind && author) } @@ -103,7 +99,8 @@ impl RepoStore { /// Re-query the local database and update all fields. /// /// Debounced: concurrent requests are coalesced into a single re-query - /// after the running one finishes. + /// after the running one finishes. The query and processing run on a + /// background thread; only the results are applied on the main thread. pub fn refresh(&mut self, cx: &mut Context) { if self.refreshing { self.refresh_dirty = true; @@ -114,84 +111,99 @@ impl RepoStore { let client = Backend::global(cx).read(cx).client(); let addr = self.addr.clone(); - let task = cx.spawn(async move |this, cx| { - loop { - let queries = async { - let db = client.database(); + let work = cx.background_spawn(async move { + let queries = async { + let db = client.database(); - let announcements = db.query(filters::announcement(&addr)).await?; - let states = db.query(filters::state(&addr)).await?; - let activity = db.query(filters::activity(&addr)).await?; + let announcements = db.query(filters::announcement(&addr)).await?; + let states = db.query(filters::state(&addr)).await?; + let activity = db.query(filters::activity(&addr)).await?; - Ok::<_, Error>((announcements, states, activity)) - } - .await; + Ok::<_, Error>((announcements, states, activity)) + } + .await?; - let (announcements, states, activity) = match queries { - Ok(results) => results, - Err(e) => { - return this.update(cx, |this, cx| { - this.refreshing = false; - this.last_error = Some(e.to_string()); - cx.notify(); - }); - } - }; + let (announcements, states, activity) = queries; - let again = this.update(cx, |this, cx| { - this.announcement = latest(announcements) - .as_ref() - .and_then(Announcement::from_event); + // Parse and sort off the main thread; only plain data + // crosses back into the entity. + let announcement = latest(announcements) + .as_ref() + .and_then(Announcement::from_event); - if let Some(state) = latest(states) { - let (refs, head) = parse_state(&state); - this.refs = refs; - this.head = head; - } + let state = latest(states).map(|state| parse_state(&state)); - this.issues.clear(); - this.patches.clear(); - this.pull_requests.clear(); - this.statuses.clear(); + let (mut issues, mut patches, mut pull_requests, mut statuses) = + (Vec::new(), Vec::new(), Vec::new(), Vec::new()); - for event in activity { - match event.kind { - Kind::GitIssue => this.issues.push(event), - Kind::GitPatch => this.patches.push(event), - Kind::GitPullRequest | Kind::GitPullRequestUpdate => { - this.pull_requests.push(event) - } - kind if RepoStatus::from_kind(kind).is_some() => { - this.statuses.push(event) - } - _ => {} - } - } - - sort_newest_first(&mut this.issues); - sort_newest_first(&mut this.patches); - sort_newest_first(&mut this.pull_requests); - - cx.notify(); - - if this.refresh_dirty { - this.refresh_dirty = false; - true - } else { - this.refreshing = false; - false - } - })?; - - if !again { - break; + for event in activity { + match event.kind { + Kind::GitIssue => issues.push(event), + Kind::GitPatch => patches.push(event), + Kind::GitPullRequest | Kind::GitPullRequestUpdate => pull_requests.push(event), + kind if RepoStatus::from_kind(kind).is_some() => statuses.push(event), + _ => {} } } - Ok(()) + sort_newest_first(&mut issues); + sort_newest_first(&mut patches); + sort_newest_first(&mut pull_requests); + + Ok::<_, Error>(( + announcement, + state, + issues, + patches, + pull_requests, + statuses, + )) }); - self.tasks.push(task); + self.tasks.push(cx.spawn(async move |this, cx| { + let (announcement, state, issues, patches, pull_requests, statuses) = match work.await { + Ok(data) => data, + Err(e) => { + return this.update(cx, |this, cx| { + this.refreshing = false; + this.last_error = Some(e.to_string()); + cx.notify(); + }); + } + }; + + let again = this.update(cx, |this, cx| { + this.announcement = announcement; + + if let Some((refs, head)) = state { + this.refs = refs; + this.head = head; + } + + this.issues = issues; + this.patches = patches; + this.pull_requests = pull_requests; + this.statuses = statuses; + + cx.notify(); + + this.refreshing = false; + if this.refresh_dirty { + this.refresh_dirty = false; + true + } else { + false + } + })?; + + // Requests that arrived while the refresh was running are + // coalesced into one follow-up refresh. + if again { + this.update(cx, |this, cx| this.refresh(cx))?; + } + + Ok(()) + })); } /// Resolve the status of a root event (issue / patch / PR) per NIP-34. @@ -213,7 +225,7 @@ impl RepoStore { /// Open an issue on this repository. pub fn open_issue(&mut self, subject: Option, content: String, cx: &mut Context) { let builder = GitIssue { - repository: self.addr.coordinate(), + repository: self.addr.clone(), content, subject, labels: Vec::new(), @@ -230,8 +242,8 @@ impl RepoStore { }; let builder = EventBuilder::new(Kind::GitPatch, patch).tags([ - Tag::coordinate(self.addr.coordinate(), None), - Tag::public_key(self.addr.owner), + Tag::coordinate(self.addr.clone(), None), + Tag::public_key(self.addr.public_key), root_marker, ]); @@ -246,9 +258,9 @@ impl RepoStore { let builder = EventBuilder::new(status.kind(), "").tags([ root_ref, - Tag::public_key(self.addr.owner), + Tag::public_key(self.addr.public_key), Tag::public_key(root.pubkey), - Tag::coordinate(self.addr.coordinate(), None), + Tag::coordinate(self.addr.clone(), None), ]); self.send(builder, cx); @@ -288,16 +300,15 @@ fn parse_state(event: &Event) -> (Vec<(String, String)>, Option) { let mut head = None; for tag in event.tags.iter() { - let kind = tag.kind(); - if kind == "HEAD" { - head = tag - .content() - .and_then(|v| v.strip_prefix("ref: refs/heads/")) - .map(str::to_owned); - } else if kind.starts_with("refs/") - && let Some(commit) = tag.content() - { - refs.push((kind.to_owned(), commit.to_owned())); + match Nip34Tag::parse(tag.as_slice()) { + Ok(Nip34Tag::Head(branch)) => head = Some(branch), + Ok(Nip34Tag::RefHead { branch, commit }) => { + refs.push((format!("refs/heads/{branch}"), commit.to_string())); + } + Ok(Nip34Tag::RefTag { name, commit }) => { + refs.push((format!("refs/tags/{name}"), commit.to_string())); + } + _ => {} } } diff --git a/crates/signed_state/src/repo_list.rs b/crates/signed_state/src/repo_list.rs index 0b61c14..bf2a2b3 100644 --- a/crates/signed_state/src/repo_list.rs +++ b/crates/signed_state/src/repo_list.rs @@ -1,9 +1,9 @@ use std::collections::HashMap; use anyhow::Error; -use gpui::{Context, Subscription, Task}; +use gpui::{AppContext, Context, Subscription, Task}; use nostr_sdk::prelude::*; -use signed_core::{Announcement, filters}; +use signed_core::{Announcement, RepoAddr, filters}; use crate::backend::{Backend, BackendEvent}; @@ -79,7 +79,8 @@ impl RepoListStore { /// Re-query the local database. Latest announcement per repository wins. /// /// Debounced: concurrent requests are coalesced into a single re-query - /// after the running one finishes. + /// after the running one finishes. The query and processing run on a + /// background thread; only the results are applied on the main thread. pub fn refresh(&mut self, cx: &mut Context) { if self.refreshing { self.refresh_dirty = true; @@ -90,63 +91,70 @@ impl RepoListStore { let client = Backend::global(cx).read(cx).client(); let author = self.author; - let task = cx.spawn(async move |this, cx| { - loop { - let filter = match author { - Some(a) => filters::announcements_by(a), - None => filters::all_announcements(), + let work = cx.background_spawn(async move { + let filter = match author { + Some(a) => filters::announcements_by(a), + None => filters::all_announcements(), + }; + + let events = client.database().query(filter).await?; + + // Dedup and sort off the main thread; only the final list + // crosses back into the entity. + let mut by_repo: HashMap = HashMap::new(); + + for event in events { + let Some(announcement) = Announcement::from_event(&event) else { + continue; }; - let events = match client.database().query(filter).await { - Ok(events) => events, - Err(_) => { - return this.update(cx, |this, _cx| { - this.refreshing = false; - }); + let addr = announcement.addr(); + + match by_repo.get(&addr) { + Some(existing) if existing.created_at >= announcement.created_at => {} + _ => { + by_repo.insert(addr, announcement); } - }; - - let again = this.update(cx, |this, cx| { - let mut by_repo: HashMap<(String, String), Announcement> = HashMap::new(); - - for event in events { - let Some(announcement) = Announcement::from_event(&event) else { - continue; - }; - - let key = (announcement.owner.to_hex(), announcement.id.clone()); - - match by_repo.get(&key) { - Some(existing) if existing.created_at >= announcement.created_at => {} - _ => { - by_repo.insert(key, announcement); - } - } - } - - let mut announcements: Vec = by_repo.into_values().collect(); - announcements.sort_by_key(|a| std::cmp::Reverse(a.created_at)); - - this.announcements = announcements; - cx.notify(); - - if this.refresh_dirty { - this.refresh_dirty = false; - true - } else { - this.refreshing = false; - false - } - })?; - - if !again { - break; } } - Ok(()) + let mut announcements: Vec = by_repo.into_values().collect(); + announcements.sort_by_key(|a| std::cmp::Reverse(a.created_at)); + + Ok::<_, Error>(announcements) }); - self.tasks.push(task); + self.tasks.push(cx.spawn(async move |this, cx| { + let announcements = match work.await { + Ok(announcements) => announcements, + // Database errors are transient; keep the last list. + Err(_) => { + return this.update(cx, |this, _cx| { + this.refreshing = false; + }); + } + }; + + let again = this.update(cx, |this, cx| { + this.announcements = announcements; + cx.notify(); + + this.refreshing = false; + if this.refresh_dirty { + this.refresh_dirty = false; + true + } else { + false + } + })?; + + // Requests that arrived while the refresh was running are + // coalesced into one follow-up refresh. + if again { + this.update(cx, |this, cx| this.refresh(cx))?; + } + + Ok(()) + })); } }