Move push flow into SyncState methods with step callbacks

crates/signed_state/src/lib.rs

@@ -30,7 +30,7 @@ pub use repo::RepoStore;
 pub use repos::RepoListStore;
 pub use signed_git::{Nip34Binding, Nip34Kind};
 use signed_nostr::NostrBackend;
-pub use state::{GitRef, SyncState};
+pub use state::{GitPushMode, GitRef, PushStep, PushStepStatus, SyncState};
 pub use sync_status::{CheckoutSyncStatus, SyncStatusStore};
 
 pub fn init(

crates/signed_state/src/push.rs

@@ -51,16 +51,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<String>) -> Self {
+    pub(crate) fn failed(relay: &RelayUrl, reason: impl Into<String>) -> Self {
         Self {
-            relay,
+            relay: relay.to_owned(),
             reason: Some(reason.into()),
         }
     }
@@ -150,7 +150,7 @@ where
         let Some(base) = grasp_base_url(relay) else {
             outcome
                 .servers
-                .push(GraspServer::failed(relay.clone(), "no domain"));
+                .push(GraspServer::failed(relay, "no domain"));
             continue;
         };
 
@@ -161,9 +161,7 @@ where
             Ok(event) => event,
             Err(e) => {
                 let reason = format!("could not sign the state event: {e}");
-                outcome
-                    .servers
-                    .push(GraspServer::failed(relay.clone(), reason));
+                outcome.servers.push(GraspServer::failed(relay, reason));
                 continue;
             }
         };
@@ -183,10 +181,7 @@ where
         };
 
         if let Err(reason) = staging {
-            log::warn!("grasp push failed: {relay}: {reason}");
-            outcome
-                .servers
-                .push(GraspServer::failed(relay.clone(), reason));
+            outcome.servers.push(GraspServer::failed(relay, reason));
             continue;
         }
 
@@ -196,10 +191,9 @@ where
                 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()));
+                    .push(GraspServer::failed(relay, e.to_string()));
             }
         }
     }

crates/signed_state/src/state.rs

@@ -1,8 +1,15 @@
+use std::path::{Path, PathBuf};
 use std::str::FromStr;
 
-use anyhow::{Error, anyhow};
+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)]
@@ -23,10 +30,41 @@ impl GitRef {
     }
 }
 
+/// 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<GitRef>,
     pub head: Option<String>,
+    /// Local repository to push from; `None` for states parsed from events.
+    pub path: Option<PathBuf>,
+    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 SyncState {
@@ -70,7 +108,243 @@ impl SyncState {
             }
         }
 
-        Self { refs, head }
+        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<Self, Error> {
+        let repo = Repo::open(path)?;
+        let state = repo.ref_state()?;
+
+        let refs = state
+            .refs
+            .iter()
+            .map(|(name, commit)| GitRef::new(name, commit))
+            .collect::<Result<Vec<_>, 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<P>(
+        &self,
+        client: &Client,
+        event: Event,
+        announcement: &Announcement,
+        cx: &mut AsyncApp,
+        on_step: &mut P,
+    ) -> Result<PushOutcome, Error>
+    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<P>(
+        &self,
+        client: &Client,
+        announcement: Event,
+        state_event: Event,
+        cx: &mut AsyncApp,
+        on_step: &mut P,
+    ) -> Result<Event, Error>
+    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::<Vec<String>>()
+                .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<S>(
+        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(())
     }
 }