sync: use purgatory

don't save new events destined for purgatory events directly to db
or serve on websockets

don't download events already in purgatory via negentropy sync

3 files changed, 73 insertions, 13 deletions

diff --git a/src/nostr/builder.rs b/src/nostr/builder.rs
index 3d7a0d8..da81e64 100644
--- a/src/nostr/builder.rs
+++ b/src/nostr/builder.rs
@@ -69,6 +69,11 @@ impl Nip34WritePolicy {
         }
     }
 
+    /// Get a reference to the purgatory for read-only access
+    pub fn purgatory(&self) -> &std::sync::Arc<crate::purgatory::Purgatory> {
+        &self.ctx.purgatory
+    }
+
     /// Handle repository announcement event
     async fn handle_announcement(&self, event: &Event) -> WritePolicyResult {
         let event_id_str = event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex());
diff --git a/src/purgatory/mod.rs b/src/purgatory/mod.rs
index 18a55d5..2987f15 100644
--- a/src/purgatory/mod.rs
+++ b/src/purgatory/mod.rs
@@ -19,6 +19,7 @@ pub use types::{PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry};
 
 use dashmap::DashMap;
 use nostr_sdk::prelude::*;
+use std::collections::HashSet;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -253,6 +254,34 @@ impl Purgatory {
         self.pr_events.remove(event_id);
     }
 
+    /// Get all event IDs currently stored in purgatory.
+    ///
+    /// Returns a HashSet of all event IDs for both state events and PR events
+    /// held in purgatory. Useful for negentropy sync to avoid fetching events
+    /// that are already in purgatory awaiting git data.
+    ///
+    /// # Returns
+    /// HashSet of event IDs (as EventId) for all events in purgatory
+    pub fn event_ids(&self) -> HashSet<EventId> {
+        let mut ids = HashSet::new();
+
+        // Collect state event IDs
+        for entry in self.state_events.iter() {
+            for state_entry in entry.value().iter() {
+                ids.insert(state_entry.event.id);
+            }
+        }
+
+        // Collect PR event IDs (only actual events, not placeholders)
+        for entry in self.pr_events.iter() {
+            if let Some(ref event) = entry.value().event {
+                ids.insert(event.id);
+            }
+        }
+
+        ids
+    }
+
     /// Get all PR placeholder event IDs (git-data-first entries without events).
     ///
     /// Returns event IDs for entries where git data arrived before the PR event.
diff --git a/src/sync/mod.rs b/src/sync/mod.rs
index 65affc6..dcdbe3a 100644
--- a/src/sync/mod.rs
+++ b/src/sync/mod.rs
@@ -159,6 +159,8 @@ pub enum ProcessResult {
     Saved,
     /// Event already existed in database
     Duplicate,
+    /// Event added to Purgatory
+    Purgatory,
     /// Event rejected by write policy
     Rejected,
 }
@@ -377,6 +379,8 @@ pub struct SyncManager {
     database: SharedDatabase,
     /// Write policy for validating incoming events
     write_policy: Nip34WritePolicy,
+    /// Purgatory for read-only access to events awaiting git data
+    purgatory: Arc<crate::purgatory::Purgatory>,
     /// Local relay for submitting synced events (enables broadcast to WebSocket subscribers)
     local_relay: LocalRelay,
     /// Configuration reference for sync settings
@@ -425,11 +429,15 @@ impl SyncManager {
         config: &Config,
         sync_metrics: Option<SyncMetrics>,
     ) -> Self {
+        // Extract purgatory from write_policy for read-only access
+        let purgatory = write_policy.purgatory().clone();
+
         Self {
             bootstrap_relay_url,
             service_domain,
             database,
             write_policy,
+            purgatory,
             local_relay,
             config: config.clone(),
             repo_sync_index: Arc::new(RwLock::new(HashMap::new())),
@@ -1621,6 +1629,7 @@ impl SyncManager {
         match result {
             WritePolicyResult::Accept => {
                 // Save event to database
+
                 if let Err(e) = database.save_event(event).await {
                     tracing::error!(
                         event_id = %event.id,
@@ -1644,14 +1653,24 @@ impl SyncManager {
                 );
                 ProcessResult::Saved
             }
-            WritePolicyResult::Reject { message, .. } => {
-                tracing::debug!(
-                    event_id = %event.id,
-                    relay = %relay_url,
-                    reason = %message,
-                    "Event rejected by write policy"
-                );
-                ProcessResult::Rejected
+            WritePolicyResult::Reject { message, status } => {
+                if status {
+                    tracing::debug!(
+                        event_id = %event.id,
+                        kind = %event.kind.as_u16(),
+                        reason = %message,
+                        "Event added to purgatory"
+                    );
+                    ProcessResult::Purgatory
+                } else {
+                    tracing::debug!(
+                        event_id = %event.id,
+                        relay = %relay_url,
+                        reason = %message,
+                        "Event rejected by write policy"
+                    );
+                    ProcessResult::Rejected
+                }
             }
         }
     }
@@ -2172,21 +2191,28 @@ impl SyncManager {
             let mut all_remote_ids = Vec::new();
             let mut failed_count = 0;
 
+            // Get event IDs currently in purgatory to avoid fetching them
+            let purgatory_ids = self.purgatory.event_ids();
+
             for (idx, result) in diff_results {
                 match result {
                     Ok(reconciliation) => {
-                        let remote_count = reconciliation.remote.len();
-                        let local_count = reconciliation.local.len();
+                        let remote_excluding_purgatory_ids: HashSet<EventId> = reconciliation
+                            .remote
+                            .difference(&purgatory_ids)
+                            .cloned()
+                            .collect();
+                        let remote_count = remote_excluding_purgatory_ids.len();
                         tracing::debug!(
                             relay = %relay_url,
                             filter_idx = idx,
                             remote_count = remote_count,
-                            local_count = local_count,
-                            remote_ids = ?reconciliation.remote,
+                            local_count = reconciliation.local.len(),
+                            remote_ids = ?remote_excluding_purgatory_ids,
                             "[DIAG TRACE] ✓ Negentropy diff results for filter {}", idx
                         );
                         if remote_count > 0 {
-                            all_remote_ids.extend(reconciliation.remote.into_iter());
+                            all_remote_ids.extend(remote_excluding_purgatory_ids.into_iter());
                         }
                     }
                     Err(e) => {