initial backend
This commit is contained in:
@@ -0,0 +1,19 @@
|
||||
[package]
|
||||
name = "signed_state"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
publish.workspace = true
|
||||
|
||||
[dependencies]
|
||||
signed_core = { path = "../signed_core" }
|
||||
signed_nostr = { path = "../signed_nostr" }
|
||||
|
||||
nostr.workspace = true
|
||||
nostr-sdk.workspace = true
|
||||
|
||||
gpui.workspace = true
|
||||
flume.workspace = true
|
||||
anyhow.workspace = true
|
||||
|
||||
[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
|
||||
rustls = "0.23"
|
||||
@@ -0,0 +1,195 @@
|
||||
use anyhow::Error;
|
||||
use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Task};
|
||||
use nostr_sdk::prelude::*;
|
||||
use signed_nostr::{NostrBackend, UniversalSigner, pump::Update};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum BackendEvent {
|
||||
/// User has no signer configured.
|
||||
NoSigner,
|
||||
/// The signer has changed (login/logout/account switch).
|
||||
SignerChanged,
|
||||
/// Relay bootstrap finished.
|
||||
Connected,
|
||||
/// A new event was received from a relay and stored in the database.
|
||||
NostrUpdate(Update),
|
||||
/// An event built locally was signed, broadcast and stored.
|
||||
Published(Box<Event>),
|
||||
/// An error occurred.
|
||||
Error(String),
|
||||
}
|
||||
|
||||
impl BackendEvent {
|
||||
pub fn error<T>(error: T) -> Self
|
||||
where
|
||||
T: Into<String>,
|
||||
{
|
||||
Self::Error(error.into())
|
||||
}
|
||||
}
|
||||
|
||||
/// Global backend entity: owns the nostr client, the signer and the
|
||||
/// notification pump. Stores subscribe to [`BackendEvent`] and re-query the
|
||||
/// local database when relevant updates arrive.
|
||||
pub struct Backend {
|
||||
inner: NostrBackend,
|
||||
current_user: Option<PublicKey>,
|
||||
tasks: Vec<Task<Result<(), Error>>>,
|
||||
}
|
||||
|
||||
struct GlobalBackend(Entity<Backend>);
|
||||
|
||||
impl Global for GlobalBackend {}
|
||||
|
||||
impl EventEmitter<BackendEvent> for Backend {}
|
||||
|
||||
impl Backend {
|
||||
/// Retrieve the global backend.
|
||||
pub fn global(cx: &App) -> Entity<Self> {
|
||||
cx.global::<GlobalBackend>().0.clone()
|
||||
}
|
||||
|
||||
pub(crate) fn set_global(entity: Entity<Self>, cx: &mut App) {
|
||||
cx.set_global(GlobalBackend(entity));
|
||||
}
|
||||
|
||||
pub(crate) fn new(inner: NostrBackend, cx: &mut Context<Self>) -> Self {
|
||||
// Pump: relays -> LMDB (automatic) -> flume -> BackendEvent::NostrUpdate.
|
||||
let (tx, rx) = flume::bounded::<Update>(4096);
|
||||
let client = inner.client();
|
||||
|
||||
let pump = cx.background_spawn(async move {
|
||||
signed_nostr::pump::run(client, tx).await;
|
||||
Ok(())
|
||||
});
|
||||
|
||||
let forward = cx.spawn(async move |this, cx| {
|
||||
while let Ok(update) = rx.recv_async().await {
|
||||
this.update(cx, |_, cx| cx.emit(BackendEvent::NostrUpdate(update)))?;
|
||||
}
|
||||
Ok(())
|
||||
});
|
||||
|
||||
Self {
|
||||
inner,
|
||||
current_user: None,
|
||||
tasks: vec![pump, forward],
|
||||
}
|
||||
}
|
||||
|
||||
/// Get the nostr client.
|
||||
pub fn client(&self) -> Client {
|
||||
self.inner.client()
|
||||
}
|
||||
|
||||
/// Get the current signer.
|
||||
pub fn signer(&self) -> UniversalSigner {
|
||||
self.inner.signer()
|
||||
}
|
||||
|
||||
/// Get the current user's public key.
|
||||
pub fn current_user(&self) -> Option<PublicKey> {
|
||||
self.current_user
|
||||
}
|
||||
|
||||
/// Update the signer (any type implementing the async signer traits,
|
||||
/// e.g. `Keys`, `NostrConnect`, a browser extension proxy).
|
||||
pub fn set_signer<T>(&mut self, new_signer: T, cx: &mut Context<Self>)
|
||||
where
|
||||
T: AsyncGetPublicKey + AsyncSignEvent + AsyncNip44 + 'static,
|
||||
<T as AsyncGetPublicKey>::Error: std::error::Error + Send + Sync + 'static,
|
||||
<T as AsyncSignEvent>::Error: std::error::Error + Send + Sync + 'static,
|
||||
<T as AsyncNip44>::Error: std::error::Error + Send + Sync + 'static,
|
||||
{
|
||||
let task = cx.spawn(async move |this, cx| {
|
||||
match new_signer.get_public_key_async().await {
|
||||
Ok(public_key) => {
|
||||
this.update(cx, |this, cx| {
|
||||
this.inner.signer().swap_inner(new_signer);
|
||||
this.current_user = Some(public_key);
|
||||
cx.emit(BackendEvent::SignerChanged);
|
||||
cx.notify();
|
||||
})?;
|
||||
}
|
||||
Err(e) => {
|
||||
this.update(cx, |_this, cx| {
|
||||
cx.emit(BackendEvent::error(e.to_string()));
|
||||
})?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
});
|
||||
self.tasks.push(task);
|
||||
}
|
||||
|
||||
/// Add relays and connect to them.
|
||||
pub fn add_relays(&mut self, urls: Vec<String>, cx: &mut Context<Self>) {
|
||||
let backend = self.inner.clone();
|
||||
|
||||
let task = cx.background_spawn(async move {
|
||||
for url in urls {
|
||||
backend.add_relay(&url).await?;
|
||||
}
|
||||
backend.connect().await;
|
||||
Ok::<(), Error>(())
|
||||
});
|
||||
|
||||
self.tasks.push(cx.spawn(async move |this, cx| {
|
||||
match task.await {
|
||||
Ok(()) => {
|
||||
this.update(cx, |_this, cx| cx.emit(BackendEvent::Connected))?;
|
||||
}
|
||||
Err(e) => {
|
||||
this.update(cx, |_this, cx| {
|
||||
cx.emit(BackendEvent::error(e.to_string()))
|
||||
})?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}));
|
||||
}
|
||||
|
||||
/// 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<Self>) {
|
||||
let backend = self.inner.clone();
|
||||
|
||||
let task = cx.background_spawn(async move {
|
||||
backend.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(())
|
||||
}));
|
||||
}
|
||||
|
||||
/// Sign, broadcast and locally store an event. Emits
|
||||
/// [`BackendEvent::Published`] on success so stores can refresh.
|
||||
pub fn send(&mut self, builder: EventBuilder, cx: &mut Context<Self>) {
|
||||
let backend = self.inner.clone();
|
||||
|
||||
let task = cx.background_spawn(async move { backend.send(builder).await });
|
||||
|
||||
self.tasks.push(cx.spawn(async move |this, cx| {
|
||||
match task.await {
|
||||
Ok(event) => {
|
||||
this.update(cx, |_this, cx| {
|
||||
cx.emit(BackendEvent::Published(Box::new(event)));
|
||||
})?;
|
||||
}
|
||||
Err(e) => {
|
||||
this.update(cx, |_this, cx| {
|
||||
cx.emit(BackendEvent::error(e.to_string()))
|
||||
})?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
mod backend;
|
||||
mod repo;
|
||||
mod repo_list;
|
||||
|
||||
pub use backend::{Backend, BackendEvent};
|
||||
pub use repo::RepoStore;
|
||||
pub use repo_list::RepoListStore;
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
use gpui::{App, AppContext, Entity};
|
||||
use signed_nostr::NostrBackend;
|
||||
|
||||
/// Initialize the backend and install it as a global. Call once at startup,
|
||||
/// before opening any window that uses the stores.
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
pub fn init(db_path: impl AsRef<Path>, cx: &mut App) -> Entity<Backend> {
|
||||
// rustls uses the `aws_lc_rs` provider by default; ignore if already installed.
|
||||
rustls::crypto::aws_lc_rs::default_provider()
|
||||
.install_default()
|
||||
.ok();
|
||||
|
||||
let path = db_path.as_ref().to_path_buf();
|
||||
let inner = cx.foreground_executor().block_on(async move {
|
||||
NostrBackend::new(path)
|
||||
.await
|
||||
.expect("failed to initialize nostr backend")
|
||||
});
|
||||
|
||||
let entity = cx.new(|cx| Backend::new(inner, cx));
|
||||
Backend::set_global(entity.clone(), cx);
|
||||
entity
|
||||
}
|
||||
|
||||
/// Initialize the backend with an in-memory database on wasm.
|
||||
#[cfg(target_arch = "wasm32")]
|
||||
pub fn init(cx: &mut App) -> Entity<Backend> {
|
||||
let inner = NostrBackend::new().expect("failed to initialize nostr backend");
|
||||
let entity = cx.new(|cx| Backend::new(inner, cx));
|
||||
Backend::set_global(entity.clone(), cx);
|
||||
entity
|
||||
}
|
||||
@@ -0,0 +1,233 @@
|
||||
use anyhow::Error;
|
||||
use gpui::{Context, Subscription, Task};
|
||||
use nostr_sdk::prelude::*;
|
||||
use signed_core::{Announcement, RepoAddr, RepoStatus, filters};
|
||||
|
||||
use crate::backend::{Backend, BackendEvent};
|
||||
|
||||
/// Per-repository store: announcement, state, issues, patches, PRs and
|
||||
/// their resolved statuses. Always derived from the local database.
|
||||
pub struct RepoStore {
|
||||
addr: RepoAddr,
|
||||
addr_string: String,
|
||||
|
||||
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>,
|
||||
|
||||
_subscription: Subscription,
|
||||
tasks: Vec<Task<Result<(), Error>>>,
|
||||
}
|
||||
|
||||
impl RepoStore {
|
||||
pub fn new(addr: RepoAddr, cx: &mut Context<Self>) -> Self {
|
||||
let addr_string = addr.to_string();
|
||||
|
||||
let subscription = cx.subscribe(&Backend::global(cx), |this, _backend, event, cx| {
|
||||
let relevant = match event {
|
||||
BackendEvent::NostrUpdate(update) => {
|
||||
update.coordinate.as_deref() == Some(this.addr_string.as_str())
|
||||
|| (update.kind == Kind::GitRepoAnnouncement
|
||||
&& update.author == this.addr.owner)
|
||||
}
|
||||
BackendEvent::Published(event) => {
|
||||
event.kind == Kind::GitRepoAnnouncement
|
||||
&& event.pubkey == this.addr.owner
|
||||
|| event.tags.iter().any(|t| {
|
||||
t.kind() == "a" && t.content() == Some(this.addr_string.as_str())
|
||||
})
|
||||
}
|
||||
_ => false,
|
||||
};
|
||||
|
||||
if relevant {
|
||||
this.refresh(cx);
|
||||
}
|
||||
});
|
||||
|
||||
let mut store = Self {
|
||||
addr,
|
||||
addr_string,
|
||||
announcement: None,
|
||||
refs: Vec::new(),
|
||||
head: None,
|
||||
issues: Vec::new(),
|
||||
patches: Vec::new(),
|
||||
pull_requests: Vec::new(),
|
||||
statuses: Vec::new(),
|
||||
_subscription: subscription,
|
||||
tasks: Vec::new(),
|
||||
};
|
||||
|
||||
store.subscribe_remote(cx);
|
||||
store.refresh(cx);
|
||||
store
|
||||
}
|
||||
|
||||
pub fn addr(&self) -> &RepoAddr {
|
||||
&self.addr
|
||||
}
|
||||
|
||||
/// Subscribe the relay pool to this repository's activity.
|
||||
fn subscribe_remote(&mut self, cx: &mut Context<Self>) {
|
||||
let addr = self.addr.clone();
|
||||
|
||||
Backend::global(cx).update(cx, |backend, cx| {
|
||||
backend.subscribe(filters::announcement(&addr), cx);
|
||||
backend.subscribe(filters::state(&addr), cx);
|
||||
backend.subscribe(filters::activity(&addr), cx);
|
||||
});
|
||||
}
|
||||
|
||||
/// Re-query the local database and update all fields.
|
||||
pub fn refresh(&mut self, cx: &mut Context<Self>) {
|
||||
let client = Backend::global(cx).read(cx).client();
|
||||
let addr = self.addr.clone();
|
||||
|
||||
let task = cx.spawn(async move |this, cx| {
|
||||
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?;
|
||||
|
||||
this.update(cx, |this, cx| {
|
||||
this.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;
|
||||
}
|
||||
|
||||
this.issues.clear();
|
||||
this.patches.clear();
|
||||
this.pull_requests.clear();
|
||||
this.statuses.clear();
|
||||
|
||||
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();
|
||||
})?;
|
||||
|
||||
Ok(())
|
||||
});
|
||||
|
||||
self.tasks.push(task);
|
||||
}
|
||||
|
||||
/// 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.coordinate(),
|
||||
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.coordinate(), None),
|
||||
Tag::public_key(self.addr.owner),
|
||||
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.owner),
|
||||
Tag::public_key(root.pubkey),
|
||||
Tag::coordinate(self.addr.coordinate(), None),
|
||||
]);
|
||||
|
||||
self.send(builder, cx);
|
||||
}
|
||||
|
||||
fn send(&self, builder: EventBuilder, cx: &mut Context<Self>) {
|
||||
Backend::global(cx).update(cx, |backend, cx| backend.send(builder, cx));
|
||||
}
|
||||
}
|
||||
|
||||
fn latest(events: Events) -> Option<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));
|
||||
}
|
||||
|
||||
/// Parse a kind `30618` state event into refs and HEAD.
|
||||
fn parse_state(event: &Event) -> (Vec<(String, String)>, Option<String>) {
|
||||
let mut refs = Vec::new();
|
||||
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()));
|
||||
}
|
||||
}
|
||||
|
||||
(refs, head)
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
use std::collections::HashMap;
|
||||
|
||||
use anyhow::Error;
|
||||
use gpui::{Context, Subscription, Task};
|
||||
use nostr_sdk::prelude::*;
|
||||
use signed_core::{Announcement, filters};
|
||||
|
||||
use crate::backend::{Backend, BackendEvent};
|
||||
|
||||
/// Store listing repository announcements (global discovery or per-author).
|
||||
pub struct RepoListStore {
|
||||
pub announcements: Vec<Announcement>,
|
||||
author: Option<PublicKey>,
|
||||
|
||||
_subscription: Subscription,
|
||||
tasks: Vec<Task<Result<(), Error>>>,
|
||||
}
|
||||
|
||||
impl RepoListStore {
|
||||
/// Create a store. If `author` is `None`, all announcements are listed.
|
||||
pub fn new(author: Option<PublicKey>, cx: &mut Context<Self>) -> Self {
|
||||
let subscription = cx.subscribe(&Backend::global(cx), |this, _backend, event, cx| {
|
||||
let relevant = match event {
|
||||
BackendEvent::NostrUpdate(update) => {
|
||||
update.kind == Kind::GitRepoAnnouncement
|
||||
&& this.author.is_none_or(|a| a == update.author)
|
||||
}
|
||||
BackendEvent::Published(event) => {
|
||||
event.kind == Kind::GitRepoAnnouncement
|
||||
&& this.author.is_none_or(|a| a == event.pubkey)
|
||||
}
|
||||
_ => false,
|
||||
};
|
||||
|
||||
if relevant {
|
||||
this.refresh(cx);
|
||||
}
|
||||
});
|
||||
|
||||
let mut store = Self {
|
||||
announcements: Vec::new(),
|
||||
author,
|
||||
_subscription: subscription,
|
||||
tasks: Vec::new(),
|
||||
};
|
||||
|
||||
store.subscribe_remote(cx);
|
||||
store.refresh(cx);
|
||||
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);
|
||||
}
|
||||
|
||||
fn subscribe_remote(&mut self, cx: &mut Context<Self>) {
|
||||
let author = self.author;
|
||||
|
||||
Backend::global(cx).update(cx, |backend, cx| {
|
||||
let filter = match author {
|
||||
Some(a) => filters::announcements_by(a),
|
||||
None => filters::all_announcements(500),
|
||||
};
|
||||
backend.subscribe(filter, cx);
|
||||
});
|
||||
}
|
||||
|
||||
/// Re-query the local database. Latest announcement per repository wins.
|
||||
pub fn refresh(&mut self, cx: &mut Context<Self>) {
|
||||
let client = Backend::global(cx).read(cx).client();
|
||||
let author = self.author;
|
||||
|
||||
let task = cx.spawn(async move |this, cx| {
|
||||
let filter = match author {
|
||||
Some(a) => filters::announcements_by(a),
|
||||
None => filters::all_announcements(500),
|
||||
};
|
||||
|
||||
let events = client.database().query(filter).await?;
|
||||
|
||||
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<Announcement> = by_repo.into_values().collect();
|
||||
announcements.sort_by_key(|a| std::cmp::Reverse(a.created_at));
|
||||
|
||||
this.announcements = announcements;
|
||||
cx.notify();
|
||||
})?;
|
||||
|
||||
Ok(())
|
||||
});
|
||||
|
||||
self.tasks.push(task);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user