Merge master into 3ca0-announcements-purgatory

1 files changed, 119 insertions, 23 deletions

diff --git a/src/sync/self_subscriber.rs b/src/sync/self_subscriber.rs
index ab10c49..4d69c9a 100644
--- a/src/sync/self_subscriber.rs
+++ b/src/sync/self_subscriber.rs
@@ -16,6 +16,8 @@ use nostr_sdk::Timestamp;
 use tokio::sync::broadcast::error::RecvError;
 use tokio::sync::{broadcast, mpsc};
 
+use crate::nostr::builder::SharedDatabase;
+
 use super::{AddFilters, RepoSyncIndex, RepoSyncNeeds, SyncLevel};
 
 // =============================================================================
@@ -99,6 +101,8 @@ pub struct SelfSubscriber {
     action_tx: mpsc::Sender<AddFilters>,
     /// Last time we connected - used for since filter on reconnect
     last_connected: Option<Timestamp>,
+    /// Database for querying existing events on startup
+    database: SharedDatabase,
 }
 
 impl SelfSubscriber {
@@ -109,11 +113,13 @@ impl SelfSubscriber {
     /// * `relay_domain` - Our service domain (used for filtering relevant repos)
     /// * `repo_sync_index` - Shared index to update with discovered repos
     /// * `action_tx` - Channel to send AddFilters actions to the SyncManager
+    /// * `database` - Database for querying existing events on startup
     pub fn new(
         own_relay_url: String,
         relay_domain: String,
         repo_sync_index: RepoSyncIndex,
         action_tx: mpsc::Sender<AddFilters>,
+        database: SharedDatabase,
     ) -> Self {
         Self {
             own_relay_url,
@@ -121,6 +127,7 @@ impl SelfSubscriber {
             repo_sync_index,
             action_tx,
             last_connected: None,
+            database,
         }
     }
 
@@ -136,6 +143,102 @@ impl SelfSubscriber {
         }
     }
 
+    /// Load existing events from database on startup
+    ///
+    /// Queries the database with two separate queries to build the initial
+    /// PendingUpdates state. This ensures all repos get Layer 2/3 filters
+    /// created, not just those returned by the WebSocket subscription
+    /// (which has limits on the number of events returned).
+    ///
+    /// Query order:
+    /// 1. First query: Get announcements (30617) to populate repo_sync_index
+    ///    with repos and their relays
+    /// 2. Second query: Get root events (1617/1618/1621) for handle_root_event()
+    ///    to add root event IDs for Layer 3 filter creation
+    ///
+    /// Returns a PendingUpdates containing all repos that need Layer 2/3 filters.
+    async fn load_existing_events(&self) -> PendingUpdates {
+        let mut pending = PendingUpdates::new();
+
+        tracing::info!("Loading all events from database");
+
+        // First query: Get all announcements to populate repo_sync_index
+        let announcement_filter = Filter::new().kind(Kind::GitRepoAnnouncement);
+
+        let announcements = match self.database.query(announcement_filter).await {
+            Ok(events) => {
+                tracing::info!(count = events.len(), "Loaded announcements from database");
+                events
+            }
+            Err(e) => {
+                tracing::error!(
+                    error = %e,
+                    "Failed to query announcements from database"
+                );
+                return pending;
+            }
+        };
+
+        // Process announcements
+        let mut announcements_loaded = 0;
+        for event in announcements.iter() {
+            if let Some(repo_id) = Self::extract_repo_id(event) {
+                let relays = Self::extract_relay_urls(event);
+                pending.add_repo(repo_id, relays, HashSet::new());
+                announcements_loaded += 1;
+            }
+        }
+
+        // Update repo_sync_index with announcements BEFORE querying root events
+        {
+            let mut index = self.repo_sync_index.write().await;
+            for (repo_id, needs) in &pending.repos {
+                let entry = index
+                    .entry(repo_id.clone())
+                    .or_insert_with(|| RepoSyncNeeds {
+                        relays: HashSet::new(),
+                        root_events: HashSet::new(),
+                        sync_level: SyncLevel::StateOnly,
+                    });
+                entry.relays.extend(needs.relays.clone());
+            }
+        }
+
+        // Second query: Get all root events for handle_root_event()
+        let root_filter =
+            Filter::new().kinds(vec![Kind::GitPatch, Kind::GitIssue, Kind::GitPullRequest]);
+
+        let root_events = match self.database.query(root_filter).await {
+            Ok(events) => {
+                tracing::info!(count = events.len(), "Loaded root events from database");
+                events
+            }
+            Err(e) => {
+                tracing::error!(
+                    error = %e,
+                    "Failed to query root events from database"
+                );
+                // Continue with just announcements
+                return pending;
+            }
+        };
+
+        // Process root events
+        let mut root_events_processed = 0;
+        for event in root_events.iter() {
+            self.handle_root_event(event, &mut pending).await;
+            root_events_processed += 1;
+        }
+
+        tracing::info!(
+            announcements_loaded = announcements_loaded,
+            root_events_processed = root_events_processed,
+            "Processed existing events from database"
+        );
+
+        pending
+    }
+
     /// Process a relay pool notification
     ///
     /// Handles incoming events from the subscription, queueing 30617 announcements
@@ -277,33 +380,22 @@ impl SelfSubscriber {
         // Subscribe to announcement and root event kinds
         // Per v4 spec: 30617, 1617, 1618, 1621 (NOT 30618)
         // Plus kind 10317 (User Grasp List) for GRASP discovery
-        // Check if we have a last_connected time for reconnect filtering
-        let filter = if let Some(last) = self.last_connected {
+        let mut filter = Filter::new().kinds(vec![
+            Kind::GitRepoAnnouncement,
+            Kind::GitPatch,
+            Kind::GitIssue,
+            Kind::GitPullRequest,
+            Kind::GitUserGraspList,
+        ]);
+        if let Some(timestamp) = self.last_connected {
             // Quick reconnect - use since filter (15 min buffer)
-            let since = Timestamp::from(last.as_secs().saturating_sub(15 * 60));
+            let since = Timestamp::from(timestamp.as_secs().saturating_sub(15 * 60));
             tracing::debug!(
                 since = %since,
                 "Using since filter for reconnect"
             );
-            Filter::new()
-                .kinds(vec![
-                    Kind::GitRepoAnnouncement, // Repository Announcements
-                    Kind::GitPatch,            // Patches
-                    Kind::GitIssue,            // Issues
-                    Kind::GitPullRequest,      // Pull Requests
-                    Kind::GitUserGraspList,    // User Grasp List
-                ])
-                .since(since)
-        } else {
-            // First connection - no since filter
-            Filter::new().kinds(vec![
-                Kind::GitRepoAnnouncement, // Repository Announcements
-                Kind::GitPatch,            // Patches
-                Kind::GitIssue,            // Issues
-                Kind::GitPullRequest,      // Pull Requests
-                Kind::GitUserGraspList,    // User Grasp List
-            ])
-        };
+            filter = filter.since(since);
+        }
 
         // Update last_connected AFTER creating filter but BEFORE subscribing
         self.last_connected = Some(Timestamp::now());
@@ -324,7 +416,11 @@ impl SelfSubscriber {
 
         let mut notifications = client.notifications();
         let batch_window = Self::get_batch_window();
-        let mut pending = PendingUpdates::new();
+
+        // Load existing events from database on startup
+        // This ensures all repos get Layer 2/3 filters created, not just those
+        // returned by the WebSocket subscription (which has limits)
+        let mut pending = self.load_existing_events().await;
 
         // Timer does NOT reset on new events - use interval
         let mut timer = tokio::time::interval(batch_window);