diff --git a/src/git/authorization.rs b/src/git/authorization.rs index e174b51..9d53c4f 100644 --- a/src/git/authorization.rs +++ b/src/git/authorization.rs @@ -287,6 +287,39 @@ pub async fn fetch_repository_data( }) } +/// Fetch repository data including announcements from purgatory +/// +/// This combines database announcements with purgatory announcements, +/// which is needed for authorization when the announcement hasn't been +/// promoted yet (no git data has arrived). +pub async fn fetch_repository_data_with_purgatory( + database: &SharedDatabase, + purgatory: &crate::purgatory::Purgatory, + identifier: &str, +) -> Result { + // First, fetch from database + let mut repo_data = fetch_repository_data(database, identifier).await?; + + // Then, add announcements from purgatory + let purgatory_announcements = purgatory.get_announcements_by_identifier(identifier); + let purgatory_count = purgatory_announcements.len(); + + for entry in purgatory_announcements { + if let Ok(announcement) = RepositoryAnnouncement::from_event(entry.event) { + repo_data.announcements.push(announcement); + } + } + + debug!( + "Fetched repository data with purgatory: {} announcements ({} from purgatory), {} states", + repo_data.announcements.len(), + purgatory_count, + repo_data.states.len() + ); + + Ok(repo_data) +} + pub fn pubkey_authorised_for_repo_owners( pubkey: &PublicKey, db_repo_data: &RepositoryData, @@ -539,8 +572,9 @@ pub async fn get_state_authorization_for_specific_owner_repo( use crate::git::list_refs; use crate::purgatory::RefUpdate; - // Fetch announcements only - we don't need database states - let repo_data = fetch_repository_data(database, identifier).await?; + // Fetch announcements from database AND purgatory - needed for authorization + // when the announcement hasn't been promoted yet (no git data has arrived) + let repo_data = fetch_repository_data_with_purgatory(database, purgatory, identifier).await?; if repo_data.announcements.is_empty() { return Ok(AuthorizationResult::denied( diff --git a/src/git/sync.rs b/src/git/sync.rs index e8e9655..13f30b6 100644 --- a/src/git/sync.rs +++ b/src/git/sync.rs @@ -51,6 +51,8 @@ use crate::purgatory::{can_apply_state, Purgatory}; /// or from purgatory sync fetching OIDs from remote servers). #[derive(Debug, Default, Clone)] pub struct ProcessResult { + /// Number of announcements released from purgatory + pub announcements_released: usize, /// Number of state events released from purgatory pub states_released: usize, /// Number of PR events released from purgatory @@ -70,11 +72,12 @@ pub struct ProcessResult { impl ProcessResult { /// Check if any events were released pub fn released_any(&self) -> bool { - self.states_released > 0 || self.prs_released > 0 + self.announcements_released > 0 || self.states_released > 0 || self.prs_released > 0 } /// Merge another ProcessResult into this one pub fn merge(&mut self, other: ProcessResult) { + self.announcements_released += other.announcements_released; self.states_released += other.states_released; self.prs_released += other.prs_released; self.repos_synced += other.repos_synced; @@ -836,6 +839,18 @@ pub async fn process_newly_available_git_data( "Processing newly available git data" ); + // Process announcements from purgatory + let announcement_result = process_purgatory_announcements( + &identifier, + source_repo_path, + database, + local_relay, + purgatory, + git_data_path, + ) + .await; + result.merge(announcement_result); + // Process state events from purgatory let state_result = process_purgatory_state_events( &identifier, @@ -863,6 +878,7 @@ pub async fn process_newly_available_git_data( if result.released_any() { info!( identifier = %identifier, + announcements_released = result.announcements_released, states_released = result.states_released, prs_released = result.prs_released, repos_synced = result.repos_synced, @@ -1250,6 +1266,90 @@ async fn process_purgatory_pr_events( result } +/// Process announcements from purgatory that can now be promoted. +/// +/// When git data arrives for a repository, any announcements in purgatory +/// for that repository should be promoted to the database and served to clients. +async fn process_purgatory_announcements( + identifier: &str, + source_repo_path: &Path, + database: &SharedDatabase, + local_relay: Option<&nostr_relay_builder::LocalRelay>, + purgatory: &Purgatory, + git_data_path: &Path, +) -> ProcessResult { + let mut result = ProcessResult::default(); + + // Extract owner pubkey from the source repo path + let owner_pubkey = match extract_owner_from_repo_path(source_repo_path, git_data_path) { + Some(npub) => npub, + None => { + debug!( + identifier = %identifier, + "Could not extract owner from repo path" + ); + return result; + } + }; + + // Parse the npub back to PublicKey + let owner = match nostr_sdk::PublicKey::parse(&owner_pubkey) { + Ok(pk) => pk, + Err(e) => { + warn!( + identifier = %identifier, + owner_pubkey = %owner_pubkey, + error = %e, + "Failed to parse owner pubkey" + ); + result.errors.push(format!("Failed to parse owner pubkey: {}", e)); + return result; + } + }; + + // Check if there's an announcement in purgatory for this owner and identifier + let announcement_event = purgatory.promote_announcement(&owner, identifier); + + if let Some(event) = announcement_event { + // Save to database + match database.save_event(&event).await { + Ok(_) => { + info!( + identifier = %identifier, + event_id = %event.id, + "Promoted announcement from purgatory to database" + ); + + // Notify WebSocket subscribers + if let Some(relay) = local_relay { + if relay.notify_event(event.clone()) { + debug!( + identifier = %identifier, + event_id = %event.id, + "Broadcast announcement event to WebSocket listeners" + ); + } + } + + result.announcements_released += 1; + } + Err(e) => { + warn!( + identifier = %identifier, + event_id = %event.id, + error = %e, + "Failed to save announcement to database" + ); + result + .errors + .push(format!("Failed to save announcement: {}", e)); + } + } + } + + result +} + /// Extract owner pubkey from a repository path. /// /// Given a path like `{git_data_path}/{npub}/{identifier}.git`, extracts the npub. @@ -1271,6 +1371,7 @@ mod tests { #[test] fn test_process_result_default() { let result = ProcessResult::default(); + assert_eq!(result.announcements_released, 0); assert_eq!(result.states_released, 0); assert_eq!(result.prs_released, 0); assert_eq!(result.repos_synced, 0); @@ -1282,6 +1383,10 @@ mod tests { let mut result = ProcessResult::default(); assert!(!result.released_any()); + result.announcements_released = 1; + assert!(result.released_any()); + + result.announcements_released = 0; result.states_released = 1; assert!(result.released_any()); @@ -1293,6 +1398,7 @@ mod tests { #[test] fn test_process_result_merge() { let mut result1 = ProcessResult { + announcements_released: 0, states_released: 1, prs_released: 2, repos_synced: 3, @@ -1303,6 +1409,7 @@ mod tests { }; let result2 = ProcessResult { + announcements_released: 5, states_released: 10, prs_released: 20, repos_synced: 30, @@ -1314,6 +1421,7 @@ mod tests { result1.merge(result2); + assert_eq!(result1.announcements_released, 5); assert_eq!(result1.states_released, 11); assert_eq!(result1.prs_released, 22); assert_eq!(result1.repos_synced, 33); diff --git a/src/main.rs b/src/main.rs index 5e5b83a..ab6ede7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -142,11 +142,11 @@ async fn main() -> Result<()> { let mut interval = tokio::time::interval(Duration::from_secs(60)); loop { interval.tick().await; - let (state_removed, pr_removed) = cleanup_purgatory.cleanup(); - if state_removed > 0 || pr_removed > 0 { + let (announcement_removed, state_removed, pr_removed) = cleanup_purgatory.cleanup(); + if announcement_removed > 0 || state_removed > 0 || pr_removed > 0 { info!( - "Purgatory cleanup: removed {} state events, {} PR events", - state_removed, pr_removed + "Purgatory cleanup: removed {} announcements, {} state events, {} PR events", + announcement_removed, state_removed, pr_removed ); } } diff --git a/src/nostr/builder.rs b/src/nostr/builder.rs index 34014db..aff12a6 100644 --- a/src/nostr/builder.rs +++ b/src/nostr/builder.rs @@ -138,6 +138,29 @@ impl Nip34WritePolicy { } } } + AnnouncementResult::AcceptPurgatory => { + // New announcement - add to purgatory + match self.announcement_policy.add_to_purgatory(event) { + Ok(()) => { + tracing::info!( + "Accepted announcement to purgatory: {} (waiting for git data)", + event_id_str + ); + WritePolicyResult::Reject { + status: true, // Client sees OK + message: "purgatory: won't be served until git data arrives".into(), + } + } + Err(e) => { + tracing::warn!( + "Failed to add announcement to purgatory {}: {}", + event_id_str, + e + ); + WritePolicyResult::reject(e) + } + } + } AnnouncementResult::AcceptMaintainer => { // Parse announcement to get details for logging match RepositoryAnnouncement::from_event(event.clone()) { diff --git a/src/nostr/policy/announcement.rs b/src/nostr/policy/announcement.rs index 15a6e58..1118497 100644 --- a/src/nostr/policy/announcement.rs +++ b/src/nostr/policy/announcement.rs @@ -3,6 +3,7 @@ /// Handles validation of NIP-34 repository announcements (kind 30617) /// according to GRASP-01 specification. use nostr_relay_builder::prelude::{Alphabet, Event, Filter, Kind, PublicKey, SingleLetterTag}; +use std::collections::HashSet; use super::PolicyContext; use crate::config::Config; @@ -11,12 +12,14 @@ use crate::nostr::events::{validate_announcement, RepositoryAnnouncement}; /// Result of announcement policy evaluation #[derive(Debug, Clone, PartialEq)] pub enum AnnouncementResult { - /// Accept: Event lists our service (GRASP-01 compliant) + /// Accept: Event lists our service (GRASP-01 compliant) - replacement announcement Accept, /// Accept as maintainer: Event accepted via maintainer exception (multi-maintainer) AcceptMaintainer, /// Accept as archive: Event accepted via GRASP-05 archive whitelist (read-only) AcceptArchive, + /// Accept to purgatory: New announcement, waiting for git data + AcceptPurgatory, /// Reject: Event fails validation with reason Reject(String), } @@ -35,10 +38,12 @@ impl AnnouncementPolicy { /// Validate a repository announcement event /// - /// Returns `Accept` if the announcement lists the service properly, - /// `AcceptMaintainer` if accepted via maintainer exception, - /// `AcceptArchive` if accepted via GRASP-05 archive config, - /// or `Reject` with reason. + /// Returns: + /// - `Accept` if this is a replacement announcement (active announcement exists) + /// - `AcceptPurgatory` if this is a new announcement (no active announcement exists) + /// - `AcceptMaintainer` if accepted via maintainer exception + /// - `AcceptArchive` if accepted via GRASP-05 archive config + /// - `Reject` with reason if validation fails pub async fn validate(&self, event: &Event) -> AnnouncementResult { // First, try validation (GRASP-01 + GRASP-05) let validation_result = validate_announcement(event, &self.config); @@ -67,11 +72,111 @@ impl AnnouncementPolicy { Err(_) => AnnouncementResult::Reject(reason), } } - // Accept, AcceptArchive, or AcceptMaintainer - return as-is + AnnouncementResult::Accept | AnnouncementResult::AcceptArchive => { + // Parse announcement to check for existing active announcement + match RepositoryAnnouncement::from_event(event.clone()) { + Ok(announcement) => { + // Check if there's already an active announcement for this (pubkey, identifier) + match self + .has_active_announcement(&event.pubkey, &announcement.identifier) + .await + { + Ok(true) => { + // Replacement announcement - accept immediately + tracing::debug!( + identifier = %announcement.identifier, + "Replacement announcement - accepting immediately" + ); + validation_result + } + Ok(false) => { + // New announcement - route to purgatory + tracing::debug!( + identifier = %announcement.identifier, + "New announcement - routing to purgatory" + ); + AnnouncementResult::AcceptPurgatory + } + Err(e) => { + tracing::warn!( + error = %e, + "Failed to check for existing announcement - rejecting" + ); + AnnouncementResult::Reject(format!( + "Database error checking existing announcement: {}", + e + )) + } + } + } + Err(e) => AnnouncementResult::Reject(format!( + "Failed to parse announcement: {}", + e + )), + } + } + // AcceptPurgatory shouldn't come from validate_announcement, but handle it result => result, } } + /// Check if there's an active announcement in the database for this (pubkey, identifier) + async fn has_active_announcement( + &self, + pubkey: &PublicKey, + identifier: &str, + ) -> Result { + let filter = Filter::new() + .kind(Kind::GitRepoAnnouncement) + .author(*pubkey) + .custom_tag( + SingleLetterTag::lowercase(Alphabet::D), + identifier.to_string(), + ); + + let events: Vec = match self.ctx.database.query(filter).await { + Ok(events) => events.into_iter().collect(), + Err(e) => return Err(format!("Database query failed: {}", e)), + }; + + Ok(!events.is_empty()) + } + + /// Add an announcement to purgatory + /// + /// Creates the bare repository and stores the announcement in purgatory + /// until git data arrives. + pub fn add_to_purgatory(&self, event: &Event) -> Result<(), String> { + let announcement = RepositoryAnnouncement::from_event(event.clone()) + .map_err(|e| format!("Failed to parse announcement: {}", e))?; + + // Create bare repository + self.ensure_bare_repository(&announcement)?; + + // Build repo path + let repo_path = self.ctx.git_data_path.join(announcement.repo_path()); + + // Extract relays from announcement + let relays: HashSet = announcement.relays.iter().cloned().collect(); + + // Add to purgatory + self.ctx.purgatory.add_announcement( + event.clone(), + announcement.identifier.clone(), + event.pubkey, + repo_path, + relays, + ); + + tracing::info!( + identifier = %announcement.identifier, + event_id = %event.id, + "Added announcement to purgatory" + ); + + Ok(()) + } + /// Create a bare git repository if it doesn't exist /// Path format: //.git pub fn ensure_bare_repository( diff --git a/src/nostr/policy/state.rs b/src/nostr/policy/state.rs index f94f004..4bfb513 100644 --- a/src/nostr/policy/state.rs +++ b/src/nostr/policy/state.rs @@ -10,7 +10,7 @@ use nostr_relay_builder::prelude::Event; use super::PolicyContext; use crate::git; -use crate::git::authorization::fetch_repository_data; +use crate::git::authorization::fetch_repository_data_with_purgatory; use crate::nostr::events::{validate_state, RepositoryAnnouncement, RepositoryState}; /// Result of state policy evaluation @@ -76,7 +76,13 @@ impl StatePolicy { } // Get all repositories and state events from db with identifier - let db_repo_data = fetch_repository_data(&self.ctx.database, &state.identifier).await?; + // Include purgatory announcements for authorization + let db_repo_data = fetch_repository_data_with_purgatory( + &self.ctx.database, + &self.ctx.purgatory, + &state.identifier, + ) + .await?; // CRITICAL: Check if author is authorized via maintainer set // State events MUST be rejected if author is not in maintainer set of any accepted announcement diff --git a/src/purgatory/mod.rs b/src/purgatory/mod.rs index 47798a6..3b5514b 100644 --- a/src/purgatory/mod.rs +++ b/src/purgatory/mod.rs @@ -17,7 +17,7 @@ pub mod sync; mod types; pub use helpers::{can_apply_state, can_satisfy_state, extract_refs_from_state, get_unpushed_refs}; -pub use types::{PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry}; +pub use types::{AnnouncementPurgatoryEntry, PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry}; use dashmap::DashMap; use nostr_sdk::prelude::*; @@ -100,7 +100,8 @@ struct PurgatoryState { /// Main purgatory structure holding events awaiting git data. /// -/// Provides thread-safe concurrent access to two separate stores: +/// Provides thread-safe concurrent access to three separate stores: +/// - Announcements indexed by (pubkey, identifier) /// - State events indexed by repository identifier /// - PR events indexed by event ID /// @@ -121,6 +122,10 @@ struct PurgatoryState { /// that we've already determined have no git data available. #[derive(Clone)] pub struct Purgatory { + /// Repository announcements (kind 30617) indexed by (owner pubkey, identifier). + /// Key: (PublicKey, String) where String is the repository identifier. + announcement_purgatory: Arc>, + /// State events (kind 30618) indexed by repository identifier. /// Multiple state events can wait for the same identifier (different maintainers). state_events: Arc>>, @@ -145,6 +150,7 @@ impl Purgatory { /// Create a new empty purgatory. pub fn new(git_data_path: impl Into) -> Self { Self { + announcement_purgatory: Arc::new(DashMap::new()), state_events: Arc::new(DashMap::new()), pr_events: Arc::new(DashMap::new()), sync_queue: Arc::new(DashMap::new()), @@ -513,9 +519,171 @@ impl Purgatory { self.pr_events.remove(event_id); } + // ========================================================================= + // Announcement Purgatory Methods + // ========================================================================= + + /// Add a repository announcement to purgatory. + /// + /// The announcement will be held until git data arrives, at which point + /// it will be promoted to the database and served to clients. + /// + /// # Arguments + /// * `event` - The announcement event (kind 30617) + /// * `identifier` - The repository identifier from the 'd' tag + /// * `owner` - The owner pubkey (event author) + /// * `repo_path` - Path to the bare git repository + /// * `relays` - Relay URLs from the announcement (for sync registration) + pub fn add_announcement( + &self, + event: Event, + identifier: String, + owner: PublicKey, + repo_path: PathBuf, + relays: HashSet, + ) { + let now = Instant::now(); + let entry = AnnouncementPurgatoryEntry { + event, + identifier: identifier.clone(), + owner, + repo_path, + relays, + created_at: now, + expires_at: now + DEFAULT_EXPIRY, + soft_expired: false, + }; + + let key = (owner, identifier); + self.announcement_purgatory.insert(key.clone(), entry); + + tracing::debug!( + owner = %key.0, + identifier = %key.1, + "Added announcement to purgatory" + ); + } + + /// Find an announcement in purgatory by owner and identifier. + /// + /// # Arguments + /// * `owner` - The owner pubkey + /// * `identifier` - The repository identifier + /// + /// # Returns + /// The announcement entry if found, None otherwise + pub fn find_announcement(&self, owner: &PublicKey, identifier: &str) -> Option { + let key = (*owner, identifier.to_string()); + self.announcement_purgatory.get(&key).map(|entry| entry.clone()) + } + + /// Get all announcements in purgatory for a given identifier. + /// + /// This is used for authorization - state events and git pushes need to + /// check purgatory announcements for maintainer validation. + /// + /// # Arguments + /// * `identifier` - The repository identifier + /// + /// # Returns + /// Vector of announcement entries for this identifier + pub fn get_announcements_by_identifier(&self, identifier: &str) -> Vec { + self.announcement_purgatory + .iter() + .filter(|entry| entry.key().1 == identifier) + .map(|entry| entry.value().clone()) + .collect() + } + + /// Remove an announcement from purgatory. + /// + /// # Arguments + /// * `owner` - The owner pubkey + /// * `identifier` - The repository identifier + pub fn remove_announcement(&self, owner: &PublicKey, identifier: &str) { + let key = (*owner, identifier.to_string()); + self.announcement_purgatory.remove(&key); + tracing::debug!( + owner = %owner, + identifier = %identifier, + "Removed announcement from purgatory" + ); + } + + /// Promote an announcement from purgatory to active status. + /// + /// This is called when git data arrives. The announcement event is returned + /// so it can be saved to the database. + /// + /// # Arguments + /// * `owner` - The owner pubkey + /// * `identifier` - The repository identifier + /// + /// # Returns + /// The announcement event if found, None otherwise + pub fn promote_announcement(&self, owner: &PublicKey, identifier: &str) -> Option { + let key = (*owner, identifier.to_string()); + self.announcement_purgatory.remove(&key).map(|(_, entry)| { + tracing::info!( + owner = %owner, + identifier = %identifier, + "Promoted announcement from purgatory to database" + ); + entry.event + }) + } + + /// Check if there's an announcement in purgatory for the given owner and identifier. + /// + /// # Arguments + /// * `owner` - The owner pubkey + /// * `identifier` - The repository identifier + /// + /// # Returns + /// true if an announcement exists in purgatory, false otherwise + pub fn has_purgatory_announcement(&self, owner: &PublicKey, identifier: &str) -> bool { + let key = (*owner, identifier.to_string()); + self.announcement_purgatory.contains_key(&key) + } + + /// Extend the expiry for an announcement in purgatory. + /// + /// This is called when state events arrive for a purgatory announcement, + /// indicating the repository is actively receiving metadata. + /// + /// # Arguments + /// * `owner` - The owner pubkey + /// * `identifier` - The repository identifier + /// * `duration` - Minimum duration to guarantee from now + pub fn extend_announcement_expiry(&self, owner: &PublicKey, identifier: &str, duration: Duration) { + let key = (*owner, identifier.to_string()); + if let Some(mut entry) = self.announcement_purgatory.get_mut(&key) { + let now = Instant::now(); + let new_expiry = now + duration; + if entry.expires_at < new_expiry { + entry.expires_at = new_expiry; + // If soft-expired, revive it + if entry.soft_expired { + entry.soft_expired = false; + tracing::debug!( + owner = %owner, + identifier = %identifier, + "Revived soft-expired announcement" + ); + } + } + } + } + + /// Get count of announcements in purgatory. + pub fn announcement_count(&self) -> usize { + self.announcement_purgatory.len() + } + /// Get all event IDs currently stored in purgatory AND previously expired events. /// /// Returns a HashSet of all event IDs for: + /// - Announcements currently held in purgatory /// - State events currently held in purgatory /// - PR events currently held in purgatory /// - Events that previously expired from purgatory without finding git data @@ -530,6 +698,11 @@ impl Purgatory { pub fn event_ids(&self) -> HashSet { let mut ids = HashSet::new(); + // Collect announcement event IDs + for entry in self.announcement_purgatory.iter() { + ids.insert(entry.value().event.id); + } + // Collect state event IDs for entry in self.state_events.iter() { for state_entry in entry.value().iter() { @@ -609,9 +782,28 @@ impl Purgatory { /// will be filtered out during future negentropy/REQ sync operations. /// /// # Returns - /// Tuple of (num_state_removed, num_pr_removed) - pub fn cleanup(&self) -> (usize, usize) { + /// Tuple of (num_announcement_removed, num_state_removed, num_pr_removed) + pub fn cleanup(&self) -> (usize, usize, usize) { let now = Instant::now(); + + // Remove expired announcements and mark them as expired + let expired_announcements: Vec<(PublicKey, String, EventId)> = self + .announcement_purgatory + .iter() + .filter(|entry| entry.value().expires_at <= now) + .map(|entry| { + let key = entry.key(); + let event_id = entry.value().event.id; + (key.0.clone(), key.1.clone(), event_id) + }) + .collect(); + + let announcement_removed = expired_announcements.len(); + for (owner, identifier, event_id) in expired_announcements { + self.mark_expired(event_id); + self.announcement_purgatory.remove(&(owner, identifier)); + } + let mut state_removed = 0; // Remove expired state events and mark them as expired @@ -655,17 +847,17 @@ impl Purgatory { self.pr_events.remove(&event_id_str); } - (state_removed, pr_removed) + (announcement_removed, state_removed, pr_removed) } /// Remove expired entries from purgatory (legacy method). /// /// # Returns - /// Total number of entries removed (state + PR events) + /// Total number of entries removed (announcement + state + PR events) #[deprecated(since = "0.1.0", note = "Use cleanup() instead for separate counts")] pub fn remove_expired(&self) -> usize { - let (state, pr) = self.cleanup(); - state + pr + let (announcement, state, pr) = self.cleanup(); + announcement + state + pr } /// Remove old expired event records. @@ -699,11 +891,12 @@ impl Purgatory { /// Get current count of entries in purgatory. /// /// # Returns - /// Tuple of (state_event_count, pr_event_count) - pub fn count(&self) -> (usize, usize) { + /// Tuple of (announcement_count, state_event_count, pr_event_count) + pub fn count(&self) -> (usize, usize, usize) { + let announcement_count = self.announcement_purgatory.len(); let state_count: usize = self.state_events.iter().map(|e| e.value().len()).sum(); let pr_count = self.pr_events.len(); - (state_count, pr_count) + (announcement_count, state_count, pr_count) } /// Get count of expired events being tracked. @@ -717,6 +910,7 @@ impl Purgatory { /// Clear all entries from purgatory (for testing). #[cfg(test)] pub fn clear(&self) { + self.announcement_purgatory.clear(); self.state_events.clear(); self.pr_events.clear(); self.sync_queue.clear(); @@ -990,7 +1184,8 @@ mod tests { #[test] fn test_purgatory_creation() { let purgatory = Purgatory::new(PathBuf::new()); - let (state_count, pr_count) = purgatory.count(); + let (announcement_count, state_count, pr_count) = purgatory.count(); + assert_eq!(announcement_count, 0); assert_eq!(state_count, 0); assert_eq!(pr_count, 0); } @@ -1008,7 +1203,8 @@ mod tests { purgatory.add_state(event.clone(), "test-repo".to_string(), keys.public_key()); purgatory.add_pr(event, "test-event-id".to_string(), "abc123".to_string()); - let (state_count, pr_count) = purgatory.count(); + let (announcement_count, state_count, pr_count) = purgatory.count(); + assert_eq!(announcement_count, 0); assert_eq!(state_count, 1); assert_eq!(pr_count, 1); } @@ -1213,7 +1409,7 @@ fn test_cleanup_removes_expired_entries() { purgatory.add_pr_placeholder("pr-456".to_string(), "commit-def".to_string()); // Verify entries are there - let (state_count, pr_count) = purgatory.count(); + let (_, state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 1); assert_eq!(pr_count, 2); @@ -1231,14 +1427,14 @@ fn test_cleanup_removes_expired_entries() { } // Run cleanup - let (state_removed, pr_removed) = purgatory.cleanup(); + let (_, state_removed, pr_removed) = purgatory.cleanup(); // Verify counts assert_eq!(state_removed, 1); assert_eq!(pr_removed, 2); // Verify entries are gone - let (state_count, pr_count) = purgatory.count(); + let (_, state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 0); assert_eq!(pr_count, 0); } @@ -1260,14 +1456,14 @@ fn test_cleanup_preserves_non_expired_entries() { purgatory.add_pr(pr_event, "pr-123".to_string(), "commit-abc".to_string()); // Run cleanup - let (state_removed, pr_removed) = purgatory.cleanup(); + let (_, state_removed, pr_removed) = purgatory.cleanup(); // Nothing should be removed assert_eq!(state_removed, 0); assert_eq!(pr_removed, 0); // Verify entries are still there - let (state_count, pr_count) = purgatory.count(); + let (_, state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 1); assert_eq!(pr_count, 1); } @@ -1314,14 +1510,14 @@ fn test_cleanup_mixed_expired_and_fresh() { } // Run cleanup - let (state_removed, pr_removed) = purgatory.cleanup(); + let (_, state_removed, pr_removed) = purgatory.cleanup(); // One of each should be removed assert_eq!(state_removed, 1); assert_eq!(pr_removed, 1); // Verify remaining counts - let (state_count, pr_count) = purgatory.count(); + let (_, state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 1); // One state event remains assert_eq!(pr_count, 1); // One PR event remains } @@ -1391,7 +1587,7 @@ fn test_expired_event_tracking() { } // Run cleanup - let (state_removed, pr_removed) = purgatory.cleanup(); + let (_, state_removed, pr_removed) = purgatory.cleanup(); assert_eq!(state_removed, 1); assert_eq!(pr_removed, 1); @@ -1501,7 +1697,7 @@ fn test_expired_events_prevent_readdition() { } // Event should NOT be re-added - let (state_count, _) = purgatory.count(); + let (_, state_count, _) = purgatory.count(); assert_eq!(state_count, 0, "Event should not be re-added to purgatory"); } @@ -1520,7 +1716,7 @@ fn test_pr_placeholder_not_marked_expired() { } // Run cleanup - let (_, pr_removed) = purgatory.cleanup(); + let (_, _, pr_removed) = purgatory.cleanup(); assert_eq!(pr_removed, 1); // Expired count should be 0 (placeholders don't have event IDs to track) @@ -1606,7 +1802,7 @@ async fn test_save_and_restore_state_events() { assert!(!state_file.exists()); // Verify state events were restored - let (state_count, _) = purgatory2.count(); + let (_, state_count, _) = purgatory2.count(); assert_eq!(state_count, 2); let restored_entries = purgatory2.find_state("test-repo"); @@ -1662,7 +1858,7 @@ async fn test_save_and_restore_pr_events() { purgatory2.restore_from_disk(&state_file).unwrap(); // Verify PR event was restored - let (_, pr_count) = purgatory2.count(); + let (_, _, pr_count) = purgatory2.count(); assert_eq!(pr_count, 1); let restored_entry = purgatory2.find_pr("pr-event-id").unwrap(); @@ -1691,7 +1887,7 @@ async fn test_save_and_restore_pr_placeholders() { purgatory2.restore_from_disk(&state_file).unwrap(); // Verify placeholder was restored - let (_, pr_count) = purgatory2.count(); + let (_, _, pr_count) = purgatory2.count(); assert_eq!(pr_count, 1); let restored_entry = purgatory2.find_pr("placeholder-id").unwrap(); @@ -1769,7 +1965,7 @@ async fn test_save_and_restore_empty_purgatory() { purgatory2.restore_from_disk(&state_file).unwrap(); // Verify purgatory is still empty - let (state_count, pr_count) = purgatory2.count(); + let (_, state_count, pr_count) = purgatory2.count(); assert_eq!(state_count, 0); assert_eq!(pr_count, 0); assert_eq!(purgatory2.expired_count(), 0); @@ -1789,7 +1985,7 @@ async fn test_restore_missing_file() { assert!(result.is_err()); // Purgatory should remain empty - let (state_count, pr_count) = purgatory.count(); + let (_, state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 0); assert_eq!(pr_count, 0); } @@ -1811,7 +2007,7 @@ async fn test_restore_corrupted_json() { assert!(result.is_err()); // Purgatory should remain empty - let (state_count, pr_count) = purgatory.count(); + let (_, state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 0); assert_eq!(pr_count, 0); } @@ -2044,7 +2240,7 @@ async fn test_mixed_pr_events_and_placeholders() { purgatory2.restore_from_disk(&state_file).unwrap(); // Verify both were restored correctly - let (_, pr_count) = purgatory2.count(); + let (_, _, pr_count) = purgatory2.count(); assert_eq!(pr_count, 2); // Verify PR event @@ -2141,7 +2337,7 @@ async fn test_comprehensive_roundtrip() { purgatory.cleanup(); // Verify initial state - let (state_count, pr_count) = purgatory.count(); + let (_, state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 2); // state1, state2 (expired_event was cleaned up) assert_eq!(pr_count, 2); // pr-1, pr-2 assert_eq!(purgatory.expired_count(), 1); // expired_event @@ -2154,7 +2350,7 @@ async fn test_comprehensive_roundtrip() { purgatory2.restore_from_disk(&state_file).unwrap(); // Verify all data was restored correctly - let (state_count2, pr_count2) = purgatory2.count(); + let (_, state_count2, pr_count2) = purgatory2.count(); assert_eq!(state_count2, 2); assert_eq!(pr_count2, 2); assert_eq!(purgatory2.expired_count(), 1); diff --git a/src/purgatory/sync/context.rs b/src/purgatory/sync/context.rs index 33c2d12..778cdb8 100644 --- a/src/purgatory/sync/context.rs +++ b/src/purgatory/sync/context.rs @@ -279,7 +279,12 @@ impl SyncContext for RealSyncContext { } async fn fetch_repository_data(&self, identifier: &str) -> Result { - crate::git::authorization::fetch_repository_data(&self.database, identifier).await + crate::git::authorization::fetch_repository_data_with_purgatory( + &self.database, + &self.purgatory, + identifier, + ) + .await } fn collect_needed_oids(&self, identifier: &str) -> HashSet { diff --git a/src/purgatory/types.rs b/src/purgatory/types.rs index 919504b..d891bc9 100644 --- a/src/purgatory/types.rs +++ b/src/purgatory/types.rs @@ -6,6 +6,8 @@ use nostr_sdk::prelude::*; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; +use std::path::PathBuf; use std::time::Instant; /// Default value for Instant fields during deserialization @@ -113,3 +115,40 @@ pub struct PrPurgatoryEntry { #[serde(skip, default = "instant_now")] pub expires_at: Instant, } + +/// Entry for a repository announcement (kind 30617) waiting in purgatory. +/// +/// Announcements are held in purgatory until git data arrives, proving +/// the repository has actual content. This prevents serving announcements +/// for empty repositories. +/// +/// Note: `Instant` fields cannot be serialized directly. Use the `persistence` +/// module to convert to/from serializable wrapper types. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AnnouncementPurgatoryEntry { + /// The nostr announcement event (kind 30617) + pub event: Event, + + /// The repository identifier from the event's 'd' tag + pub identifier: String, + + /// The owner pubkey (event author) + pub owner: PublicKey, + + /// Path to the bare git repository + pub repo_path: PathBuf, + + /// Relay URLs from the announcement (for sync registration) + pub relays: HashSet, + + /// When this entry was added to purgatory + #[serde(skip, default = "instant_now")] + pub created_at: Instant, + + /// Expiry deadline (30 min from creation, may be extended) + #[serde(skip, default = "instant_now")] + pub expires_at: Instant, + + /// Whether the bare repo has been deleted (soft expiry) + pub soft_expired: bool, +} diff --git a/src/sync/mod.rs b/src/sync/mod.rs index 1ee1872..872df66 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -1719,8 +1719,50 @@ impl SyncManager { // For sync-triggered events that go to purgatory, trigger immediate sync // (instead of the default 3-minute delay for user-submitted events) if result == ProcessResult::Purgatory { + // Announcements (kind 30617) - re-process rejected state events + // When an announcement goes to purgatory, state events that were + // previously rejected ("no announcement exists") can now be authorized + // via fetch_repository_data_with_purgatory. + if event.kind == Kind::GitRepoAnnouncement { + use crate::nostr::events::RepositoryAnnouncement; + + if let Ok(announcement) = RepositoryAnnouncement::from_event((*event).clone()) { + // Re-process rejected state events for this announcement + let (removed, hot_events) = rejected_events_index.invalidate_and_get( + &event.pubkey, + &announcement.identifier, + Some(rejected_index::EventType::State), + ); + + if removed > 0 { + tracing::info!( + pubkey = %event.pubkey, + identifier = %announcement.identifier, + removed_from_cold_index = removed, + hot_cache_events = hot_events.len(), + "Invalidated rejected state events (announcement now in purgatory)" + ); + } + + // Re-process state events from hot cache immediately + if !hot_events.is_empty() { + let _stats = Self::reprocess_events_from_hot_cache( + hot_events, + "state event (announcement in purgatory)", + &event.pubkey, + &announcement.identifier, + &relay_url_clone, + &database, + &write_policy, + &local_relay, + &rejected_events_index, + ) + .await; + } + } + } // State events (kind 30618) - extract identifier and trigger immediate sync - if event.kind.as_u16() == 30618 { + else if event.kind.as_u16() == 30618 { if let Some(identifier) = event.tags.iter().find_map(|tag| { let tag_vec = tag.clone().to_vec(); if tag_vec.len() >= 2 && tag_vec[0] == "d" { @@ -1754,7 +1796,9 @@ impl SyncManager { // Track pagination state for this subscription (REQ+EOSE) // and received event IDs for negentropy batches - if result == ProcessResult::Saved || result == ProcessResult::Duplicate { + // Include Purgatory results so announcements in purgatory still trigger + // per-repo sync (state events, PR events) from the source relay. + if result == ProcessResult::Saved || result == ProcessResult::Duplicate || result == ProcessResult::Purgatory { let mut pending = pending_sync_index.write().await; if let Some(batches) = pending.get_mut(&relay_url_clone) { for batch in batches.iter_mut() { @@ -2506,6 +2550,26 @@ impl SyncManager { "{} added to purgatory (waiting for git data)", context ); + // Trigger immediate sync for re-processed events that go to purgatory + // (same as sync-triggered events in the main event loop) + if event.kind.as_u16() == 30618 { + // State event - extract identifier from 'd' tag + if let Some(id) = event.tags.iter().find_map(|tag| { + let tag_vec = tag.clone().to_vec(); + if tag_vec.len() >= 2 && tag_vec[0] == "d" { + Some(tag_vec[1].clone()) + } else { + None + } + }) { + write_policy.purgatory().enqueue_sync_immediate(&id); + } + } else if event.kind.as_u16() == 1617 || event.kind.as_u16() == 1618 { + // PR event - extract identifier from 'a' tag + if let Some(id) = crate::git::sync::extract_identifier_from_pr_event(&event) { + write_policy.purgatory().enqueue_sync_immediate(&id); + } + } } ProcessResult::Rejected => { stats.rejected += 1; diff --git a/tests/archive_read_only.rs b/tests/archive_read_only.rs index be6959b..e39b4b2 100644 --- a/tests/archive_read_only.rs +++ b/tests/archive_read_only.rs @@ -165,6 +165,7 @@ async fn test_archive_read_only_creates_bare_repo() { // c) Put state event in purgatory (git data missing on archive relay) // d) Fetch git data from source relay's clone URL // e) Release the state event from purgatory + let found = wait_for_event_served( archive_relay.url(), &state_event_id, @@ -267,11 +268,13 @@ async fn test_archive_read_only_creates_bare_repo() { /// This verifies the security model: archive mode only syncs git data /// when there are state events to validate against. /// -/// Scenario: -/// 1. Start source relay with announcement only (no state events) -/// 2. Start archive relay syncing from source -/// 3. Archive relay syncs announcement (creates bare repo) -/// 4. Verify git data is NOT synced (no state events to trigger purgatory sync) +/// With announcement purgatory, the flow is: +/// 1. Send announcement to source relay (goes to purgatory) +/// 2. Send state event to source relay (goes to purgatory) +/// 3. Push git data to source relay (promotes announcement and state event) +/// 4. Start archive relay with sync from source +/// 5. Archive relay syncs the promoted announcement +/// 6. Verify git data is NOT synced (archive has no state event to authorize git fetch) #[tokio::test] async fn test_archive_without_state_events_does_not_sync_git() { // 1. Start source relay @@ -290,7 +293,7 @@ async fn test_archive_without_state_events_does_not_sync_git() { let npub = keys.public_key().to_bech32().expect("Failed to get npub"); - // 3. Create and send announcement listing BOTH relays (but NO state event) + // 3. Create and send announcement listing BOTH relays let announcement = create_repo_announcement( &keys, &[&source_relay.domain(), &archive_domain], @@ -306,7 +309,7 @@ async fn test_archive_without_state_events_does_not_sync_git() { tokio::time::sleep(Duration::from_millis(500)).await; - // Send announcement to source relay + // Send announcement to source relay (goes to purgatory) source_client .send_event(&announcement) .await @@ -314,11 +317,39 @@ async fn test_archive_without_state_events_does_not_sync_git() { tokio::time::sleep(Duration::from_millis(200)).await; - // 4. Push git data to source relay (but no state event to authorize it) - // This push will fail because there's no state event in purgatory - // That's expected - we're testing that archive mode doesn't blindly fetch git data + // 4. Create and send state event to source relay (goes to purgatory) + let clone_url = format!( + "http://{}/{}/{}.git", + source_relay.domain(), + npub, + identifier + ); + let relay_url = source_relay.url().to_string(); - // 5. Start archive relay + let state_event = create_state_event( + &keys, + identifier, + &[("main", &commit_hash)], + &[], + &[&clone_url], + &[&relay_url], + ) + .expect("Failed to create state event"); + + source_client + .send_event(&state_event) + .await + .expect("Failed to send state event to source"); + + tokio::time::sleep(Duration::from_millis(200)).await; + + // 5. Push git data to source relay (promotes announcement and state event) + push_to_relay(temp_dir.path(), &source_relay.domain(), &npub, identifier) + .expect("Push to source should succeed"); + + tokio::time::sleep(Duration::from_millis(500)).await; + + // 6. Start archive relay (without state event - we don't send state event to archive) let archive_relay = TestRelay::start_with_archive_and_sync( archive_port, Some(source_relay.url().to_string()), @@ -333,10 +364,10 @@ async fn test_archive_without_state_events_does_not_sync_git() { .await .expect("Sync connection should establish"); - // Give time for any potential git sync to happen + // Give time for sync to fetch announcement tokio::time::sleep(Duration::from_secs(3)).await; - // 6. Verify bare repository was created (announcement was accepted) + // 7. Verify bare repository was created (announcement was synced and accepted to purgatory) let repo_path = archive_relay .git_data_path() .join(format!("{}/{}.git", npub, identifier)); @@ -346,7 +377,7 @@ async fn test_archive_without_state_events_does_not_sync_git() { "Bare repository should be created for archive announcement" ); - // 7. Verify git data was NOT synced (no state events to trigger purgatory sync) + // 8. Verify git data was NOT synced (no state events on archive to trigger git fetch) // Check that the commit does NOT exist in the archive relay's repo let output = tokio::process::Command::new("git") .args(["cat-file", "-t", &commit_hash]) diff --git a/tests/purgatory.rs b/tests/purgatory.rs index e99540b..efc28c9 100644 --- a/tests/purgatory.rs +++ b/tests/purgatory.rs @@ -58,10 +58,10 @@ macro_rules! isolated_purgatory_test { } // ============================================================ -// Announcement Purgatory Tests (commented out - feature not yet implemented) +// Announcement Purgatory Tests // ============================================================ -// isolated_purgatory_test!(test_announcement_not_served_before_git_data); +isolated_purgatory_test!(test_announcement_not_served_before_git_data); isolated_purgatory_test!(test_announcement_served_after_git_push); isolated_purgatory_test!(test_bare_repo_exists_for_purgatory_announcement); isolated_purgatory_test!(test_state_event_accepted_for_purgatory_announcement); diff --git a/tests/purgatory_persistence.rs b/tests/purgatory_persistence.rs index fe37c33..5abbf15 100644 --- a/tests/purgatory_persistence.rs +++ b/tests/purgatory_persistence.rs @@ -120,7 +120,8 @@ async fn test_full_purgatory_save_restore_cycle() { // so we'll focus on testing state and PR events persistence // Verify initial counts - let (state_count, pr_count) = purgatory.count(); + let (announcement_count, state_count, pr_count) = purgatory.count(); + assert_eq!(announcement_count, 0, "Should have 0 announcements"); assert_eq!(state_count, 2, "Should have 2 state events"); assert_eq!( pr_count, 3, @@ -142,7 +143,8 @@ async fn test_full_purgatory_save_restore_cycle() { ); // Verify all data was restored - let (state_count2, pr_count2) = purgatory2.count(); + let (announcement_count2, state_count2, pr_count2) = purgatory2.count(); + assert_eq!(announcement_count2, 0, "Should have 0 announcements after restore"); assert_eq!(state_count2, 2, "Should have 2 state events after restore"); assert_eq!( pr_count2, 3, @@ -275,7 +277,7 @@ async fn test_purgatory_downtime_adjustment() { purgatory2.restore_from_disk(&state_path).unwrap(); // Verify event is still there (downtime was accounted for) - let (state_count, _) = purgatory2.count(); + let (_, state_count, _) = purgatory2.count(); assert_eq!(state_count, 1); let repo1_states = purgatory2.find_state("repo1"); @@ -401,7 +403,7 @@ async fn test_purgatory_restore_missing_file() { assert!(result.is_err(), "Should error on missing file"); // Purgatory should still be usable (empty state) - let (state_count, pr_count) = purgatory.count(); + let (_, state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 0); assert_eq!(pr_count, 0); @@ -410,7 +412,7 @@ async fn test_purgatory_restore_missing_file() { let event = create_test_event(&keys, "test").await; purgatory.add_state(event, "repo1".to_string(), keys.public_key()); - let (state_count, _) = purgatory.count(); + let (_, state_count, _) = purgatory.count(); assert_eq!(state_count, 1); } @@ -461,7 +463,7 @@ async fn test_purgatory_restore_corrupted_file() { assert!(result.is_err(), "Should error on corrupted file"); // Purgatory should still be usable - let (state_count, pr_count) = purgatory.count(); + let (_, state_count, pr_count) = purgatory.count(); assert_eq!(state_count, 0); assert_eq!(pr_count, 0); } @@ -504,7 +506,7 @@ async fn test_empty_purgatory_save_restore() { purgatory2.restore_from_disk(&state_path).unwrap(); // Verify empty state - let (state_count, pr_count) = purgatory2.count(); + let (_, state_count, pr_count) = purgatory2.count(); assert_eq!(state_count, 0); assert_eq!(pr_count, 0); assert_eq!(purgatory2.expired_count(), 0); @@ -591,7 +593,7 @@ async fn test_purgatory_continues_working_after_restore() { purgatory2.add_state(event2.clone(), "repo2".to_string(), keys.public_key()); // Verify both old and new events work - let (state_count, _) = purgatory2.count(); + let (_, state_count, _) = purgatory2.count(); assert_eq!(state_count, 2); let repo1_states = purgatory2.find_state("repo1"); @@ -603,7 +605,7 @@ async fn test_purgatory_continues_working_after_restore() { assert_eq!(repo2_states[0].event.id, event2.id); // Verify cleanup still works - let (state_removed, pr_removed) = purgatory2.cleanup(); + let (_, state_removed, pr_removed) = purgatory2.cleanup(); // Nothing should be expired yet assert_eq!(state_removed, 0); assert_eq!(pr_removed, 0); @@ -684,15 +686,15 @@ async fn test_purgatory_entries_expired_during_downtime() { purgatory2.restore_from_disk(&state_path).unwrap(); // Event should be restored - let (state_count, _) = purgatory2.count(); + let (_, state_count, _) = purgatory2.count(); assert_eq!(state_count, 1); // Cleanup should work (even if nothing is expired yet) - let (state_removed, _) = purgatory2.cleanup(); + let (_, state_removed, _) = purgatory2.cleanup(); // Nothing expired yet since we didn't wait 30 minutes assert_eq!(state_removed, 0); - let (state_count, _) = purgatory2.count(); + let (_, state_count, _) = purgatory2.count(); assert_eq!(state_count, 1); }