328 lines
10 KiB
Rust
328 lines
10 KiB
Rust
use std::time::Duration;
|
|
|
|
use anyhow::Error;
|
|
use gpui::{AppContext, Context, Subscription, Task};
|
|
use nostr_sdk::prelude::*;
|
|
use signed_core::{Announcement, RepoAddr, RepoStatus, filters, parse_state};
|
|
|
|
use crate::backend::{Backend, BackendEvent};
|
|
|
|
/// Delay between a refresh request and the actual re-query, so bursts of
|
|
/// events (e.g. per-event `NostrUpdate`s) collapse into one query.
|
|
const REFRESH_DEBOUNCE: Duration = Duration::from_millis(300);
|
|
|
|
/// Per-repository store: announcement, state, issues, patches, PRs and
|
|
/// their resolved statuses. Always derived from the local database.
|
|
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>,
|
|
pub patches: Vec<Event>,
|
|
pub pull_requests: Vec<Event>,
|
|
statuses: Vec<Event>,
|
|
/// Error of the last action initiated from this store, if any.
|
|
pub last_error: Option<String>,
|
|
refreshing: bool,
|
|
refresh_dirty: bool,
|
|
/// A refresh is waiting out [`REFRESH_DEBOUNCE`].
|
|
debouncing: bool,
|
|
tasks: Vec<Task<Result<(), Error>>>,
|
|
_subscription: Subscription,
|
|
}
|
|
|
|
impl RepoStore {
|
|
pub fn new(addr: RepoAddr, cx: &mut Context<Self>) -> Self {
|
|
let backend = Backend::global(cx);
|
|
|
|
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);
|
|
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.public_key;
|
|
let coordinate = event.tags.coordinates().into_iter().any(|c| c == this.addr);
|
|
|
|
coordinate || (kind && author)
|
|
}
|
|
_ => false,
|
|
};
|
|
|
|
if relevant {
|
|
this.refresh(cx);
|
|
}
|
|
});
|
|
|
|
let mut store = Self {
|
|
addr,
|
|
announcement: None,
|
|
refs: Vec::new(),
|
|
head: None,
|
|
issues: Vec::new(),
|
|
patches: Vec::new(),
|
|
pull_requests: Vec::new(),
|
|
statuses: Vec::new(),
|
|
last_error: None,
|
|
refreshing: false,
|
|
refresh_dirty: false,
|
|
debouncing: false,
|
|
_subscription: subscription,
|
|
tasks: Vec::new(),
|
|
};
|
|
|
|
store.subscribe_remote(cx);
|
|
store.refresh(cx);
|
|
store
|
|
}
|
|
|
|
pub fn addr(&self) -> &RepoAddr {
|
|
&self.addr
|
|
}
|
|
|
|
/// Fetch this repository's events from the bootstrap relays (one-shot,
|
|
/// auto-closing subscription).
|
|
fn subscribe_remote(&mut self, cx: &mut Context<Self>) {
|
|
let addr = self.addr.clone();
|
|
|
|
Backend::global(cx).update(cx, |backend, cx| {
|
|
backend.subscribe_bootstrap(
|
|
vec![
|
|
filters::announcement(&addr),
|
|
filters::state(&addr),
|
|
filters::activity(&addr),
|
|
],
|
|
cx,
|
|
);
|
|
});
|
|
}
|
|
|
|
/// Re-query the local database and update all fields.
|
|
///
|
|
/// Debounced: a short delay collapses bursts of requests (e.g. per-event
|
|
/// `NostrUpdate`s), and requests that arrive while a query is running are
|
|
/// folded into one follow-up query. 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<Self>) {
|
|
if self.refreshing {
|
|
self.refresh_dirty = true;
|
|
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);
|
|
})
|
|
});
|
|
|
|
self.tasks.push(task);
|
|
}
|
|
|
|
/// One query + apply cycle (debounced entry point).
|
|
fn run_refresh(&mut self, cx: &mut Context<Self>) {
|
|
self.refreshing = true;
|
|
|
|
let client = Backend::global(cx).read(cx).client();
|
|
let addr = self.addr.clone();
|
|
|
|
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?;
|
|
|
|
Ok::<_, Error>((announcements, states, activity))
|
|
}
|
|
.await?;
|
|
|
|
let (announcements, states, activity) = queries;
|
|
|
|
// 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);
|
|
|
|
let state = latest(states).map(|state| parse_state(&state));
|
|
|
|
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 => 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),
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
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(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.
|
|
pub fn status_of(&self, root: &Event) -> RepoStatus {
|
|
let maintainers = self
|
|
.announcement
|
|
.as_ref()
|
|
.map(|a| a.maintainers.as_slice())
|
|
.unwrap_or(&[]);
|
|
|
|
let events = self
|
|
.statuses
|
|
.iter()
|
|
.filter(|e| signed_core::references_root(e, &root.id));
|
|
|
|
signed_core::resolve_status(events, &root.pubkey, maintainers)
|
|
}
|
|
|
|
/// Open an issue on this repository.
|
|
pub fn open_issue(&mut self, subject: Option<String>, content: String, cx: &mut Context<Self>) {
|
|
let builder = GitIssue {
|
|
repository: self.addr.clone(),
|
|
content,
|
|
subject,
|
|
labels: Vec::new(),
|
|
}
|
|
.into_event_builder();
|
|
|
|
self.send(builder, cx);
|
|
}
|
|
|
|
/// Send a root patch (`git format-patch` output) to this repository.
|
|
pub fn send_root_patch(&mut self, patch: String, cx: &mut Context<Self>) {
|
|
let Ok(root_marker) = Tag::parse(["t", "root"]) else {
|
|
return;
|
|
};
|
|
|
|
let builder = EventBuilder::new(Kind::GitPatch, patch).tags([
|
|
Tag::coordinate(self.addr.clone(), None),
|
|
Tag::public_key(self.addr.public_key),
|
|
root_marker,
|
|
]);
|
|
|
|
self.send(builder, cx);
|
|
}
|
|
|
|
/// Set the status of a root event (requires being the root author or a maintainer).
|
|
pub fn set_status(&mut self, root: &Event, status: RepoStatus, cx: &mut Context<Self>) {
|
|
let Ok(root_ref) = Tag::parse(["e", &root.id.to_hex(), "", "root"]) else {
|
|
return;
|
|
};
|
|
|
|
let builder = EventBuilder::new(status.kind(), "").tags([
|
|
root_ref,
|
|
Tag::public_key(self.addr.public_key),
|
|
Tag::public_key(root.pubkey),
|
|
Tag::coordinate(self.addr.clone(), None),
|
|
]);
|
|
|
|
self.send(builder, cx);
|
|
}
|
|
|
|
fn send(&mut self, builder: EventBuilder, cx: &mut Context<Self>) {
|
|
self.last_error = None;
|
|
|
|
let rx = Backend::global(cx).update(cx, |backend, cx| backend.send(builder, cx));
|
|
|
|
let task = cx.spawn(async move |this, cx| {
|
|
if let Ok(Err(e)) = rx.recv_async().await {
|
|
this.update(cx, |this, cx| {
|
|
this.last_error = Some(e.to_string());
|
|
cx.notify();
|
|
})?;
|
|
}
|
|
|
|
Ok(())
|
|
});
|
|
|
|
self.tasks.push(task);
|
|
}
|
|
}
|
|
|
|
fn latest<I>(events: I) -> Option<Event>
|
|
where
|
|
I: IntoIterator<Item = Event>,
|
|
{
|
|
events.into_iter().max_by_key(|e| e.created_at)
|
|
}
|
|
|
|
fn sort_newest_first(events: &mut [Event]) {
|
|
events.sort_by_key(|e| std::cmp::Reverse(e.created_at));
|
|
}
|