Add SyncStatusDirectory for per-surface status caching

crates/signed_state/src/lib.rs

@@ -29,7 +29,7 @@ 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, SyncEvent, SyncStatusStore};
+pub use sync_status::{CheckoutSyncStatus, SyncEvent, SyncStatusDirectory, SyncStatusStore};
 
 pub fn init(
     db_path: impl AsRef<Path>,

crates/signed_state/src/sync_status.rs

@@ -1,5 +1,5 @@
 use std::cell::RefCell;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
 use std::time::Duration;
 
@@ -205,7 +205,15 @@ impl SyncStatusStore {
         });
 
         self.tasks.push(cx.spawn(async move |this, cx| {
-            let updates = work.await?;
+            // Reset the slot even on failure, or every later pass folds forever.
+            let updates = match work.await {
+                Ok(updates) => updates,
+                Err(error) => {
+                    log::warn!("failed to compute nostr sync statuses: {error}");
+                    Vec::new()
+                }
+            };
+
             let mut again = false;
 
             this.update(cx, |this, cx| {
@@ -244,3 +252,116 @@ impl SyncStatusStore {
         }));
     }
 }
+
+/// Sync status cache bound to one panel or screen.
+pub struct SyncStatusDirectory {
+    statuses: HashMap<RepoAddr, Vec<CheckoutSyncStatus>>,
+    pending: RefCell<HashSet<RepoAddr>>,
+    _subscriptions: Vec<Subscription>,
+}
+
+impl SyncStatusDirectory {
+    /// Create a directory, following the sync store's events.
+    pub fn new(cx: &mut App) -> Entity<Self> {
+        let store = SyncStatusStore::global(cx);
+
+        cx.new(|cx| {
+            let subscription =
+                cx.subscribe(&store, |this: &mut Self, _store, event, cx| match event {
+                    SyncEvent::Updated(updates) => this.apply(updates.clone(), cx),
+                });
+
+            Self {
+                statuses: HashMap::new(),
+                pending: RefCell::new(HashSet::new()),
+                _subscriptions: vec![subscription],
+            }
+        })
+    }
+
+    /// Queue a probe for addrs this surface renders.
+    pub fn track(&mut self, addrs: impl IntoIterator<Item = RepoAddr>, cx: &mut Context<Self>) {
+        let missing: Vec<RepoAddr> = addrs
+            .into_iter()
+            .filter(|addr| {
+                !self.statuses.contains_key(addr)
+                    && !self.with_pending(|pending| pending.contains(addr))
+            })
+            .collect();
+
+        if missing.is_empty() {
+            return;
+        }
+
+        self.with_pending(|pending| pending.extend(missing.iter().cloned()));
+
+        let store = SyncStatusStore::global(cx);
+        store.read(cx).probe(missing);
+    }
+
+    /// Statuses for a repo, or empty; a miss queues a probe.
+    pub fn statuses_of(&self, addr: &RepoAddr, cx: &App) -> Vec<CheckoutSyncStatus> {
+        self.queue(addr, cx);
+        self.statuses.get(addr).cloned().unwrap_or_default()
+    }
+
+    /// Commits available locally but missing on Nostr.
+    pub fn unsynced(&self, addr: &RepoAddr, cx: &App) -> usize {
+        self.queue(addr, cx);
+
+        self.statuses
+            .get(addr)
+            .map(|list| {
+                list.iter()
+                    .map(|checkout| checkout.status.ahead_total)
+                    .sum()
+            })
+            .unwrap_or(0)
+    }
+
+    fn queue(&self, addr: &RepoAddr, cx: &App) {
+        if self.with_pending(|pending| pending.insert(addr.clone())) {
+            let store = SyncStatusStore::global(cx);
+            store.read(cx).probe([addr.clone()]);
+        }
+    }
+
+    fn apply(&mut self, updates: Vec<(RepoAddr, Vec<CheckoutSyncStatus>)>, cx: &mut Context<Self>) {
+        if updates.is_empty() {
+            let changed = !self.statuses.is_empty();
+            self.statuses.clear();
+            self.with_pending(|pending| pending.clear());
+
+            if changed {
+                cx.notify();
+            }
+
+            return;
+        }
+
+        let mut changed = false;
+
+        for (addr, list) in updates {
+            let relevant = self.statuses.contains_key(&addr)
+                || self.with_pending(|pending| pending.remove(&addr));
+
+            if !relevant {
+                continue;
+            }
+
+            if self.statuses.get(&addr) != Some(&list) {
+                self.statuses.insert(addr, list);
+                changed = true;
+            }
+        }
+
+        if changed {
+            cx.notify();
+        }
+    }
+
+    fn with_pending<R>(&self, then: impl FnOnce(&mut HashSet<RepoAddr>) -> R) -> R {
+        let mut pending = self.pending.borrow_mut();
+        then(&mut pending)
+    }
+}