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

wip

Browse Source
This commit is contained in:
reya
2026-09-04 16:02:51 +07:00
parent 17de4f6376
commit 1d224218df
38 changed files with 894 additions and 2482 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/signed_state/src/backend.rs
+42 -134
View File
@@ -1,5 +1,6 @@
use std::collections::hash_map::DefaultHasher;
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::hash::{Hash, Hasher};
use std::path::{Path, PathBuf};
use std::str::FromStr;
@@ -45,8 +46,6 @@ pub enum BackendEvent {
PassphraseRequired,
/// The signer changed on login, logout or account switch.
SignerChanged,
/// Relay bootstrap finished.
Connected,
/// A new event was received from a relay and stored in the database.
NostrUpdate(Update),
/// A negentropy sync completed.
@@ -80,7 +79,6 @@ pub struct Backend {
client: Client,
signer: UniversalSigner,
current_user: Option<PublicKey>,
connected: bool,
sync_progress: Option<(u64, u64)>,
/// True when the stored credential is NIP-49 encrypted.
passphrase_required: bool,
@@ -151,7 +149,6 @@ impl Backend {
client,
signer,
current_user: None,
connected: false,
sync_progress: None,
passphrase_required: false,
recent_fetches: HashMap::new(),
@@ -186,11 +183,7 @@ impl Backend {
self.tasks.push(cx.spawn(async move |this, cx| {
match task.await {
Ok(()) => {
this.update(cx, |this, cx| {
this.connected = true;
cx.emit(BackendEvent::Connected);
cx.notify();
})?;
this.update(cx, |_this, cx| cx.notify())?;
}
Err(e) => {
this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
@@ -993,22 +986,14 @@ impl Backend {
self.tasks.push(cx.spawn(async move |this, cx| {
let result = async {
let events = client.fetch_events(filters::grasp_list(public_key)).await?;
let urls: Vec<String> = events
let events: Vec<Event> = client
.fetch_events(filters::grasp_list(public_key))
.await?
.into_iter()
.max_by_key(|e| e.created_at)
.map(|e| {
e.tags
.iter()
.filter(|t| t.kind() == "g")
.filter_map(|t| t.content().map(str::to_owned))
.collect()
})
.unwrap_or_default();
.collect();
for url in urls {
client.add_relay(&url).await.ok();
for url in latest_grasp_list_servers(events) {
client.add_relay(url.as_str()).await.ok();
}
client.connect().await;
@@ -1049,11 +1034,6 @@ impl Backend {
cx.emit(BackendEvent::error(message));
}
/// Whether the relay bootstrap has completed.
pub fn is_connected(&self) -> bool {
self.connected
}
/// Progress of the in-flight negentropy sync, if any.
pub fn sync_progress(&self) -> Option<(u64, u64)> {
self.sync_progress
@@ -1106,11 +1086,7 @@ impl Backend {
self.tasks.push(cx.spawn(async move |this, cx| {
match task.await {
Ok(()) => {
this.update(cx, |this, cx| {
this.connected = true;
cx.emit(BackendEvent::Connected);
cx.notify();
})?;
this.update(cx, |_this, cx| cx.notify())?;
}
Err(e) => {
this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
@@ -1120,43 +1096,6 @@ impl Backend {
}));
}
/// Add discovery-only relays, e.g. NIP-65 indexers, and connect to them.
pub fn add_discovery_relays(&mut self, urls: Vec<String>, cx: &mut Context<Self>) {
let client = self.client.clone();
let task = cx.background_spawn(async move {
for url in urls {
client
.add_relay(&url)
.capabilities(RelayCapabilities::DISCOVERY)
.await?;
}
client.connect().await;
Ok::<(), Error>(())
});
self.tasks.push(cx.spawn(async move |this, cx| {
if let Err(e) = task.await {
this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
}
Ok(())
}));
}
/// Start a persistent subscription.
pub fn subscribe(&mut self, filter: Filter, cx: &mut Context<Self>) {
let client = self.client.clone();
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 {
this.update(cx, |_this, cx| cx.emit(BackendEvent::error(e.to_string())))?;
}
Ok(())
}));
}
/// Whether an identical fetch started within [`FETCH_DEDUP_WINDOW`] is still recent.
///
/// Records the fingerprint when returning `false`, pruning expired entries first.
@@ -1298,44 +1237,11 @@ impl Backend {
let client = self.client.clone();
let signer = self.signer.clone();
cx.spawn(async move |this, cx| {
self.publish_task(cx, async move {
// Sign with the current signer, broadcast and save locally.
// The event is immediately visible to database queries.
let work = cx.background_spawn(async move {
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::<Vec<String>>()
.join(", ");
return Err(anyhow!("event not accepted by any relay: {reasons}"));
}
Ok(event)
});
let result = work.await;
match &result {
Ok(event) => {
this.update(cx, |_this, cx| {
cx.emit(BackendEvent::Published(Box::new(event.clone())));
})
.ok();
}
Err(e) => {
this.update(cx, |_this, cx| {
cx.emit(BackendEvent::error(e.to_string()));
})
.ok();
}
}
result
let event = builder.finalize_async(&signer).await?;
broadcast_event(&client, &event).await
})
}
@@ -1346,25 +1252,17 @@ impl Backend {
cx: &mut Context<Self>,
) -> Task<Result<Event, Error>> {
let client = self.client.clone();
self.publish_task(cx, async move { broadcast_event(&client, &event).await })
}
/// Run `work` in the background, then emit its outcome as a [`BackendEvent`].
fn publish_task(
&mut self,
cx: &mut Context<Self>,
work: impl Future<Output = Result<Event, Error>> + 'static + Send,
) -> Task<Result<Event, Error>> {
cx.spawn(async move |this, cx| {
let work = cx.background_spawn(async move {
let output = client.send_event(&event).await?;
if output.success.is_empty() && !output.failed.is_empty() {
let reasons = output
.failed
.values()
.cloned()
.collect::<Vec<String>>()
.join(", ");
return Err(anyhow!("event not accepted by any relay: {reasons}"));
}
Ok(event.clone())
});
let result = work.await;
let result = cx.background_spawn(work).await;
match &result {
Ok(event) => {
@@ -1385,15 +1283,6 @@ impl Backend {
})
}
/// Publish a NIP-34 repository announcement, kind 30617, with the current signer.
pub fn publish_announcement(
&mut self,
announcement: GitRepositoryAnnouncement,
cx: &mut Context<Self>,
) -> Task<Result<Event, Error>> {
self.send(announcement.into_event_builder(), cx)
}
/// Sign, broadcast and store an event without awaiting the result.
fn send_fire_and_forget(&mut self, builder: EventBuilder, cx: &mut Context<Self>) {
let task = self.send(builder, cx);
@@ -1433,6 +1322,25 @@ impl Backend {
}
}
/// Broadcast an event and fail when no relay accepted it.
///
/// The client stores accepted events locally, visible to database queries.
async fn broadcast_event(client: &Client, event: &Event) -> Result<Event, Error> {
let output = client.send_event(event).await?;
if output.success.is_empty() && !output.failed.is_empty() {
let reasons = output
.failed
.values()
.cloned()
.collect::<Vec<String>>()
.join(", ");
return Err(anyhow!("event not accepted by any relay: {reasons}"));
}
Ok(event.clone())
}
/// Fingerprint of a relay and filter set, for fetch dedup.
///
/// Relays and filters are sorted first, so the fingerprint is order-independent.
@@ -1603,8 +1511,8 @@ fn latest_grasp_list_servers(events: Vec<Event>) -> Vec<RelayUrl> {
.unwrap_or_default()
}
/// Resolve the user's published grasp servers.
pub(crate) async fn user_grasp_list_servers(
/// Resolve the user's published grasp servers from the local database.
pub async fn user_grasp_list_servers(
client: Client,
user: PublicKey,
) -> Result<Vec<RelayUrl>, Error> {
crates/signed_state/src/checkouts.rs
+12 -32
View File
@@ -13,6 +13,7 @@ use signed_core::{Announcement, RepoAddr};
use crate::backend::{Backend, BackendEvent};
use crate::git_store::GitStore;
use crate::local_repos::LocalReposStore;
use crate::refresh::{RefreshGate, RefreshRequest};
use crate::repo_list::RepoListStore;
/// Delay between a refresh request and the actual re-computation.
@@ -83,10 +84,8 @@ pub struct CheckoutsStore {
///
/// A recompute defaults the base the same way.
requested_head: HashMap<RepoAddr, Option<String>>,
refreshing: bool,
refresh_dirty: bool,
/// A refresh is waiting out [`REFRESH_DEBOUNCE`].
debouncing: bool,
/// Refresh coalescing, see [`RefreshGate`].
refresh: RefreshGate,
_subscriptions: Vec<Subscription>,
tasks: Vec<Task<Result<(), Error>>>,
}
@@ -144,9 +143,7 @@ impl CheckoutsStore {
push_requested: HashSet::new(),
push_statuses: Arc::new(HashMap::new()),
requested_head: HashMap::new(),
refreshing: false,
refresh_dirty: false,
debouncing: false,
refresh: RefreshGate::default(),
_subscriptions: subscriptions,
tasks: Vec::new(),
};
@@ -236,22 +233,14 @@ impl CheckoutsStore {
///
/// Requests arriving while a pass runs fold into a follow-up.
pub fn refresh(&mut self, cx: &mut Context<Self>) {
if self.refreshing {
self.refresh_dirty = true;
if self.refresh.request() != RefreshRequest::Schedule {
return;
}
if self.debouncing {
return;
}
self.debouncing = true;
let task = cx.spawn(async move |this, cx| {
cx.background_executor().timer(REFRESH_DEBOUNCE).await;
this.update(cx, |this, cx| {
this.debouncing = false;
this.run_refresh(cx);
})
this.update(cx, |this, cx| this.run_refresh(cx))
});
self.tasks.push(task);
@@ -259,7 +248,7 @@ impl CheckoutsStore {
/// One resolve and apply cycle, the debounced entry point.
fn run_refresh(&mut self, cx: &mut Context<Self>) {
self.refreshing = true;
self.refresh.begin();
// Inputs snapshot, all cheap shared reads.
let records = {
@@ -362,7 +351,7 @@ impl CheckoutsStore {
Err(_) => {
// Git reads are best-effort, keep the last results.
return this.update(cx, |this, _cx| {
this.refreshing = false;
this.refresh.abort();
});
}
};
@@ -373,13 +362,7 @@ impl CheckoutsStore {
this.push_statuses = Arc::new(push_statuses);
cx.notify();
this.refreshing = false;
if this.refresh_dirty {
this.refresh_dirty = false;
true
} else {
false
}
this.refresh.finish()
})?;
if again {
@@ -388,8 +371,8 @@ impl CheckoutsStore {
// Keep the statuses current while any repository panel is open.
this.update(cx, |this, cx| {
if poll && !this.debouncing && !this.refreshing {
this.debouncing = true;
if poll && this.refresh.idle() {
this.refresh.debounce();
// Open panels get the fast cadence.
// Each cycle fetches every watched checkout's remote.
let delay = if this.status_requested.is_empty() {
@@ -400,10 +383,7 @@ impl CheckoutsStore {
let task = cx.spawn(async move |this, cx| {
cx.background_executor().timer(delay).await;
this.update(cx, |this, cx| {
this.debouncing = false;
this.run_refresh(cx);
})
this.update(cx, |this, cx| this.run_refresh(cx))
});
this.tasks.push(task);
crates/signed_state/src/lib.rs
+2 -2
View File
@@ -3,12 +3,13 @@ mod checkouts;
mod git_store;
mod local_repos;
mod profile;
mod refresh;
mod repo;
mod repo_list;
use std::path::{Path, PathBuf};
pub use backend::{Backend, BackendEvent};
pub use backend::{Backend, BackendEvent, user_grasp_list_servers};
pub use checkouts::{CheckoutStatus, CheckoutsStore, pr_proposes_checkout};
pub use git_store::GitStore;
use gpui::{App, AppContext, Entity};
@@ -18,7 +19,6 @@ pub use profile::{Profile, ProfileStore};
pub use repo::RepoStore;
pub use repo_list::{RepoActivityCounts, RepoListStore};
use signed_nostr::new_backend;
pub use utils::shorten_pubkey;
/// Initialize the backend and stores, and install them as globals.
/// Call once at startup, before opening any window that uses the stores.
crates/signed_state/src/refresh.rs
+79
View File
@@ -0,0 +1,79 @@
/// Refresh coalescing shared by the event stores.
///
/// [`crate::RepoStore`], [`crate::RepoListStore`] and [`crate::CheckoutsStore`]
/// re-query their inputs on a debounce timer with the same policy:
/// a request arriving while a run is in flight is folded into a follow-up run,
/// a request arriving while the debounce timer is pending is dropped by it.
#[derive(Debug, Default)]
pub struct RefreshGate {
/// A run is in flight.
running: bool,
/// A request arrived while a run was in flight.
dirty: bool,
/// The debounce timer is pending.
debouncing: bool,
}
/// What a refresh request decided.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RefreshRequest {
/// No run or timer covers the request, start the debounce timer.
Schedule,
/// A run or pending timer already covers the request.
Fold,
}
impl RefreshGate {
/// Whether a run is in flight.
pub fn running(&self) -> bool {
self.running
}
/// Whether the debounce timer is pending.
pub fn debouncing(&self) -> bool {
self.debouncing
}
/// Whether no run is in flight and no timer is pending.
pub fn idle(&self) -> bool {
!self.running && !self.debouncing
}
/// A new refresh request arrived.
///
/// Folded into a follow-up run while one is in flight, dropped while the
/// debounce timer is pending, otherwise starts the timer.
pub fn request(&mut self) -> RefreshRequest {
if self.running {
self.dirty = true;
RefreshRequest::Fold
} else if self.debouncing {
RefreshRequest::Fold
} else {
self.debouncing = true;
RefreshRequest::Schedule
}
}
/// A timer was started without a request, e.g. a poll cycle.
pub fn debounce(&mut self) {
self.debouncing = true;
}
/// The debounce timer fired and the run starts now.
pub fn begin(&mut self) {
self.debouncing = false;
self.running = true;
}
/// The run ended. Whether a request arrived while it ran.
pub fn finish(&mut self) -> bool {
self.running = false;
std::mem::take(&mut self.dirty)
}
/// The run was abandoned, e.g. on error. Pending follow-up requests survive.
pub fn abort(&mut self) {
self.running = false;
}
}
crates/signed_state/src/repo.rs
+15 -166
View File
@@ -9,15 +9,15 @@ use gpui::{AppContext, AsyncApp, Context, SharedString, Subscription, Task, Weak
use nostr::event::IntoEventBuilder;
use nostr_sdk::prelude::*;
use signed_core::{
Announcement, COVER_NOTE_KIND, Deletions, RepoAddr, RepoStatus, build_state, cover_note,
filters, labels_and_subject, parse_state, pull_request_patch, pull_request_patches,
subject_override,
Announcement, Deletions, RepoAddr, RepoStatus, filters, parse_state, pull_request_patch,
pull_request_patches,
};
use crate::backend::{
Backend, BackendEvent, grasp_base_url, grasp06_prs_url, pr_clone_urls, user_grasp_list_servers,
};
use crate::git_store::GitStore;
use crate::refresh::{RefreshGate, RefreshRequest};
/// Delay between a refresh request and the actual re-query.
const REFRESH_DEBOUNCE: Duration = Duration::from_millis(300);
@@ -34,8 +34,6 @@ const MAX_PATCH_EVENT_BYTES: usize = 60 * 1024;
pub struct RepoStore {
addr: RepoAddr,
pub announcement: Option<Announcement>,
/// `(refname, commit-id)` pairs from the latest state announcement.
pub refs: Vec<(String, String)>,
/// Branch pointed to by `HEAD` in the latest state announcement.
pub head: Option<String>,
pub issues: Vec<Event>,
@@ -49,11 +47,6 @@ pub struct RepoStore {
/// Computed with [`Self::status_by_root`] on every refresh.
open_issue_count: usize,
open_pr_count: usize,
/// Kind-1624 cover notes and kind-1985 label events.
///
/// They reference this repository's roots, used by ngit and GitWorkshop.
cover_notes: Vec<Event>,
labels: Vec<Event>,
/// Incremented on every applied refresh.
///
/// Views key their derived-data caches to it instead of recomputing on every render.
@@ -71,12 +64,9 @@ pub struct RepoStore {
/// Root events, issues, patches and PRs, already fetched per root.
///
/// The per-root fetches cover NIP-22 comments and statuses without an `a` tag.
/// Also kind-1624 cover notes and kind-1985 labels.
root_fetches: HashSet<EventId>,
refreshing: bool,
refresh_dirty: bool,
/// A refresh is waiting out [`REFRESH_DEBOUNCE`].
debouncing: bool,
/// Refresh coalescing, see [`RefreshGate`].
refresh: RefreshGate,
tasks: Vec<Task<Result<(), Error>>>,
_subscription: Subscription,
}
@@ -101,10 +91,8 @@ impl RepoStore {
// Status events may omit their `a` tag, NIP-34.
// Any status event may reference a root of this repository.
let status = RepoStatus::from_kind(update.kind).is_some();
// Cover notes and labels carry no `a` tag either.
let annotation = update.kind == COVER_NOTE_KIND || update.kind == Kind::Label;
deletion || coordinate || (author && kind) || comment || status || annotation
deletion || coordinate || (author && kind) || comment || status
}
BackendEvent::Published(event) => {
let kind = event.kind == Kind::GitRepoAnnouncement;
@@ -128,7 +116,6 @@ impl RepoStore {
let mut store = Self {
addr,
announcement: None,
refs: Vec::new(),
head: None,
issues: Vec::new(),
patches: Vec::new(),
@@ -137,16 +124,12 @@ impl RepoStore {
status_by_root: HashMap::new(),
open_issue_count: 0,
open_pr_count: 0,
cover_notes: Vec::new(),
labels: Vec::new(),
version: 0,
last_error: None,
last_warning: None,
repo_relays: HashSet::new(),
root_fetches: HashSet::new(),
refreshing: false,
refresh_dirty: false,
debouncing: false,
refresh: RefreshGate::default(),
_subscription: subscription,
tasks: Vec::new(),
};
@@ -225,22 +208,14 @@ impl RepoStore {
/// Re-query the local database and update all fields.
pub fn refresh(&mut self, cx: &mut Context<Self>) {
if self.refreshing {
self.refresh_dirty = true;
if self.refresh.request() != RefreshRequest::Schedule {
return;
}
if self.debouncing {
return;
}
self.debouncing = true;
let task = cx.spawn(async move |this, cx| {
cx.background_executor().timer(REFRESH_DEBOUNCE).await;
this.update(cx, |this, cx| {
this.debouncing = false;
this.run_refresh(cx);
})
this.update(cx, |this, cx| this.run_refresh(cx))
});
self.tasks.retain(|task| !task.is_ready());
@@ -248,7 +223,7 @@ impl RepoStore {
}
fn run_refresh(&mut self, cx: &mut Context<Self>) {
self.refreshing = true;
self.refresh.begin();
let backend = Backend::global(cx);
let client = backend.read(cx).client();
@@ -282,7 +257,6 @@ impl RepoStore {
let (mut issues, mut patches, mut pull_requests, mut statuses, mut comments) =
(Vec::new(), Vec::new(), Vec::new(), Vec::new(), Vec::new());
let (mut cover_notes, mut labels): (Vec<Event>, Vec<Event>) = (Vec::new(), Vec::new());
for event in activity {
if deletions.is_deleted(&event) {
@@ -338,35 +312,11 @@ impl RepoStore {
// Cover notes, 1624, and label events, 1985, carry no `a` tag.
// Query them per root like comments and statuses.
let mut seen_cover_notes: HashSet<EventId> = cover_notes.iter().map(|e| e.id).collect();
let mut seen_labels: HashSet<EventId> = labels.iter().map(|e| e.id).collect();
let db = client.database();
let roots = issues
.iter()
.chain(&patches)
.chain(&pull_requests)
.map(|e| e.id);
for root in roots {
for event in db.query(filters::annotations_for([root])).await? {
if deletions.is_deleted(&event) {
continue;
}
if event.kind == COVER_NOTE_KIND && seen_cover_notes.insert(event.id) {
cover_notes.push(event);
} else if event.kind == Kind::Label && seen_labels.insert(event.id) {
labels.push(event);
}
}
}
// The events are only stored for interop and nothing displays them.
sort_newest_first(&mut issues);
sort_newest_first(&mut patches);
sort_newest_first(&mut pull_requests);
sort_oldest_first(&mut comments);
sort_newest_first(&mut cover_notes);
sort_newest_first(&mut labels);
// Resolve every root's status once here.
// Render paths do HashMap lookups instead of per-root status scans.
@@ -402,8 +352,6 @@ impl RepoStore {
open_issue_count,
open_pr_count,
comments,
cover_notes,
labels,
))
});
@@ -420,13 +368,11 @@ impl RepoStore {
open_issue_count,
open_pr_count,
comments,
cover_notes,
labels,
) = match work.await {
Ok(data) => data,
Err(e) => {
return this.update(cx, |this, cx| {
this.refreshing = false;
this.refresh.abort();
this.last_error = Some(e.to_string());
cx.notify();
});
@@ -445,8 +391,7 @@ impl RepoStore {
.unwrap_or_default();
this.connect_announced_relays(&relays, cx);
if let Some((refs, head)) = state {
this.refs = refs;
if let Some((_, head)) = state {
this.head = head;
}
@@ -457,11 +402,9 @@ impl RepoStore {
this.status_by_root = status_by_root;
this.open_issue_count = open_issue_count;
this.open_pr_count = open_pr_count;
this.cover_notes = cover_notes;
this.labels = labels;
this.version = this.version.wrapping_add(1);
// Comments, statuses without an `a` tag, cover notes and labels.
// Comments and statuses without an `a` tag.
// None are addressed to the repository.
// Fetch them by the root events they reference.
// Use the bootstrap relays and the relays this repository announced.
@@ -482,11 +425,9 @@ impl RepoStore {
if !new_roots.is_empty() {
this.root_fetches.extend(new_roots.iter().copied());
// Batch the per-root filters.
// One statuses filter and one annotations filter cover all new roots.
// One filter per root costs a negentropy reconciliation per relay.
let mut root_filters = filters::comments_for(new_roots.clone());
root_filters.push(filters::statuses_for(new_roots.iter().copied()));
root_filters.push(filters::annotations_for(new_roots));
let announced: Vec<RelayUrl> = this.repo_relays.iter().cloned().collect();
let backend = Backend::global(cx);
@@ -498,13 +439,7 @@ impl RepoStore {
cx.notify();
this.refreshing = false;
if this.refresh_dirty {
this.refresh_dirty = false;
true
} else {
false
}
this.refresh.finish()
})?;
// Requests that arrived while the refresh was running.
@@ -528,40 +463,6 @@ impl RepoStore {
self.version
}
/// The effective cover note of `root`, kind 1624, if any.
pub fn cover_note_of(&self, root: &Event) -> Option<&Event> {
let maintainers = self
.announcement
.as_ref()
.map(Announcement::effective_maintainers)
.unwrap_or_default();
cover_note(root, &self.cover_notes, &maintainers)
}
/// The effective hashtag labels of `root`.
pub fn labels_of(&self, root: &Event) -> Vec<String> {
let maintainers = self
.announcement
.as_ref()
.map(Announcement::effective_maintainers)
.unwrap_or_default();
let (labels, _) = labels_and_subject(root, &self.labels, &maintainers);
labels
}
/// The effective subject or title override of `root`, if any.
pub fn subject_of(&self, root: &Event) -> Option<String> {
let maintainers = self
.announcement
.as_ref()
.map(Announcement::effective_maintainers)
.unwrap_or_default();
subject_override(root, &self.labels, &maintainers)
}
/// Number of open issues.
///
/// Issues whose resolved status is [`RepoStatus::Open`].
@@ -1067,58 +968,6 @@ impl RepoStore {
self.send(builder, cx);
}
/// Publish a repository state announcement
pub fn publish_state(&mut self, cx: &mut Context<Self>) {
self.last_error = None;
let backend = Backend::global(cx);
let Some(user) = backend.read(cx).current_user() else {
self.last_error = Some("Sign in to publish repository state".into());
cx.notify();
return;
};
if !self.is_author(&user) {
self.last_error = Some("Only the repository owner can publish state".into());
cx.notify();
return;
}
let cache = GitStore::global(cx).cache().clone();
let addr = self.addr.clone();
let clone_urls: Vec<String> = self
.announcement
.as_ref()
.map(|a| a.clone.iter().map(ToString::to_string).collect())
.unwrap_or_default();
let work = cx.background_spawn(async move {
let repo = cache.ensure_clone(&addr, &clone_urls)?;
signed_git::repo_ref_state(&repo)
});
self.tasks.push(cx.spawn(async move |this, cx| {
let state = match work.await {
Ok(state) => state,
Err(e) => {
return this.update(cx, |this, cx| {
this.last_error = Some(e.to_string());
cx.notify();
});
}
};
this.update(cx, |this, cx| {
let builder =
build_state(&this.addr.identifier, &state.refs, state.head.as_deref());
this.send(builder, cx);
})?;
Ok(())
}));
}
/// Merge a pull request.
pub fn merge_pull_request(&mut self, root: &Event, cx: &mut Context<Self>) {
self.last_error = None;
crates/signed_state/src/repo_list.rs
+12 -36
View File
@@ -8,6 +8,7 @@ use nostr_sdk::prelude::*;
use signed_core::{Announcement, Deletions, RepoAddr, filters, repo_addr};
use crate::backend::{Backend, BackendEvent};
use crate::refresh::{RefreshGate, RefreshRequest};
/// Delay between a refresh request and the actual re-query.
///
@@ -53,10 +54,8 @@ pub struct RepoListStore {
/// Used for the Popular ranking of the explore list.
pub counts: Arc<HashMap<RepoAddr, RepoActivityCounts>>,
author: Option<PublicKey>,
refreshing: bool,
refresh_dirty: bool,
/// A refresh is waiting out [`REFRESH_DEBOUNCE`].
debouncing: bool,
/// Refresh coalescing, see [`RefreshGate`].
refresh: RefreshGate,
tasks: Vec<Task<Result<(), Error>>>,
_subscription: Subscription,
}
@@ -118,9 +117,7 @@ impl RepoListStore {
last_activity: Arc::new(HashMap::new()),
counts: Arc::new(HashMap::new()),
author,
refreshing: false,
refresh_dirty: false,
debouncing: false,
refresh: RefreshGate::default(),
_subscription: subscription,
tasks: Vec::new(),
};
@@ -132,13 +129,6 @@ impl RepoListStore {
store
}
/// Scope the list to an author, or clear the scope with `None`.
pub fn set_author(&mut self, author: Option<PublicKey>, cx: &mut Context<Self>) {
self.author = author;
self.subscribe_remote(cx);
self.refresh(cx);
}
/// Negentropy-sync announcements with the bootstrap relays.
fn subscribe_remote(&mut self, cx: &mut Context<Self>) {
let backend = Backend::global(cx);
@@ -160,9 +150,9 @@ impl RepoListStore {
/// Query the local database immediately, no debounce.
/// Stored announcements appear as soon as the app opens.
fn refresh_initial(&mut self, cx: &mut Context<Self>) {
debug_assert!(!self.debouncing);
if self.refreshing {
self.refresh_dirty = true;
debug_assert!(!self.refresh.debouncing());
if self.refresh.running() {
self.refresh.request();
return;
}
self.run_refresh(cx);
@@ -170,22 +160,14 @@ impl RepoListStore {
/// Re-query the local database.
pub fn refresh(&mut self, cx: &mut Context<Self>) {
if self.refreshing {
self.refresh_dirty = true;
if self.refresh.request() != RefreshRequest::Schedule {
return;
}
if self.debouncing {
return;
}
self.debouncing = true;
let task = cx.spawn(async move |this, cx| {
cx.background_executor().timer(REFRESH_DEBOUNCE).await;
this.update(cx, |this, cx| {
this.debouncing = false;
this.run_refresh(cx);
})
this.update(cx, |this, cx| this.run_refresh(cx))
});
self.tasks.push(task);
@@ -193,7 +175,7 @@ impl RepoListStore {
/// One query and apply cycle, the debounced entry point.
fn run_refresh(&mut self, cx: &mut Context<Self>) {
self.refreshing = true;
self.refresh.begin();
let backend = Backend::global(cx);
let client = backend.read(cx).client();
@@ -315,7 +297,7 @@ impl RepoListStore {
// Database errors are transient, keep the last list.
Err(_) => {
return this.update(cx, |this, _cx| {
this.refreshing = false;
this.refresh.abort();
});
}
};
@@ -326,13 +308,7 @@ impl RepoListStore {
this.counts = Arc::new(counts);
cx.notify();
this.refreshing = false;
if this.refresh_dirty {
this.refresh_dirty = false;
true
} else {
false
}
this.refresh.finish()
})?;
// Requests that arrived while the refresh was running.