4 files changed, 265 insertions, 25 deletions

diff --git a/src/sync/algorithms.rs b/src/sync/algorithms.rs
index 39788bc..9899abc 100644
--- a/src/sync/algorithms.rs
+++ b/src/sync/algorithms.rs
@@ -25,8 +25,10 @@ use super::{ConnectionStatus, PendingBatch, RelayState};
 /// this repo need to sync from", it's "what repos does this relay need to sync".
 #[derive(Debug, Clone, Default)]
 pub struct RelaySyncNeeds {
-    /// Repos that need to be synced from this relay
+    /// Repos that need full L2+L3 sync from this relay
     pub repos: HashSet<String>,
+    /// Repos that only need state event sync (purgatory announcements)
+    pub state_only_repos: HashSet<String>,
     /// Root events that need to be tracked from this relay
     pub root_events: HashSet<EventId>,
 }
@@ -67,8 +69,15 @@ pub fn derive_relay_targets(
         for relay_url in &needs.relays {
             let entry = relay_targets.entry(relay_url.clone()).or_default();
 
-            entry.repos.insert(repo_id.clone());
-            entry.root_events.extend(needs.root_events.iter().cloned());
+            match needs.sync_level {
+                super::SyncLevel::Full => {
+                    entry.repos.insert(repo_id.clone());
+                    entry.root_events.extend(needs.root_events.iter().cloned());
+                }
+                super::SyncLevel::StateOnly => {
+                    entry.state_only_repos.insert(repo_id.clone());
+                }
+            }
         }
     }
 
@@ -96,7 +105,7 @@ pub fn compute_actions(
     pending: &HashMap<String, Vec<PendingBatch>>,
     confirmed: &HashMap<String, RelayState>,
 ) -> Vec<AddFilters> {
-    use crate::sync::filters::build_layer2_and_layer3_filters;
+    use crate::sync::filters::build_sync_level_aware_filters;
 
     let mut actions = Vec::new();
 
@@ -140,14 +149,22 @@ pub fn compute_actions(
             .map(|state| state.root_events.clone())
             .unwrap_or_default();
 
-        // Calculate what's NEW (not in pending, not in confirmed)
-        let new_repos: HashSet<String> = target_needs
+        // Calculate what's NEW for full repos (not in pending, not in confirmed)
+        let new_full_repos: HashSet<String> = target_needs
             .repos
             .difference(&pending_repos)
             .filter(|repo| !confirmed_repos.contains(*repo))
             .cloned()
             .collect();
 
+        // Calculate what's NEW for state-only repos
+        let new_state_only_repos: HashSet<String> = target_needs
+            .state_only_repos
+            .difference(&pending_repos)
+            .filter(|repo| !confirmed_repos.contains(*repo))
+            .cloned()
+            .collect();
+
         let new_events: HashSet<EventId> = target_needs
             .root_events
             .difference(&pending_events)
@@ -156,13 +173,23 @@ pub fn compute_actions(
             .collect();
 
         // If there's anything new, create an AddFilters action
-        if !new_repos.is_empty() || !new_events.is_empty() {
-            let filters = build_layer2_and_layer3_filters(&new_repos, &new_events, None);
+        if !new_full_repos.is_empty() || !new_state_only_repos.is_empty() || !new_events.is_empty()
+        {
+            let filters = build_sync_level_aware_filters(
+                &new_full_repos,
+                &new_state_only_repos,
+                &new_events,
+                None,
+            );
+
+            // Combine all repos into pending items (pending tracking doesn't need sync level)
+            let mut all_new_repos = new_full_repos;
+            all_new_repos.extend(new_state_only_repos);
 
             actions.push(AddFilters {
                 relay_url: relay_url.clone(),
                 items: PendingItems {
-                    repos: new_repos,
+                    repos: all_new_repos,
                     root_events: new_events,
                 },
                 filters,
@@ -204,6 +231,7 @@ mod tests {
             ModRepoSyncNeeds {
                 relays,
                 root_events,
+                sync_level: Default::default(),
             },
         );
 
@@ -229,6 +257,7 @@ mod tests {
                 ModRepoSyncNeeds {
                     relays,
                     root_events: HashSet::new(),
+                    sync_level: Default::default(),
                 },
             );
         }
@@ -252,6 +281,7 @@ mod tests {
             ModRepoSyncNeeds {
                 relays,
                 root_events: HashSet::new(),
+                sync_level: Default::default(),
             },
         );
 
@@ -285,6 +315,7 @@ mod tests {
             ModRepoSyncNeeds {
                 relays: relays1,
                 root_events: root_events1,
+                sync_level: Default::default(),
             },
         );
 
@@ -299,6 +330,7 @@ mod tests {
             ModRepoSyncNeeds {
                 relays: relays2,
                 root_events: root_events2,
+                sync_level: Default::default(),
             },
         );
 
@@ -332,6 +364,7 @@ mod tests {
             "wss://relay1.com".to_string(),
             RelaySyncNeeds {
                 repos: vec!["repo1".to_string()].into_iter().collect(),
+                state_only_repos: HashSet::new(),
                 root_events: HashSet::new(),
             },
         );
@@ -366,6 +399,7 @@ mod tests {
             "wss://relay1.com".to_string(),
             RelaySyncNeeds {
                 repos: vec!["repo1".to_string()].into_iter().collect(),
+                state_only_repos: HashSet::new(),
                 root_events: HashSet::new(),
             },
         );
@@ -389,6 +423,7 @@ mod tests {
             "wss://relay1.com".to_string(),
             RelaySyncNeeds {
                 repos: vec!["repo1".to_string()].into_iter().collect(),
+                state_only_repos: HashSet::new(),
                 root_events: HashSet::new(),
             },
         );
@@ -428,6 +463,7 @@ mod tests {
             "wss://relay1.com".to_string(),
             RelaySyncNeeds {
                 repos: vec!["repo1".to_string()].into_iter().collect(),
+                state_only_repos: HashSet::new(),
                 root_events: HashSet::new(),
             },
         );
@@ -465,6 +501,7 @@ mod tests {
             "wss://relay1.com".to_string(),
             RelaySyncNeeds {
                 repos: vec!["repo1".to_string()].into_iter().collect(),
+                state_only_repos: HashSet::new(),
                 root_events: HashSet::new(),
             },
         );
@@ -510,6 +547,7 @@ mod tests {
                 ]
                 .into_iter()
                 .collect(),
+                state_only_repos: HashSet::new(),
                 root_events: HashSet::new(),
             },
         );
@@ -572,6 +610,7 @@ mod tests {
             "wss://relay1.com".to_string(),
             RelaySyncNeeds {
                 repos: HashSet::new(),
+                state_only_repos: HashSet::new(),
                 root_events: vec![event_id].into_iter().collect(),
             },
         );
@@ -599,6 +638,7 @@ mod tests {
             "wss://new-relay.com".to_string(),
             RelaySyncNeeds {
                 repos: vec!["repo1".to_string()].into_iter().collect(),
+                state_only_repos: HashSet::new(),
                 root_events: HashSet::new(),
             },
         );
diff --git a/src/sync/filters.rs b/src/sync/filters.rs
index 3592489..1215e81 100644
--- a/src/sync/filters.rs
+++ b/src/sync/filters.rs
@@ -245,6 +245,37 @@ pub fn build_layer2_and_layer3_filters(
     filters
 }
 
+/// Builds filters respecting SyncLevel for each repo
+///
+/// StateOnly repos only get state event filters (kind 30618).
+/// Full repos get all L2/L3 filters (state + repo-tagging + root event).
+///
+/// # Arguments
+/// * `full_repos` - Repos needing full L2+L3 sync
+/// * `state_only_repos` - Repos needing only state event sync (purgatory)
+/// * `root_events` - Root event IDs (only used for Full repos)
+/// * `since` - Optional timestamp for incremental sync
+pub fn build_sync_level_aware_filters(
+    full_repos: &HashSet<String>,
+    state_only_repos: &HashSet<String>,
+    root_events: &HashSet<EventId>,
+    since: Option<Timestamp>,
+) -> Vec<Filter> {
+    let mut filters = Vec::new();
+
+    // All repos (both Full and StateOnly) need state event filters
+    let all_repos: HashSet<String> = full_repos.union(state_only_repos).cloned().collect();
+    filters.extend(state_event_filters_for_our_repos(&all_repos, since));
+
+    // Only Full repos get repo-tagging and root event filters
+    if !full_repos.is_empty() {
+        filters.extend(tagged_one_of_our_repo_event_filters(full_repos, since));
+    }
+    filters.extend(tagged_one_of_our_root_event_filters(root_events, since));
+
+    filters
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/src/sync/mod.rs b/src/sync/mod.rs
index d6634ff..cd62380 100644
--- a/src/sync/mod.rs
+++ b/src/sync/mod.rs
@@ -85,6 +85,19 @@ use rejected_index::RejectedEventsIndex;
 // Supporting Data Structures
 // =============================================================================
 
+/// Level of sync needed for a repository
+///
+/// Purgatory announcements only need state events synced (to validate git data).
+/// Promoted repos need full L2/L3 sync (patches, issues, PRs, etc.).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum SyncLevel {
+    /// Full L2 + L3 sync (promoted repos with git data)
+    #[default]
+    Full,
+    /// Only state events (kind 30618) - for purgatory announcements
+    StateOnly,
+}
+
 /// What repos and root events need to be synced
 #[derive(Debug, Clone, Default)]
 pub struct RepoSyncNeeds {
@@ -92,6 +105,8 @@ pub struct RepoSyncNeeds {
     pub relays: HashSet<String>,
     /// Root event IDs - 1617/1618/1621 - that reference this repo
     pub root_events: HashSet<EventId>,
+    /// Sync level - StateOnly for purgatory, Full for promoted repos
+    pub sync_level: SyncLevel,
 }
 
 /// Connection status for a relay
@@ -382,6 +397,40 @@ async fn run_daily_timer(
     }
 }
 
+/// Background task that periodically syncs purgatory announcements into repo_sync_index.
+///
+/// Runs every 5 seconds by default (200ms when `NGIT_TEST=1`).
+/// For each announcement currently in purgatory, ensures there is a `StateOnly` entry in
+/// `repo_sync_index`. New entries trigger `handle_new_sync_filters` which connects to the
+/// relay URLs listed in the announcement and subscribes to state events (kind 30618).
+///
+/// This is the sole registration path for purgatory announcements:
+/// - Sync-path announcements: registered here within one interval of arriving.
+/// - User-submitted purgatory announcements: the SelfSubscriber never sees them
+///   (they're rejected from DB), so this timer is the only registration path.
+async fn run_purgatory_announcement_sync(
+    sync_manager: Arc<Mutex<SyncManager>>,
+    mut shutdown_rx: broadcast::Receiver<()>,
+) {
+    let interval = if std::env::var("NGIT_TEST").as_deref() == Ok("1") {
+        Duration::from_millis(200)
+    } else {
+        Duration::from_secs(5)
+    };
+    loop {
+        tokio::select! {
+            _ = tokio::time::sleep(interval) => {
+                let mut manager = sync_manager.lock().await;
+                manager.sync_purgatory_announcements_to_index().await;
+            }
+            _ = shutdown_rx.recv() => {
+                tracing::debug!("Purgatory announcement sync timer received shutdown signal");
+                break;
+            }
+        }
+    }
+}
+
 // Combined Health and Metrics Checker
 
 /// Background task for cleaning up expired entries from the rejected events index
@@ -936,9 +985,29 @@ impl SyncManager {
 
                         // Create REQ+EOSE subscriptions using original semantic filters
                         // This queries by kind/author/tags instead of by ID, which may
-                        // succeed even when ID-based queries fail
-                        let fallback_filters = filters::build_layer2_and_layer3_filters(
-                            &batch_repos,
+                        // succeed even when ID-based queries fail.
+                        // Split batch_repos by SyncLevel to avoid sending Layer 2 filters
+                        // (#a/#A/#q) for StateOnly (purgatory) repos - those PRs would be
+                        // rejected as orphan and then silently dropped by nostr-sdk deduplication.
+                        let (full_repos, state_only_repos) = {
+                            let repo_index = self.repo_sync_index.read().await;
+                            let mut full = HashSet::new();
+                            let mut state_only = HashSet::new();
+                            for repo_ref in &batch_repos {
+                                match repo_index.get(repo_ref).map(|n| n.sync_level) {
+                                    Some(SyncLevel::StateOnly) => {
+                                        state_only.insert(repo_ref.clone());
+                                    }
+                                    _ => {
+                                        full.insert(repo_ref.clone());
+                                    }
+                                }
+                            }
+                            (full, state_only)
+                        };
+                        let fallback_filters = filters::build_sync_level_aware_filters(
+                            &full_repos,
+                            &state_only_repos,
                             &batch_root_events,
                             None,
                         );
@@ -1272,7 +1341,7 @@ impl SyncManager {
     /// to be batched and create Layer 2/3 filters before we mark sync complete.
     ///
     /// The 6-second delay is based on:
-    /// - Self-subscriber batch window: 5 seconds (configurable via NGIT_SYNC_BATCH_WINDOW_MS)
+    /// - Self-subscriber batch window: 5 seconds (200ms when `NGIT_TEST=1`)
     /// - Buffer for processing: 1 second
     ///
     /// Called after each batch is confirmed to detect completion.
@@ -1486,7 +1555,17 @@ impl SyncManager {
             run_rejected_index_cleanup(cleanup_manager, cleanup_shutdown).await;
         });
 
-        // 11. Main loop - handle actions from self-subscriber, disconnect, EOSE, and connect notifications
+        // 11. Spawn purgatory announcement sync timer (every 5s)
+        // Ensures purgatory announcements (including user-submitted ones that never
+        // touch the DB) are registered in repo_sync_index as StateOnly so that
+        // state event subscriptions are established on their listed relay URLs.
+        let purgatory_sync_manager = Arc::clone(&sync_manager);
+        let purgatory_sync_shutdown = shutdown_tx.subscribe();
+        tokio::spawn(async move {
+            run_purgatory_announcement_sync(purgatory_sync_manager, purgatory_sync_shutdown).await;
+        });
+
+        // 12. Main loop - handle actions from self-subscriber, disconnect, EOSE, and connect notifications
         loop {
             // Wait for an event without holding the lock
             tokio::select! {
@@ -1719,6 +1798,10 @@ impl SyncManager {
 
                         // For sync-triggered events that go to purgatory, trigger immediate sync
                         // (instead of the default 3-minute delay for user-submitted events)
+                        //
+                        // Note: announcement events (kind 30617) are registered in repo_sync_index
+                        // by the purgatory announcement sync timer (run_purgatory_announcement_sync)
+                        // rather than inline here.
                         if result == ProcessResult::Purgatory {
                             // State events (kind 30618) - extract identifier and trigger immediate sync
                             if event.kind.as_u16() == 30618 {
@@ -2303,6 +2386,80 @@ impl SyncManager {
         }
     }
 
+    /// Sync purgatory announcements into repo_sync_index as StateOnly entries.
+    ///
+    /// Called periodically by the purgatory announcement sync timer (every 5s).
+    /// For each announcement currently in purgatory, ensures a `StateOnly` entry
+    /// exists in `repo_sync_index`. New entries are then picked up by
+    /// `handle_new_sync_filters` which connects to listed relay URLs and subscribes
+    /// to state events for that repo.
+    ///
+    /// Idempotent: existing entries are not downgraded (a promoted Full entry stays Full).
+    async fn sync_purgatory_announcements_to_index(&mut self) {
+        use crate::sync::algorithms::{compute_actions, derive_relay_targets};
+
+        // Collect all purgatory announcements (snapshot - no async holds)
+        let announcements = self.purgatory.announcements_for_sync();
+
+        if announcements.is_empty() {
+            return;
+        }
+
+        // Register any new entries in repo_sync_index as StateOnly
+        let mut new_relay_urls: std::collections::HashSet<String> = std::collections::HashSet::new();
+        {
+            let mut index = self.repo_sync_index.write().await;
+            for (repo_id, relays) in &announcements {
+                let entry = index.entry(repo_id.clone()).or_insert_with(|| {
+                    tracing::debug!(
+                        repo_id = %repo_id,
+                        "Registering purgatory announcement in repo_sync_index as StateOnly"
+                    );
+                    RepoSyncNeeds {
+                        relays: std::collections::HashSet::new(),
+                        root_events: std::collections::HashSet::new(),
+                        sync_level: SyncLevel::StateOnly,
+                    }
+                });
+                // Don't downgrade an already-Full entry
+                // Add any new relay URLs
+                for relay in relays {
+                    if entry.relays.insert(relay.clone()) {
+                        new_relay_urls.insert(relay.clone());
+                    }
+                }
+            }
+        }
+
+        if new_relay_urls.is_empty() {
+            return;
+        }
+
+        // For any relay URLs that are new, compute and send AddFilters actions
+        let all_targets = {
+            let repo_index = self.repo_sync_index.read().await;
+            derive_relay_targets(&repo_index)
+        };
+
+        let actions = {
+            let pending_index = self.pending_sync_index.read().await;
+            let relay_index = self.relay_sync_index.read().await;
+            compute_actions(&all_targets, &pending_index, &relay_index)
+        };
+
+        for action in actions {
+            // Only act on relays that have new URLs (avoids redundant work)
+            if new_relay_urls.contains(&action.relay_url) {
+                tracing::info!(
+                    relay = %action.relay_url,
+                    repos = action.items.repos.len(),
+                    "Purgatory sync timer: connecting to new relay from purgatory announcement"
+                );
+                self.handle_new_sync_filters(action).await;
+            }
+        }
+    }
+
     /// Handle a relay disconnection
     ///
     /// This method is called when the event loop terminates and sends a disconnect notification.
diff --git a/src/sync/self_subscriber.rs b/src/sync/self_subscriber.rs
index 86e4583..4d69c9a 100644
--- a/src/sync/self_subscriber.rs
+++ b/src/sync/self_subscriber.rs
@@ -18,7 +18,7 @@ use tokio::sync::{broadcast, mpsc};
 
 use crate::nostr::builder::SharedDatabase;
 
-use super::{AddFilters, RepoSyncIndex, RepoSyncNeeds};
+use super::{AddFilters, RepoSyncIndex, RepoSyncNeeds, SyncLevel};
 
 // =============================================================================
 // LoopControl - Result of notification processing
@@ -60,6 +60,7 @@ impl PendingUpdates {
         let entry = self.repos.entry(repo_id).or_insert_with(|| RepoSyncNeeds {
             relays: HashSet::new(),
             root_events: HashSet::new(),
+            sync_level: SyncLevel::Full,
         });
         entry.relays.extend(relays);
         entry.root_events.extend(root_events);
@@ -132,14 +133,14 @@ impl SelfSubscriber {
 
     /// Get batch window from environment or use default
     ///
-    /// Reads `NGIT_SYNC_BATCH_WINDOW_MS` environment variable.
+    /// When `NGIT_TEST=1` is set, uses 200ms for faster test execution.
     /// Default: 5000ms (5 seconds)
     fn get_batch_window() -> Duration {
-        std::env::var("NGIT_SYNC_BATCH_WINDOW_MS")
-            .ok()
-            .and_then(|s| s.parse::<u64>().ok())
-            .map(Duration::from_millis)
-            .unwrap_or(Duration::from_millis(5000))
+        if std::env::var("NGIT_TEST").as_deref() == Ok("1") {
+            Duration::from_millis(200)
+        } else {
+            Duration::from_millis(5000)
+        }
     }
 
     /// Load existing events from database on startup
@@ -197,6 +198,7 @@ impl SelfSubscriber {
                     .or_insert_with(|| RepoSyncNeeds {
                         relays: HashSet::new(),
                         root_events: HashSet::new(),
+                        sync_level: SyncLevel::StateOnly,
                     });
                 entry.relays.extend(needs.relays.clone());
             }
@@ -570,7 +572,12 @@ impl SelfSubscriber {
                 .or_insert_with(|| RepoSyncNeeds {
                     relays: HashSet::new(),
                     root_events: HashSet::new(),
+                    sync_level: SyncLevel::Full,
                 });
+            // Upgrade sync_level to Full - this handles the case where the entry
+            // already exists as StateOnly (purgatory announcement) and is now being
+            // promoted (git data arrived and the event was broadcast via notify_event).
+            entry.sync_level = SyncLevel::Full;
             entry.relays.extend(needs.relays);
             entry.root_events.extend(needs.root_events);
 
@@ -594,21 +601,26 @@ impl SelfSubscriber {
                 continue;
             }
 
-            // Build filters for these repos
-            let filters = crate::sync::filters::build_layer2_and_layer3_filters(
+            // Build filters for these repos (sync-level-aware)
+            let filters = crate::sync::filters::build_sync_level_aware_filters(
                 &needs.repos,
+                &needs.state_only_repos,
                 &needs.root_events,
                 None,
             );
 
             // Log before moving values
-            let repo_count = needs.repos.len();
+            let repo_count = needs.repos.len() + needs.state_only_repos.len();
             let event_count = needs.root_events.len();
 
+            // Combine all repos into pending items
+            let mut all_repos = needs.repos;
+            all_repos.extend(needs.state_only_repos);
+
             let action = AddFilters {
                 relay_url: relay_url.clone(),
                 items: crate::sync::PendingItems {
-                    repos: needs.repos,
+                    repos: all_repos,
                     root_events: needs.root_events,
                 },
                 filters,