From e72e04c471884a84e3c8dc9665ff517fa5bff547 Mon Sep 17 00:00:00 2001 From: Ren Amamiya Date: Wed, 14 Oct 2026 02:08:20 +0000 Subject: [PATCH] feat: multi-steps push progress (#30) Reviewed-on: https://git.reya.info/reya/signed/pulls/30 --- crates/signed_core/src/lib.rs | 2 - crates/signed_core/src/state.rs | 114 ----- crates/signed_state/src/backend.rs | 284 ++++------- crates/signed_state/src/lib.rs | 2 + crates/signed_state/src/push.rs | 99 +--- crates/signed_state/src/repo.rs | 34 +- crates/signed_state/src/state.rs | 444 ++++++++++++++++++ crates/signed_state/src/sync_status.rs | 14 +- crates/workspace/src/views/dialog_state.rs | 85 +++- .../workspace/src/views/repo/init_dialog.rs | 14 +- crates/workspace/src/views/repo/mod.rs | 59 ++- .../src/views/sidebar/create_repo_dialog.rs | 14 +- 12 files changed, 753 insertions(+), 412 deletions(-) delete mode 100644 crates/signed_core/src/state.rs create mode 100644 crates/signed_state/src/state.rs diff --git a/crates/signed_core/src/lib.rs b/crates/signed_core/src/lib.rs index 66e6812..076deed 100644 --- a/crates/signed_core/src/lib.rs +++ b/crates/signed_core/src/lib.rs @@ -5,7 +5,6 @@ pub mod filters; pub mod git_event; pub mod inbox; pub mod pull_request; -pub mod state; pub mod status; pub mod upstream; @@ -16,6 +15,5 @@ pub use filters::Filters; pub use git_event::GitEvent; pub use inbox::{InboxItem, InboxReadState, ThreadResolver}; pub use pull_request::PullRequest; -pub use state::RepoState; pub use status::RepoStatus; pub use upstream::Upstream; diff --git a/crates/signed_core/src/state.rs b/crates/signed_core/src/state.rs deleted file mode 100644 index f7af5a2..0000000 --- a/crates/signed_core/src/state.rs +++ /dev/null @@ -1,114 +0,0 @@ -use nostr::prelude::*; - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct RepoState { - pub refs: Vec<(String, String)>, - pub head: Option, -} - -impl RepoState { - pub fn build(id: &str, refs: &[(String, String)], head: Option<&str>) -> EventBuilder { - let mut tags: Vec = vec![Tag::identifier(id.to_owned())]; - - for (name, commit) in refs { - tags.push(Tag::parse([name.as_str(), commit.as_str()]).expect("valid ref tag")); - } - - if let Some(head) = head { - tags.push( - Tag::parse(["HEAD", &format!("ref: refs/heads/{head}")]).expect("valid HEAD tag"), - ); - } - - EventBuilder::new(Kind::RepoState, "").tags(tags) - } - - pub fn parse(event: &Event) -> Self { - let mut refs = Vec::new(); - let mut head = None; - - for tag in event.tags.iter() { - 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())); - } - _ => {} - } - } - - Self { refs, head } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - const COMMIT_A: &str = "aa231c4c6a5777dc89b42207b499891a344add5c"; - const COMMIT_B: &str = "59429cfc6cb35b0a1ddace73b5a5c5ed57b8f5ca"; - - fn keys() -> Keys { - Keys::new( - SecretKey::from_hex("0000000000000000000000000000000000000000000000000000000000000001") - .expect("valid secret key"), - ) - } - - fn state_event(tags: &[&[&str]]) -> Event { - let tags: Vec = tags - .iter() - .map(|t| Tag::parse(t.to_vec()).expect("valid tag")) - .collect(); - - EventBuilder::new(Kind::RepoState, "") - .tags(tags) - .finalize(&keys()) - .expect("signed event") - } - - #[test] - fn parses_heads_and_tags() { - let event = state_event(&[ - &["HEAD", "ref: refs/heads/main"], - &["refs/heads/main", COMMIT_A], - &["refs/heads/dev", COMMIT_B], - &["refs/tags/v1.0", COMMIT_A], - ]); - - let state = RepoState::parse(&event); - - assert_eq!(state.head.as_deref(), Some("main")); - assert_eq!( - state.refs, - vec![ - ("refs/heads/main".to_owned(), COMMIT_A.to_owned()), - ("refs/heads/dev".to_owned(), COMMIT_B.to_owned()), - ("refs/tags/v1.0".to_owned(), COMMIT_A.to_owned()), - ] - ); - } - - #[test] - fn build_round_trips_through_parse() { - let refs = [ - ("refs/heads/main".to_owned(), COMMIT_A.to_owned()), - ("refs/heads/dev".to_owned(), COMMIT_B.to_owned()), - ("refs/tags/v1.0".to_owned(), COMMIT_A.to_owned()), - ]; - - let event = RepoState::build("my-repo", &refs, Some("main")) - .finalize(&keys()) - .expect("signed event"); - - assert_eq!(event.kind, Kind::RepoState); - assert_eq!(event.tags.identifier().as_deref(), Some("my-repo")); - - let state = RepoState::parse(&event); - assert_eq!(state.refs, refs); - assert_eq!(state.head.as_deref(), Some("main")); - } -} diff --git a/crates/signed_state/src/backend.rs b/crates/signed_state/src/backend.rs index de1a045..62dc33b 100644 --- a/crates/signed_state/src/backend.rs +++ b/crates/signed_state/src/backend.rs @@ -1,5 +1,5 @@ use std::collections::HashSet; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use std::str::FromStr; use std::time::{Duration, Instant}; @@ -19,8 +19,9 @@ use crate::bootstrap::{ }; use crate::git_store::Mirrors; use crate::inbox::Inbox; -use crate::push::{PushOutcome, grasp_base_url, grasp_clone_url, push_staged_to_grasps}; +use crate::push::{PushOutcome, grasp_base_url, grasp_clone_url}; use crate::repos::RepoListStore; +use crate::state::{GitPushMode, GitRef, PushStepStatus, SyncState}; pub const USER_KEYRING: &str = "Signed Safe Storage"; pub const NOSTR_CONNECT_TIMEOUT: u64 = 60; @@ -35,6 +36,10 @@ pub enum BackendEvent { ProfileUpdates(Vec), RepoUpdates(Vec), Synced, + PushProgress { + addr: RepoAddr, + status: PushStepStatus, + }, Error(String), } @@ -523,6 +528,7 @@ impl Backend { } let owner = public_key.to_bech32().unwrap(); + let addr = RepoAddr::new(public_key, repo_id.clone()); let servers = grasp_servers.clone(); let client = self.client.clone(); let signer = self.signer.clone(); @@ -537,7 +543,7 @@ impl Backend { folder.join(dir_name) }; - cx.spawn(async move |_this, cx| { + cx.spawn(async move |this, cx| { let work = cx.background_spawn({ let destination = destination.clone(); let name = name.clone(); @@ -574,20 +580,29 @@ impl Backend { Some(commit_sha), ); - let event = announce_repository_and_push( - &client, - &signer, - announcement, - &repo_id, - &owner, - &servers, - vec![("refs/heads/main".to_owned(), commit)], - Some("main".to_owned()), - &destination, - cx, - |path, base, owner, repo_id| Repo::open(path)?.push_main(base, owner, repo_id), - ) - .await?; + let state = SyncState { + refs: vec![GitRef { + name: "refs/heads/main".to_owned(), + commit: commit_sha, + }], + head: Some("main".to_owned()), + path: Some(destination.clone()), + mode: GitPushMode::Main, + }; + + let on_step = &mut |status: PushStepStatus, cx: &mut AsyncApp| { + this.update(cx, |_, cx| { + cx.emit(BackendEvent::PushProgress { + addr: addr.clone(), + status, + }); + })?; + Ok::<(), Error>(()) + }; + + let event = + announce_and_push_or_retract(&client, &signer, &state, announcement, cx, on_step) + .await?; let announcement = Announcement::from_event(&event) .ok_or_else(|| anyhow!("failed to parse announcement"))?; @@ -635,19 +650,20 @@ impl Backend { } let owner = public_key.to_bech32().unwrap(); + let addr = RepoAddr::new(public_key, repo_id.clone()); let servers = grasp_servers.clone(); let client = self.client.clone(); let signer = self.signer.clone(); - cx.spawn(async move |_this, cx| { + cx.spawn(async move |this, cx| { let (mut state, euc) = cx .background_spawn({ let path = path.clone(); async move { let repo = Repo::open(&path)?; - let state = repo.ref_state()?; let euc = repo.root_commit()?; + let state = SyncState::from_repo(&path)?; Ok::<_, Error>((state, euc)) } }) @@ -660,28 +676,26 @@ impl Backend { if let Some(branch) = chosen { let wanted = format!("refs/heads/{branch}"); - if state.refs.iter().any(|(name, _)| name == &wanted) { + if state.refs.iter().any(|git_ref| git_ref.name == wanted) { state.head = Some(branch.to_owned()); } } - let euc = euc.and_then(|commit| Sha1Hash::from_str(&commit).ok()); + let euc = euc.and_then(|commit: String| Sha1Hash::from_str(&commit).ok()); let ann = build_announcement(&repo_id, &name, &description, &owner, &servers, euc); - let event = announce_repository_and_push( - &client, - &signer, - ann, - &repo_id, - &owner, - &servers, - state.refs.clone(), - state.head.clone(), - &path, - cx, - |path, base, owner, repo_id| Repo::open(path)?.push_all(base, owner, repo_id), - ) - .await?; + let on_step = &mut |status: PushStepStatus, cx: &mut AsyncApp| { + this.update(cx, |_, cx| { + cx.emit(BackendEvent::PushProgress { + addr: addr.clone(), + status, + }); + })?; + Ok::<(), Error>(()) + }; + + let event = + announce_and_push_or_retract(&client, &signer, &state, ann, cx, on_step).await?; if let Some(base) = servers.first().and_then(grasp_base_url) { let url = format!("{base}/{owner}/{repo_id}.git"); @@ -754,9 +768,6 @@ impl Backend { cx.notify(); }); - let owner = announcement.owner.to_bech32().unwrap(); - let repo_id = announcement.id.clone(); - let relays = announcement.relays.clone(); let client = self.client.clone(); let signer = self.signer.clone(); @@ -774,74 +785,50 @@ impl Backend { let mut state = { let work = cx.background_spawn({ let path = path.clone(); - async move { Repo::open(&path).and_then(|repo| repo.ref_state()) } + async move { SyncState::from_repo(&path) } }); work.await? }; // Keep the announced default branch in `HEAD` when it is among the pushed refs. - let heads: Vec<&str> = state - .refs - .iter() - .filter_map(|(name, _)| name.strip_prefix("refs/heads/")) - .collect(); - if let Some(head) = announced_head - && heads.iter().any(|branch| *branch == head) + && state + .refs + .iter() + .any(|git_ref| git_ref.name.strip_prefix("refs/heads/") == Some(head.as_str())) { state.head = Some(head); } - // Grasp servers authorize a push by the state event they hold in purgatory. - let refs = state.refs.clone(); - let head = state.head.clone(); - - let outcome = if refs.is_empty() { + let outcome = if state.refs.is_empty() { PushOutcome::default() } else { - let push = cx.background_spawn({ - let client = client.clone(); - let signer = signer.clone(); - let path = path.clone(); - let owner = owner.clone(); - let repo_id = repo_id.clone(); - let relays = relays.clone(); - let refs = refs.clone(); - let head = head.clone(); - async move { - push_staged_to_grasps( - &client, - &signer, - &repo_id, - &refs, - head.as_deref(), - &path, - &owner, - &relays, - |path, base, owner, repo_id| { - Repo::open(path)?.push_all(base, owner, repo_id) - }, - ) - .await - } - }); - push.await + // Grasp servers authorize a push by the state event they hold in purgatory. + let builder = state.state_event(&announcement.id); + let event = builder.finalize_async(&signer).await?; + + let on_step = &mut |status: PushStepStatus, cx: &mut AsyncApp| { + this.update(cx, |_, cx| { + cx.emit(BackendEvent::PushProgress { + addr: addr.clone(), + status, + }); + })?; + Ok::<(), Error>(()) + }; + + state + .push_to_grasps(&client, event, &announcement, cx, on_step) + .await? }; - if !refs.is_empty() && outcome.accepted() == 0 { + if !state.refs.is_empty() && outcome.accepted() == 0 { bail!( "could not push the repository to any grasp server: {}", outcome.failure_summary() ); } - // Fan the state out to the relays once a git server holds the objects. - if let Some(state_event) = &outcome.state_event - && let Err(e) = client.send_event(state_event).broadcast().await - { - log::warn!("failed to broadcast repository state: {e}"); - } - Ok(outcome) }) } @@ -1187,113 +1174,44 @@ fn build_announcement( } } -// Announce the repository, then stage the state event and push. -#[allow(clippy::too_many_arguments)] -async fn announce_repository_and_push( +/// Announces a repository and pushes its state to grasp servers, +/// or retracts the announcement if the push fails. +async fn announce_and_push_or_retract( client: &Client, signer: &S, + state: &SyncState, announcement: GitRepositoryAnnouncement, - repo_id: &str, - owner: &str, - servers: &[RelayUrl], - refs: Vec<(String, String)>, - head: Option, - path: &Path, cx: &mut AsyncApp, - push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error> + Send + 'static, + on_step: &mut P, ) -> Result where - S: AsyncGetPublicKey + AsyncSignEvent + Clone, + S: AsyncGetPublicKey + AsyncSignEvent + ?Sized, + P: FnMut(PushStepStatus, &mut AsyncApp) -> Result<(), Error>, { - for url in servers { - if let Err(e) = client.add_relay(url).and_connect().await { - log::warn!("failed to connect grasp relay {url}: {e}"); - } - } + let state_event = state + .state_event(&announcement.id) + .finalize_async(signer) + .await?; + let announcement_event = announcement.finalize_async(signer).await?; - let event = announcement.finalize_async(signer).await?; - let output = client.send_event(&event).broadcast().await?; - - if output.success.is_empty() { - let reasons = output - .failed - .values() - .cloned() - .collect::>() - .join(", "); - bail!("event not accepted by any relay: {reasons}"); - } - - // The state event is the push authorization. - let outcome = if refs.is_empty() { - PushOutcome::default() - } else { - let push = { - let client = client.clone(); - let signer = S::clone(signer); - let repo_id = repo_id.to_owned(); - let owner = owner.to_owned(); - let servers = servers.to_vec(); - let head = head.clone(); - let path = path.to_path_buf(); - cx.background_spawn(async move { - push_staged_to_grasps( - &client, - &signer, - &repo_id, - &refs, - head.as_deref(), - &path, - &owner, - &servers, - push, - ) - .await - }) - }; - push.await - }; - - if outcome.accepted() == 0 { - // Retract the announcement so the repository is not left announced without content. - let retracted = async { - let deletion = EventDeletionRequest::new() - .id(event.id) - .finalize_async(signer) - .await?; - let output = client.send_event(&deletion).broadcast().await?; - - // Without relay acceptance the deletion request stays live. - if output.success.is_empty() { - bail!("deletion event not accepted by any relay"); - } - - Ok::<(), Error>(()) - } - .await; - - return Err(match retracted { - Ok(()) => anyhow!( - "The repository was announced, but the push to every grasp server failed: {}. \ - The announcement has been retracted", - outcome.failure_summary() - ), - Err(e) => anyhow!( - "The repository was announced, but the push to every grasp server failed: {}. \ - Retracting the announcement failed: {e}; delete the repository to clean it up", - outcome.failure_summary() - ), - }); - } - - // Fan the state out to the relays once a git server holds the objects. - if let Some(state_event) = &outcome.state_event - && let Err(e) = client.send_event(state_event).broadcast().await + match state + .announce_and_push(client, announcement_event.clone(), state_event, cx, on_step) + .await { - log::warn!("failed to broadcast repository state: {e}"); - } + Ok(event) => Ok(event), + Err(e) => { + let retracted = + SyncState::retract_announcement(client, signer, announcement_event.id).await; + let retraction = match retracted { + Ok(()) => "The announcement has been retracted".to_owned(), + Err(error) => format!( + "Retracting the announcement failed: {error}; delete the repository to clean it up" + ), + }; - Ok(event) + bail!("{e}. {retraction}"); + } + } } enum UpdateEvent { diff --git a/crates/signed_state/src/lib.rs b/crates/signed_state/src/lib.rs index bcbd7e1..fc47a83 100644 --- a/crates/signed_state/src/lib.rs +++ b/crates/signed_state/src/lib.rs @@ -10,6 +10,7 @@ mod push; mod refresh; mod repo; mod repos; +mod state; mod sync_status; use std::path::{Path, PathBuf}; @@ -29,6 +30,7 @@ pub use repo::RepoStore; pub use repos::RepoListStore; pub use signed_git::{Nip34Binding, Nip34Kind}; use signed_nostr::NostrBackend; +pub use state::{GitPushMode, GitRef, PushStep, PushStepStatus, SyncState}; pub use sync_status::{CheckoutSyncStatus, SyncStatusStore}; pub fn init( diff --git a/crates/signed_state/src/push.rs b/crates/signed_state/src/push.rs index 56f7171..006ff6a 100644 --- a/crates/signed_state/src/push.rs +++ b/crates/signed_state/src/push.rs @@ -1,9 +1,5 @@ -use std::path::Path; - -use anyhow::Error; use nostr::prelude::Url; use nostr_sdk::prelude::*; -use signed_core::RepoState; /// `ws://` grasp servers, like ngit, use `http(s)://`. pub(crate) fn grasp_base_url(relay: &RelayUrl) -> Option { @@ -50,16 +46,16 @@ pub struct GraspServer { } impl GraspServer { - fn ok(relay: RelayUrl) -> Self { + pub(crate) fn ok(relay: RelayUrl) -> Self { Self { relay, reason: None, } } - fn failed(relay: RelayUrl, reason: impl Into) -> Self { + pub(crate) fn failed(relay: &RelayUrl, reason: impl Into) -> Self { Self { - relay, + relay: relay.to_owned(), reason: Some(reason.into()), } } @@ -115,93 +111,4 @@ impl PushOutcome { self.failure_summary() )) } - - /// All staged events carry the same refs; the newest timestamp wins on the relays. - fn keep_newest(&mut self, event: Event) { - if self - .state_event - .as_ref() - .is_none_or(|current| event.created_at > current.created_at) - { - self.state_event = Some(event); - } - } -} - -#[allow(clippy::too_many_arguments)] -pub(crate) async fn push_staged_to_grasps( - client: &Client, - signer: &S, - repo_id: &str, - refs: &[(String, String)], - head: Option<&str>, - path: &Path, - owner: &str, - servers: &[RelayUrl], - push: impl Fn(&Path, &str, &str, &str) -> Result<(), Error>, -) -> PushOutcome -where - S: AsyncGetPublicKey + AsyncSignEvent + ?Sized, -{ - let mut outcome = PushOutcome::default(); - - for relay in servers { - let Some(base) = grasp_base_url(relay) else { - outcome - .servers - .push(GraspServer::failed(relay.clone(), "no domain")); - continue; - }; - - let event = match RepoState::build(repo_id, refs, head) - .finalize_async(signer) - .await - { - Ok(event) => event, - Err(e) => { - let reason = format!("could not sign the state event: {e}"); - outcome - .servers - .push(GraspServer::failed(relay.clone(), reason)); - continue; - } - }; - - // The relay parks the accepted event in purgatory, authorizing the paired git push. - let staging = match client.add_relay(relay).and_connect().await { - Ok(_) => match client.send_event(&event).to([relay.clone()]).await { - Ok(output) if output.success.contains_key(relay) => Ok(()), - Ok(output) => Err(output - .failed - .get(relay) - .cloned() - .unwrap_or_else(|| "relay did not confirm the event".to_owned())), - Err(e) => Err(format!("could not send the state event to {relay}: {e}")), - }, - Err(e) => Err(format!("could not add relay {relay}: {e}")), - }; - - if let Err(reason) = staging { - log::warn!("grasp push failed: {relay}: {reason}"); - outcome - .servers - .push(GraspServer::failed(relay.clone(), reason)); - continue; - } - - match push(path, &base, owner, repo_id) { - Ok(()) => { - outcome.keep_newest(event); - outcome.servers.push(GraspServer::ok(relay.clone())); - } - Err(e) => { - log::warn!("grasp push failed: {relay}: {e}"); - outcome - .servers - .push(GraspServer::failed(relay.clone(), e.to_string())); - } - } - } - - outcome } diff --git a/crates/signed_state/src/repo.rs b/crates/signed_state/src/repo.rs index fba0a34..a1b29d6 100644 --- a/crates/signed_state/src/repo.rs +++ b/crates/signed_state/src/repo.rs @@ -8,8 +8,7 @@ use nostr::event::IntoEventBuilder; use nostr_sdk::prelude::*; use settings::{EventFetchingStrategy, SettingsStore}; use signed_core::{ - Announcement, Deletions, Filters, GitEvent, PullRequest, RepoAddr, RepoState, RepoStatus, - filters, + Announcement, Deletions, Filters, GitEvent, PullRequest, RepoAddr, RepoStatus, filters, }; use signed_git::{Nip34Binding, Repo}; @@ -19,6 +18,7 @@ use crate::checkouts::CheckoutsStore; use crate::patch_series::{MAX_PATCH_EVENT_BYTES, PatchSeries}; use crate::push::{PushOutcome, grasp_base_url, grasp06_prs_url, pr_clone_urls}; use crate::repos::RepoListStore; +use crate::state::{GitRef, PushStepStatus, SyncState}; use crate::sync_status::SyncStatusStore; pub struct RepoStore { @@ -34,7 +34,7 @@ pub struct RepoStore { pub loaded: bool, pub head: Option, // Refs and timestamp of the latest state event, for sync detection. - pub state_refs: Vec<(String, String)>, + pub state_refs: Vec, pub state_at: Option, pub issues: Vec, pub patches: Vec, @@ -48,6 +48,7 @@ pub struct RepoStore { // The repository is out of sync on the rejected servers until republished. pub last_push_warning: Option, pub pushing: bool, + pub push_steps: Vec, pub cloning: bool, // Avoids re-subscribing and re-fetching on every refresh. repo_relays: HashSet, @@ -101,6 +102,7 @@ impl RepoStore { last_warning: None, last_push_warning: None, pushing: false, + push_steps: Vec::new(), cloning: false, repo_relays: HashSet::new(), root_fetches: HashSet::new(), @@ -131,6 +133,7 @@ impl RepoStore { last_warning: None, last_push_warning: None, pushing: false, + push_steps: Vec::new(), cloning: false, repo_relays: HashSet::new(), root_fetches: HashSet::new(), @@ -169,6 +172,14 @@ impl RepoStore { let backend = Backend::global(cx); cx.subscribe(&backend, |this, _backend, event, cx| { + if let BackendEvent::PushProgress { addr, status } = event { + if this.addr.as_ref() == Some(addr) { + this.apply_push_step(status.clone(), cx); + } + + return; + } + let Some(addr) = this.addr.as_ref() else { return; }; @@ -203,6 +214,19 @@ impl RepoStore { self.addr.as_ref() } + fn apply_push_step(&mut self, status: PushStepStatus, cx: &mut Context) { + match self + .push_steps + .iter_mut() + .find(|entry| entry.step() == status.step()) + { + Some(entry) => *entry = status, + None => self.push_steps.push(status), + } + + cx.notify(); + } + pub fn name(&self) -> SharedString { self.announcement .as_ref() @@ -354,7 +378,7 @@ impl RepoStore { let all_states = states.into_iter().filter(|e| !deletions.is_deleted(e)); let state = - utils::latest(all_states).map(|event| (RepoState::parse(&event), event.created_at)); + utils::latest(all_states).map(|event| (SyncState::parse(&event), event.created_at)); let (mut issues, mut patches, mut pull_requests, mut statuses, mut comments) = (Vec::new(), Vec::new(), Vec::new(), Vec::new(), Vec::new()); @@ -1186,6 +1210,7 @@ impl RepoStore { cx: &mut Context, ) -> Task> { self.pushing = true; + self.push_steps.clear(); self.last_error = None; self.last_push_warning = None; cx.notify(); @@ -1195,6 +1220,7 @@ impl RepoStore { this.update(cx, |this, cx| { this.pushing = false; + this.push_steps.clear(); match &result { Ok(outcome) => { diff --git a/crates/signed_state/src/state.rs b/crates/signed_state/src/state.rs new file mode 100644 index 0000000..901651f --- /dev/null +++ b/crates/signed_state/src/state.rs @@ -0,0 +1,444 @@ +use std::path::{Path, PathBuf}; +use std::str::FromStr; + +use anyhow::{Error, anyhow, bail}; +use bitcoin_hashes::sha1::Hash as Sha1Hash; +use gpui::{AppContext, AsyncApp}; +use nostr::prelude::*; +use nostr_sdk::prelude::*; +use signed_core::Announcement; +use signed_git::Repo; + +use crate::push::{GraspServer, PushOutcome, grasp_base_url}; + +/// A git ref with the commit it points at. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GitRef { + /// Full ref name, like `refs/heads/main` or `refs/tags/v1.0`. + pub name: String, + pub commit: Sha1Hash, +} + +impl GitRef { + pub fn new(name: impl Into, commit: &str) -> Result { + let commit = Sha1Hash::from_str(commit).map_err(|_| anyhow!("invalid commit: {commit}"))?; + + Ok(Self { + name: name.into(), + commit, + }) + } +} + +/// How much to push to the grasp servers. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum GitPushMode { + /// Mirror all branches and tags. + #[default] + All, + /// Push `refs/heads/main` only (repo creation). + Main, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SyncState { + pub refs: Vec, + pub head: Option, + /// Local repository to push from; `None` for states parsed from events. + pub path: Option, + pub mode: GitPushMode, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PushStep { + AnnounceRepository, + /// Publish the state event to one grasp server. + StageStateEvent(RelayUrl), + /// Git push to one grasp server. + PushObjects(RelayUrl), + /// Fan the state event out to the relays. + BroadcastStateEvent, +} + +#[derive(Debug, Clone)] +pub enum PushStepStatus { + Running(PushStep), + Done(PushStep), + Failed(PushStep, String), +} + +impl PushStep { + pub fn label(&self) -> String { + match self { + Self::AnnounceRepository => "Announce repository".to_owned(), + Self::StageStateEvent(relay) => format!("Publish state event -> {relay}"), + Self::BroadcastStateEvent => "Publish state event to relays".to_owned(), + Self::PushObjects(relay) => format!( + "Push objects -> {}", + grasp_base_url(relay).unwrap_or_else(|| relay.to_string()) + ), + } + } +} + +impl PushStepStatus { + pub fn step(&self) -> &PushStep { + match self { + Self::Running(step) | Self::Done(step) | Self::Failed(step, _) => step, + } + } +} + +impl SyncState { + pub fn build(id: &str, refs: &[GitRef], head: Option<&str>) -> EventBuilder { + let mut tags: Vec = vec![Tag::identifier(id.to_owned())]; + + for git_ref in refs { + let commit = git_ref.commit.to_string(); + tags.push(Tag::parse([git_ref.name.as_str(), commit.as_str()]).expect("valid ref tag")); + } + + if let Some(head) = head { + tags.push( + Tag::parse(["HEAD", &format!("ref: refs/heads/{head}")]).expect("valid HEAD tag"), + ); + } + + EventBuilder::new(Kind::RepoState, "").tags(tags) + } + + pub fn parse(event: &Event) -> Self { + let mut refs = Vec::new(); + let mut head = None; + + for tag in event.tags.iter() { + match Nip34Tag::parse(tag.as_slice()) { + Ok(Nip34Tag::Head(branch)) => head = Some(branch), + Ok(Nip34Tag::RefHead { branch, commit }) => { + refs.push(GitRef { + name: format!("refs/heads/{branch}"), + commit, + }); + } + Ok(Nip34Tag::RefTag { name, commit }) => { + refs.push(GitRef { + name: format!("refs/tags/{name}"), + commit, + }); + } + _ => {} + } + } + + Self { + refs, + head, + path: None, + mode: GitPushMode::default(), + } + } + + /// Reads the refs and head of a local repository, ready to push. + pub fn from_repo(path: &Path) -> Result { + let repo = Repo::open(path)?; + let state = repo.ref_state()?; + + let refs = state + .refs + .iter() + .map(|(name, commit)| GitRef::new(name, commit)) + .collect::, Error>>()?; + + Ok(Self { + refs, + head: state.head, + path: Some(path.to_path_buf()), + mode: GitPushMode::default(), + }) + } + + pub fn with_mode(mut self, mode: GitPushMode) -> Self { + self.mode = mode; + self + } + + /// State event for these refs; the caller signs it. + pub fn state_event(&self, repo_id: &str) -> EventBuilder { + Self::build(repo_id, &self.refs, self.head.as_deref()) + } + + /// Push the state event and objects to the grasp servers, + /// calling `on_step` for each step. + pub async fn push_to_grasps

( + &self, + client: &Client, + event: Event, + announcement: &Announcement, + cx: &mut AsyncApp, + on_step: &mut P, + ) -> Result + where + P: FnMut(PushStepStatus, &mut AsyncApp) -> Result<(), Error>, + { + let path = self + .path + .as_deref() + .ok_or_else(|| anyhow!("push requires a local repository"))?; + + let owner = announcement + .owner + .to_bech32() + .map_err(|error| anyhow!("could not encode the repository owner: {error}"))?; + + let mut outcome = PushOutcome::default(); + + for relay in &announcement.relays { + let step = PushStep::StageStateEvent(relay.clone()); + + // The relay must accept the state event before the git push is allowed. + on_step(PushStepStatus::Running(step.clone()), cx)?; + + // Get base URL for the git push + let base = grasp_base_url(relay).unwrap_or_default(); + + // Ensure the relay is connected before sending the event. + client.add_relay(relay).and_connect().await?; + + let staging = client.send_event(&event).to([relay.clone()]).await; + // Whether the relay accepted the event + let sent_failed = !staging + .as_ref() + .is_ok_and(|output| output.success.contains_key(relay)); + + if sent_failed { + let reason = match &staging { + Ok(output) => output + .failed + .get(relay) + .cloned() + .unwrap_or_else(|| "relay did not confirm the event".to_owned()), + Err(e) => format!("could not send the state event to {relay}: {e}"), + }; + + // Run the callback to report the failure + on_step(PushStepStatus::Failed(step, reason.clone()), cx)?; + + // Record the failure in the outcome and continue to the next relay + outcome.servers.push(GraspServer::failed(relay, reason)); + continue; + } + + // Run the callback to report the success + on_step(PushStepStatus::Done(step), cx)?; + + // Run the git push in the background, reporting progress through the callback + let step = PushStep::PushObjects(relay.clone()); + on_step(PushStepStatus::Running(step.clone()), cx)?; + + let pushed = cx.background_spawn({ + let path = path.to_path_buf(); + let base = base.clone(); + let owner = owner.clone(); + let repo_id = announcement.id.clone(); + let mode = self.mode; + + async move { + let repo = Repo::open(&path)?; + match mode { + GitPushMode::Main => repo.push_main(&base, &owner, &repo_id), + GitPushMode::All => repo.push_all(&base, &owner, &repo_id), + } + } + }); + + match pushed.await { + Ok(()) => { + // Run the callback to report the success + on_step(PushStepStatus::Done(step), cx)?; + + outcome.state_event = Some(event.clone()); + outcome.servers.push(GraspServer::ok(relay.clone())); + } + Err(e) => { + // Run the callback to report the failure + on_step(PushStepStatus::Failed(step, e.to_string()), cx)?; + + outcome + .servers + .push(GraspServer::failed(relay, e.to_string())); + } + } + } + + // Fan the state out to the relays once a git server holds the objects. + if outcome.state_event.is_some() { + let step = PushStep::BroadcastStateEvent; + on_step(PushStepStatus::Running(step.clone()), cx)?; + + if let Err(e) = client.send_event(&event).broadcast().await { + on_step(PushStepStatus::Failed(step, e.to_string()), cx)?; + log::warn!("failed to broadcast repository state: {e}"); + } else { + on_step(PushStepStatus::Done(step), cx)?; + } + } + + Ok(outcome) + } + + /// Announce the repository and push the state and objects to the grasp servers. + /// + /// Both events are signed by the caller, so a failure at the call site can retract the announcement by id. + pub async fn announce_and_push

( + &self, + client: &Client, + announcement: Event, + state_event: Event, + cx: &mut AsyncApp, + on_step: &mut P, + ) -> Result + where + P: FnMut(PushStepStatus, &mut AsyncApp) -> Result<(), Error>, + { + let step = PushStep::AnnounceRepository; + on_step(PushStepStatus::Running(step.clone()), cx)?; + + let published = Announcement::from_event(&announcement) + .ok_or_else(|| anyhow!("failed to parse announcement"))?; + + // Ensure the relays are connected before sending the announcement event. + for url in &published.relays { + client.add_relay(url).and_connect().await.ok(); + } + + let output = client.send_event(&announcement).broadcast().await?; + + // Handle the case where no relay accepted the announcement event. + if output.success.is_empty() { + let reasons = output + .failed + .values() + .cloned() + .collect::>() + .join(", "); + + // Run the callback to report the failure + let step = PushStepStatus::Failed(step, format!("event not accepted: {reasons}")); + on_step(step, cx)?; + + bail!("event not accepted: {reasons}"); + } + + on_step(PushStepStatus::Done(step), cx)?; + + let outcome = if self.refs.is_empty() { + PushOutcome::default() + } else { + self.push_to_grasps(client, state_event, &published, cx, on_step) + .await? + }; + + if outcome.accepted() == 0 { + bail!( + "Repository was announced, but the push to grasp servers failed: {}", + outcome.failure_summary() + ); + } + + Ok(announcement) + } + + /// Retracts a live announcement; call when `announce_and_push` fails. + pub async fn retract_announcement( + client: &Client, + signer: &S, + id: EventId, + ) -> Result<(), Error> + where + S: AsyncGetPublicKey + AsyncSignEvent + ?Sized, + { + let builder = EventDeletionRequest::new().id(id); + let deletion = builder.finalize_async(signer).await?; + let output = client.send_event(&deletion).broadcast().await?; + + // Without relay acceptance the deletion request stays live. + if output.success.is_empty() { + bail!("deletion event not accepted by any relay"); + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const COMMIT_A: &str = "aa231c4c6a5777dc89b42207b499891a344add5c"; + const COMMIT_B: &str = "59429cfc6cb35b0a1ddace73b5a5c5ed57b8f5ca"; + + fn git_ref(name: &str, commit: &str) -> GitRef { + GitRef::new(name, commit).expect("valid ref") + } + + fn keys() -> Keys { + Keys::new( + SecretKey::from_hex("0000000000000000000000000000000000000000000000000000000000000001") + .expect("valid secret key"), + ) + } + + fn state_event(tags: &[&[&str]]) -> Event { + let tags: Vec = tags + .iter() + .map(|t| Tag::parse(t.to_vec()).expect("valid tag")) + .collect(); + + EventBuilder::new(Kind::RepoState, "") + .tags(tags) + .finalize(&keys()) + .expect("signed event") + } + + #[test] + fn parses_heads_and_tags() { + let event = state_event(&[ + &["HEAD", "ref: refs/heads/main"], + &["refs/heads/main", COMMIT_A], + &["refs/heads/dev", COMMIT_B], + &["refs/tags/v1.0", COMMIT_A], + ]); + + let state = SyncState::parse(&event); + + assert_eq!(state.head.as_deref(), Some("main")); + assert_eq!( + state.refs, + vec![ + git_ref("refs/heads/main", COMMIT_A), + git_ref("refs/heads/dev", COMMIT_B), + git_ref("refs/tags/v1.0", COMMIT_A), + ] + ); + } + + #[test] + fn build_round_trips_through_parse() { + let refs = vec![ + git_ref("refs/heads/main", COMMIT_A), + git_ref("refs/heads/dev", COMMIT_B), + git_ref("refs/tags/v1.0", COMMIT_A), + ]; + + let event = SyncState::build("my-repo", &refs, Some("main")) + .finalize(&keys()) + .expect("signed event"); + + assert_eq!(event.kind, Kind::RepoState); + assert_eq!(event.tags.identifier().as_deref(), Some("my-repo")); + + let state = SyncState::parse(&event); + assert_eq!(state.refs, refs); + assert_eq!(state.head.as_deref(), Some("main")); + } +} diff --git a/crates/signed_state/src/sync_status.rs b/crates/signed_state/src/sync_status.rs index cfbe4a9..2c26fbb 100644 --- a/crates/signed_state/src/sync_status.rs +++ b/crates/signed_state/src/sync_status.rs @@ -5,13 +5,14 @@ use std::time::Duration; use anyhow::Error; use gpui::{App, AppContext, Context, Entity, Global, Subscription}; use nostr_sdk::prelude::*; -use signed_core::{Deletions, Filters, RepoAddr, RepoState}; +use signed_core::{Deletions, Filters, RepoAddr}; use signed_git::{Repo, RepoSyncStatus}; use crate::backend::{Backend, BackendEvent}; use crate::checkouts::CheckoutsStore; use crate::refresh::{RefreshGate, RefreshRequest}; use crate::repos::RepoListStore; +use crate::state::{GitRef, SyncState}; const SYNC_POLL: Duration = Duration::from_secs(180); const MAX_SYNC_CHECKOUTS: usize = 8; @@ -167,7 +168,7 @@ impl SyncStatusStore { let deletion_events = db.query(Filters::deletions()).await?; let deletions = Deletions::from_events(deletion_events); - let mut states: HashMap> = HashMap::new(); + let mut states: HashMap> = HashMap::new(); for (addr, _) in &targets { let events = db.query(addr.state_filter()).await?; let latest = utils::latest( @@ -178,7 +179,7 @@ impl SyncStatusStore { let Some(event) = latest else { continue; }; - states.insert(addr.clone(), RepoState::parse(&event).refs); + states.insert(addr.clone(), SyncState::parse(&event).refs); } let mut statuses: HashMap> = HashMap::new(); @@ -187,10 +188,15 @@ impl SyncStatusStore { continue; }; + let remote_refs: Vec<(String, String)> = nostr_refs + .iter() + .map(|git_ref| (git_ref.name.clone(), git_ref.commit.to_string())) + .collect(); + let mut list = Vec::new(); for path in paths.iter().take(MAX_SYNC_CHECKOUTS) { let synced = - Repo::try_open(path).and_then(|repo| repo.sync_status(nostr_refs).ok()); + Repo::try_open(path).and_then(|repo| repo.sync_status(&remote_refs).ok()); if let Some(status) = synced { list.push(CheckoutSyncStatus { diff --git a/crates/workspace/src/views/dialog_state.rs b/crates/workspace/src/views/dialog_state.rs index 2eda596..e7e3aa9 100644 --- a/crates/workspace/src/views/dialog_state.rs +++ b/crates/workspace/src/views/dialog_state.rs @@ -1,12 +1,19 @@ use gpui::prelude::*; -use gpui::{AnyElement, App, SharedString, div}; -use gpui_component::ActiveTheme; +use gpui::{AnyElement, App, Context, SharedString, Subscription, div}; +use gpui_component::spinner::Spinner; +use gpui_component::{ActiveTheme, Icon, IconName, Sizable, h_flex, v_flex}; +use signed_core::RepoAddr; +use signed_state::{Backend, BackendEvent, PushStepStatus}; /// Progress of an async dialog action: a busy flag that disables the form and an error shown below it. #[derive(Debug, Default)] pub struct DialogProgress { pub busy: bool, pub error: Option, + // The repository whose push checklist is shown; `None` outside a publish. + push_addr: Option, + pub push_steps: Vec, + _push_subscription: Option, } impl DialogProgress { @@ -16,6 +23,37 @@ impl DialogProgress { self.error = None; } + /// Starts a publish for the repository at `addr` and listens for its push steps. + pub fn begin_push(&mut self, addr: RepoAddr, cx: &mut Context) { + self.begin(); + self.push_addr = Some(addr); + self.push_steps.clear(); + + if self._push_subscription.is_none() { + let backend = Backend::global(cx); + self._push_subscription = Some(cx.subscribe(&backend, |this, _backend, event, cx| { + let BackendEvent::PushProgress { addr, status } = event else { + return; + }; + + if this.push_addr.as_ref() != Some(addr) { + return; + } + + match this + .push_steps + .iter_mut() + .find(|step| step.step() == status.step()) + { + Some(step) => *step = status.clone(), + None => this.push_steps.push(status.clone()), + } + + cx.notify(); + })); + } + } + /// Marks an action as failed, enabling the form and showing `message`. pub fn fail(&mut self, message: impl Into) { self.busy = false; @@ -33,3 +71,46 @@ pub fn error_row(error: &Option, cx: &App) -> Option { .into_any_element() }) } + +/// The push checklist of a running publish, `None` before any step is reported. +pub fn push_steps_row(steps: &[PushStepStatus], cx: &App) -> Option { + if steps.is_empty() { + return None; + } + + let rows = steps + .iter() + .map(|status| { + let label = SharedString::from(status.step().label()); + + match status { + PushStepStatus::Running(_) => h_flex() + .gap_2() + .text_sm() + .text_color(cx.theme().muted_foreground) + .child(Spinner::new().small()) + .child(label) + .into_any_element(), + PushStepStatus::Done(_) => h_flex() + .gap_2() + .text_sm() + .text_color(cx.theme().foreground) + .child(Icon::new(IconName::CircleCheck).small()) + .child(label) + .into_any_element(), + PushStepStatus::Failed(_, reason) => h_flex() + .gap_2() + .text_sm() + .text_color(cx.theme().danger) + .child(Icon::new(IconName::Close).small()) + .child(SharedString::from(format!( + "{}: {reason}", + status.step().label() + ))) + .into_any_element(), + } + }) + .collect::>(); + + Some(v_flex().gap_1().children(rows).into_any_element()) +} diff --git a/crates/workspace/src/views/repo/init_dialog.rs b/crates/workspace/src/views/repo/init_dialog.rs index 8b87a6a..f566408 100644 --- a/crates/workspace/src/views/repo/init_dialog.rs +++ b/crates/workspace/src/views/repo/init_dialog.rs @@ -12,12 +12,13 @@ use gpui_component::input::{Input, InputState, Textarea}; use gpui_component::select::{Select, SelectState}; use gpui_component::{ActiveTheme, Disableable, WindowExt}; use settings::SettingsStore; +use signed_core::RepoAddr; use signed_git::Repo; use signed_state::Backend; use signed_ui::SelectOption; use super::RepoDetailView; -use crate::views::dialog_state::{DialogProgress, error_row}; +use crate::views::dialog_state::{DialogProgress, error_row, push_steps_row}; use crate::views::sidebar::grasp_servers::{ GraspServersState, grasp_servers_field, load_user_grasp_servers, }; @@ -120,6 +121,7 @@ pub fn open( ) .child(grasp_servers_field(&grasp_state, &relay_input, cx)), ) + .children(push_steps_row(&state.read(cx).push_steps, cx)) .children(error_row(&error, cx)) .child( DialogFooter::new().justify_end().child( @@ -189,9 +191,15 @@ fn init_repository( return; } - state.update(cx, |state, _| state.begin()); - let backend = Backend::global(cx); + let Some(user) = backend.read(cx).current_user() else { + state.update(cx, |state, _| state.fail("Sign in to publish a repository")); + return; + }; + + let addr = RepoAddr::new(user, RepoAddr::identifier_from_name(&name)); + state.update(cx, |state, cx| state.begin_push(addr, cx)); + let task = backend.update(cx, |backend, cx| { backend.publish_local_repo( local_path.clone(), diff --git a/crates/workspace/src/views/repo/mod.rs b/crates/workspace/src/views/repo/mod.rs index c750ed5..463fd87 100644 --- a/crates/workspace/src/views/repo/mod.rs +++ b/crates/workspace/src/views/repo/mod.rs @@ -26,7 +26,8 @@ use signed_core::{Announcement, RepoAddr, RepoStatus}; use signed_git::{FileCommit, GitCache, RefSync, Repo}; use signed_state::{ Backend, CheckoutStatus, CheckoutSyncStatus, CheckoutsStore, LocalReposStore, Mirrors, - Nip34Binding, Nip34Kind, ProfileStore, RepoListStore, RepoStore, SyncStatusStore, + Nip34Binding, Nip34Kind, ProfileStore, PushStepStatus, RepoListStore, RepoStore, + SyncStatusStore, }; use signed_ui::{ Avatar, CountBadge, DropdownButton, PixelAvatar, copy_row, menu_copy_row, ref_selector_trigger, @@ -1747,6 +1748,59 @@ impl RepoDetailView { ) } + pub(super) fn render_push_progress_banner(&self, cx: &Context) -> Option { + let store = self.store.read(cx); + if !store.pushing || store.push_steps.is_empty() { + return None; + } + + let steps = store + .push_steps + .iter() + .map(|status| { + let label = SharedString::from(status.step().label()); + + match status { + PushStepStatus::Running(_) => h_flex() + .gap_2() + .text_sm() + .text_color(cx.theme().muted_foreground) + .child(Spinner::new().small()) + .child(label) + .into_any_element(), + PushStepStatus::Done(_) => h_flex() + .gap_2() + .text_sm() + .text_color(cx.theme().foreground) + .child(Icon::new(IconName::CircleCheck).small()) + .child(label) + .into_any_element(), + PushStepStatus::Failed(_, reason) => h_flex() + .gap_2() + .text_sm() + .text_color(cx.theme().danger) + .child(Icon::new(IconName::Close).small()) + .child(SharedString::from(format!( + "{}: {reason}", + status.step().label() + ))) + .into_any_element(), + } + }) + .collect::>(); + + Some( + v_flex() + .p_4() + .gap_1() + .w_full() + .bg(cx.theme().muted) + .child(div().text_sm().child("Pushing")) + .children(steps) + .into_any_element(), + ) + } + pub(super) fn render_push_warning_banner(&self, cx: &Context) -> Option { let store = self.store.read(cx); let warning = store.last_push_warning.clone()?; @@ -2189,6 +2243,9 @@ impl Render for RepoDetailView { .id("repo") .size_full() .when_some(banner, |this, banner| this.child(banner)) + .when_some(self.render_push_progress_banner(cx), |this, banner| { + this.child(banner) + }) .when_some(self.render_push_warning_banner(cx), |this, banner| { this.child(banner) }) diff --git a/crates/workspace/src/views/sidebar/create_repo_dialog.rs b/crates/workspace/src/views/sidebar/create_repo_dialog.rs index 9cc933e..9eb7fcd 100644 --- a/crates/workspace/src/views/sidebar/create_repo_dialog.rs +++ b/crates/workspace/src/views/sidebar/create_repo_dialog.rs @@ -10,11 +10,12 @@ use gpui_component::form::{field, v_form}; use gpui_component::input::{Input, InputState, Textarea}; use gpui_component::{Disableable, IconName, WindowExt, h_flex}; use settings::SettingsStore; +use signed_core::RepoAddr; use signed_state::{Backend, CheckoutsStore}; use super::super::open_repo_panel; use super::grasp_servers::{GraspServersState, grasp_servers_field, load_user_grasp_servers}; -use crate::views::dialog_state::{DialogProgress, error_row}; +use crate::views::dialog_state::{DialogProgress, error_row, push_steps_row}; /// Progress of the create-repository flow, so async results can be rendered. pub type CreateRepoState = DialogProgress; @@ -112,6 +113,7 @@ pub fn open(dock_area: WeakEntity, window: &mut Window, cx: &mut App) ) .child(grasp_servers_field(&grasp_state, &relay_input, cx)), ) + .children(push_steps_row(&state.read(cx).push_steps, cx)) .children(error_row(&error, cx)) .child( DialogFooter::new().justify_end().child( @@ -209,9 +211,15 @@ fn create_repository( return; } - state.update(cx, |state, _| state.begin()); - let backend = Backend::global(cx); + let Some(user) = backend.read(cx).current_user() else { + state.update(cx, |state, _| state.fail("Sign in to create a repository")); + return; + }; + + let addr = RepoAddr::new(user, RepoAddr::identifier_from_name(&name)); + state.update(cx, |state, cx| state.begin_push(addr, cx)); + let task = backend.update(cx, |backend, cx| { backend.create_repository(&name, &description, folder, servers, cx) });