reya/signed
1
0
Fork 0
Code Issues Pull Requests Actions 6 Releases Activity

update issue panel

Browse Source
This commit is contained in:
reya
2026-08-23 19:54:50 +07:00
parent 1956cb96bb
commit 6eccd65b93
13 changed files with 470 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/signed_state/src/backend.rs
+68
View File
@@ -670,6 +670,30 @@ impl Backend {
}));
}
/// Connect to relays announced by a repository (NIP-34 `relays` tag) and
/// fetch its events from them: a one-shot auto-closing subscription for
/// `filters`, plus a negentropy sync so issues, patches and PRs stored
/// only on those relays are not missed.
///
/// Best-effort: failures are logged, not surfaced, because the bootstrap
/// relays already cover the repository. The relays stay in the pool, so
/// events the user publishes for this repository also reach them.
pub fn connect_repo_relays(
&mut self,
relays: Vec<RelayUrl>,
filters: Vec<Filter>,
cx: &mut Context<Self>,
) {
let client = self.client.clone();
self.tasks.push(cx.spawn(async move |_this, _cx| {
if let Err(e) = connect_repo_relays_only(&client, relays, filters).await {
log::warn!("repo relay fetch failed: {e}");
}
Ok(())
}));
}
/// Start a one-shot subscription targeted only at the bootstrap relays,
/// auto-closing after EOSE or a short timeout. Matching events are stored
/// in the database and surface as [`BackendEvent::NostrUpdate`] while the
@@ -832,6 +856,50 @@ impl Backend {
}
}
/// Add the given relays, connect to them, and fetch the filters: a one-shot
/// subscription (auto-closing after EOSE) plus a negentropy sync per filter
/// as a second pass, so events that race with the subscription or relays
/// with flaky EOSE behavior can't be missed. Relays without NEG-XX support
/// just fail the sync step; the subscription already covered them.
async fn connect_repo_relays_only(
client: &Client,
relays: Vec<RelayUrl>,
filters: Vec<Filter>,
) -> Result<(), Error> {
if relays.is_empty() {
return Ok(());
}
for url in &relays {
client.add_relay(url).await?;
}
client.connect().await;
let opts = SubscribeAutoCloseOptions::default()
.exit_policy(ReqExitPolicy::ExitOnEOSE)
.timeout(Some(Duration::from_secs(10)));
let target: HashMap<&str, Vec<Filter>> = relays
.iter()
.map(|url| (url.as_str(), filters.clone()))
.collect();
client.subscribe(target).close_on(opts).await?;
for filter in filters {
let sync_opts = SyncOptions::default().initial_timeout(Duration::from_secs(5));
if let Err(e) = client
.sync(filter)
.with(relays.iter())
.opts(sync_opts)
.await
{
log::warn!("repo relay negentropy sync failed: {e}");
}
}
Ok(())
}
/// Subscribe only on the bootstrap relays, auto-closing after EOSE or a
/// short timeout. Use for one-shot data fetches (repo events, profiles)
/// instead of persistent gossip-routed subscriptions.
crates/signed_state/src/repo.rs
+110 -11
View File
@@ -1,3 +1,4 @@
use std::collections::HashSet;
use std::time::Duration;
use anyhow::Error;
@@ -32,6 +33,13 @@ pub struct RepoStore {
statuses: Vec<Event>,
/// Error of the last action initiated from this store, if any.
pub last_error: Option<String>,
/// Relays announced by this repository (NIP-34 `relays` tag) that we
/// have already been asked to connect to and fetch from, to avoid
/// re-subscribing on every refresh.
repo_relays: HashSet<RelayUrl>,
/// Root events (issues, patches, PRs) for which a NIP-22 comment fetch
/// has already been requested, to avoid re-fetching on every refresh.
comment_roots: HashSet<EventId>,
refreshing: bool,
refresh_dirty: bool,
/// A refresh is waiting out [`REFRESH_DEBOUNCE`].
@@ -41,7 +49,7 @@ pub struct RepoStore {
}
impl RepoStore {
pub fn new(addr: RepoAddr, cx: &mut Context<Self>) -> Self {
pub fn new(addr: RepoAddr, announced_relays: Vec<RelayUrl>, cx: &mut Context<Self>) -> Self {
let backend = Backend::global(cx);
let subscription = cx.subscribe(&backend, |this, _backend, event, cx| {
@@ -53,8 +61,12 @@ impl RepoStore {
let coordinate = update.coordinate.as_ref() == Some(&this.addr);
let author = update.author == this.addr.public_key;
let kind = update.kind == Kind::GitRepoAnnouncement;
// NIP-22 comments carry no `a` tag, so they can't be
// matched by coordinate; any comment may reference this
// repository's roots.
let comment = update.kind == Kind::Comment;
deletion || coordinate || (author && kind)
deletion || coordinate || (author && kind) || comment
}
BackendEvent::Published(event) => {
let kind = event.kind == Kind::GitRepoAnnouncement;
@@ -82,6 +94,8 @@ impl RepoStore {
comments: Vec::new(),
statuses: Vec::new(),
last_error: None,
repo_relays: HashSet::new(),
comment_roots: HashSet::new(),
refreshing: false,
refresh_dirty: false,
debouncing: false,
@@ -90,6 +104,10 @@ impl RepoStore {
};
store.subscribe_remote(cx);
// The announcement we opened the repo from may already list its
// relays; connect to them right away instead of waiting for the
// bootstrap fetch to return the same event.
store.connect_announced_relays(&announced_relays, cx);
store.refresh(cx);
store
}
@@ -98,6 +116,42 @@ impl RepoStore {
&self.addr
}
/// Filters that make up a repository: announcement, state, activity and
/// deletions targeting it.
fn repo_filters(addr: &RepoAddr) -> Vec<Filter> {
let mut filters = vec![
filters::announcement(addr),
filters::state(addr),
filters::activity(addr),
];
// Deletion requests (NIP-09/62) must be known before any event of
// this repository can be shown.
filters.extend(filters::deletions_for_repo(addr));
filters
}
/// Fetch this repository's events from the relays announced in its
/// NIP-34 `relays` tag. Deduplicated: each relay is only contacted once
/// per store, so refreshes after the first are no-ops unless the
/// announcement lists new relays.
fn connect_announced_relays(&mut self, relays: &[RelayUrl], cx: &mut Context<Self>) {
let new: Vec<RelayUrl> = relays
.iter()
.filter(|url| !self.repo_relays.contains(*url))
.cloned()
.collect();
if new.is_empty() {
return;
}
self.repo_relays.extend(new.iter().cloned());
let backend = Backend::global(cx);
let addr = self.addr.clone();
backend.update(cx, |backend, cx| {
backend.connect_repo_relays(new, Self::repo_filters(&addr), cx);
});
}
/// Fetch this repository's events from the bootstrap relays (one-shot,
/// auto-closing subscription).
fn subscribe_remote(&mut self, cx: &mut Context<Self>) {
@@ -105,15 +159,7 @@ impl RepoStore {
let addr = self.addr.clone();
backend.update(cx, |backend, cx| {
let mut repo_filters = vec![
filters::announcement(&addr),
filters::state(&addr),
filters::activity(&addr),
];
// Deletion requests (NIP-09/62) must be known before any
// event of this repository can be shown.
repo_filters.extend(filters::deletions_for_repo(&addr));
backend.subscribe_bootstrap(repo_filters, cx);
backend.subscribe_bootstrap(Self::repo_filters(&addr), cx);
});
}
@@ -196,6 +242,24 @@ impl RepoStore {
}
}
// NIP-22 comments reference their root via an `E`/`e` tag rather
// than the repository's `a` tag, so query them by the root events
// of this repository.
let db = client.database();
let mut seen_comments: HashSet<EventId> = comments.iter().map(|e| e.id).collect();
let roots = issues
.iter()
.chain(&patches)
.chain(&pull_requests)
.map(|e| e.id);
for filter in filters::comments_for(roots) {
for event in db.query(filter).await? {
if seen_comments.insert(event.id) {
comments.push(event);
}
}
}
sort_newest_first(&mut issues);
sort_newest_first(&mut patches);
sort_newest_first(&mut pull_requests);
@@ -230,6 +294,15 @@ impl RepoStore {
let again = this.update(cx, |this, cx| {
this.announcement = announcement;
// The announcement may list relays for this repository's
// activity; connect to any we haven't fetched from yet.
let relays = this
.announcement
.as_ref()
.map(|a| a.relays.clone())
.unwrap_or_default();
this.connect_announced_relays(&relays, cx);
if let Some((refs, head)) = state {
this.refs = refs;
this.head = head;
@@ -241,6 +314,32 @@ impl RepoStore {
this.comments = comments;
this.statuses = statuses;
// Comments are not addressed to the repository, so fetch
// them by the root events they reference, on the bootstrap
// relays and on the relays this repository announced.
let roots = this
.issues
.iter()
.chain(&this.patches)
.chain(&this.pull_requests)
.map(|e| e.id)
.collect::<HashSet<EventId>>();
let new_roots: Vec<EventId> = roots
.iter()
.filter(|id| !this.comment_roots.contains(id))
.copied()
.collect();
if !new_roots.is_empty() {
this.comment_roots.extend(new_roots.iter().copied());
let comment_filters = filters::comments_for(new_roots);
let announced: Vec<RelayUrl> = this.repo_relays.iter().cloned().collect();
let backend = Backend::global(cx);
backend.update(cx, |backend, cx| {
backend.subscribe_bootstrap(comment_filters.clone(), cx);
backend.connect_repo_relays(announced, comment_filters, cx);
});
}
cx.notify();
this.refreshing = false;