From 70d0197e85ae4ef85202781f6d2dc9e76bd508b3 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Wed, 24 Dec 2025 08:02:12 +0000 Subject: feat(purgatory): add broken purgatory implementation --- src/nostr/builder.rs | 116 +++++++++++++++++++++++++++++---- src/nostr/policy/mod.rs | 5 ++ src/nostr/policy/pr_event.rs | 149 +++++++++++++++++++++++++++++++++++++++++++ src/nostr/policy/state.rs | 55 +++++++++++++++- 4 files changed, 310 insertions(+), 15 deletions(-) (limited to 'src/nostr') diff --git a/src/nostr/builder.rs b/src/nostr/builder.rs index 8dd6291..2b4d524 100644 --- a/src/nostr/builder.rs +++ b/src/nostr/builder.rs @@ -13,7 +13,7 @@ use nostr_relay_builder::prelude::*; use crate::config::{Config, DatabaseBackend}; use crate::nostr::events::{ - RepositoryAnnouncement, RepositoryState, KIND_PR, KIND_PR_UPDATE, KIND_REPOSITORY_ANNOUNCEMENT, + RepositoryAnnouncement, KIND_PR, KIND_PR_UPDATE, KIND_REPOSITORY_ANNOUNCEMENT, KIND_REPOSITORY_STATE, KIND_USER_GRASP_LIST, }; use crate::nostr::policy::{ @@ -57,8 +57,9 @@ impl Nip34WritePolicy { domain: impl Into, database: SharedDatabase, git_data_path: impl Into, + purgatory: std::sync::Arc, ) -> Self { - let ctx = PolicyContext::new(domain, database, git_data_path); + let ctx = PolicyContext::new(domain, database, git_data_path, purgatory); Self { announcement_policy: AnnouncementPolicy::new(ctx.clone()), state_policy: StatePolicy::new(ctx.clone()), @@ -143,21 +144,50 @@ impl Nip34WritePolicy { match self.state_policy.validate(event) { StateResult::Accept => { - // Parse state to get HEAD and branch info - match RepositoryState::from_event(event.clone()) { - Ok(_state) => { - // Process state alignment asynchronously - if let Err(e) = self.state_policy.process_state_event(event).await { - tracing::warn!("Failed to process state event {}: {}", event_id_str, e); + // Parse state to get identifier for purgatory message + let identifier = 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 } + }) + .unwrap_or_else(|| "unknown".to_string()); - tracing::debug!("Accepted repository state: {}", event_id_str); + // Process state alignment asynchronously + match self.state_policy.process_state_event(event).await { + Ok(0) => { + // No repos aligned - event was added to purgatory + tracing::info!( + "State event {} added to purgatory: waiting for git data for identifier {}", + event_id_str, + identifier + ); + WritePolicyResult::Reject { + status: true, // Client sees OK + message: format!( + "purgatory: state event stored, waiting for git push for {}", + identifier + ) + .into(), + } + } + Ok(count) => { + // Successfully aligned repos + tracing::debug!( + "Accepted repository state {}: aligned {} repo(s)", + event_id_str, + count + ); WritePolicyResult::Accept } Err(e) => { - tracing::warn!("Failed to parse repository state {}: {}", event_id_str, e); - // Still accept the event even if we can't parse it - // The validation passed, so it's structurally valid + tracing::warn!("Failed to process state event {}: {}", event_id_str, e); + // Still accept the event even if processing failed WritePolicyResult::Accept } } @@ -173,6 +203,58 @@ impl Nip34WritePolicy { async fn handle_pr_event(&self, event: &Event) -> WritePolicyResult { let event_id_str = event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex()); + // Check if git data exists (checks placeholders and commit existence) + match self.pr_event_policy.check_git_data_exists(event).await { + Ok(false) => { + // No git data exists - add to purgatory + let commit = event + .tags + .iter() + .find_map(|tag| { + let tag_vec = tag.clone().to_vec(); + if tag_vec.len() >= 2 && tag_vec[0] == "c" { + Some(tag_vec[1].clone()) + } else { + None + } + }) + .unwrap_or_else(|| "unknown".to_string()); + + tracing::info!( + "PR event {} added to purgatory: waiting for git push with commit {}", + event_id_str, + commit + ); + + // Add to purgatory + self.ctx + .purgatory + .add_pr(event.clone(), event.id.to_hex(), commit.clone()); + + return WritePolicyResult::Reject { + status: true, // Client sees OK + message: format!( + "purgatory: PR event stored, waiting for git push with commit {}", + commit + ) + .into(), + }; + } + Ok(true) => { + // Git data exists - proceed with normal validation + tracing::debug!("Git data exists for PR event {}", event_id_str); + } + Err(e) => { + // Error checking git data - reject event + tracing::warn!( + "Failed to check git data for PR event {}: {}", + event_id_str, + e + ); + return WritePolicyResult::reject(format!("Failed to check git data: {}", e)); + } + } + // Validate refs/nostr refs for this PR event // This deletes any refs/nostr/ that points to wrong commit if let Err(e) = self.pr_event_policy.validate_nostr_ref(event).await { @@ -289,7 +371,10 @@ pub struct RelayWithDatabase { /// Returns a `RelayWithDatabase` struct containing: /// - The `LocalRelay` for handling WebSocket connections /// - The `SharedDatabase` for direct database queries (e.g., push authorization) -pub async fn create_relay(config: &Config) -> Result { +pub async fn create_relay( + config: &Config, + purgatory: Arc, +) -> Result { tracing::info!("Configuring nostr relay with GRASP-01 validation..."); // Determine database path @@ -337,7 +422,10 @@ pub async fn create_relay(config: &Config) -> Result { // Build relay with GRASP-01 validation // Clone Arc for the write policy so both relay and policy can access the database let git_data_path = config.effective_git_data_path(); - let write_policy = Nip34WritePolicy::new(&config.domain, database.clone(), &git_data_path); + + // Create write policy with purgatory integration + let write_policy = + Nip34WritePolicy::new(&config.domain, database.clone(), &git_data_path, purgatory); let relay = LocalRelayBuilder::default() .database(database.clone()) diff --git a/src/nostr/policy/mod.rs b/src/nostr/policy/mod.rs index 19db5f6..2a446fe 100644 --- a/src/nostr/policy/mod.rs +++ b/src/nostr/policy/mod.rs @@ -16,6 +16,8 @@ pub use related::{ReferenceResult, RelatedEventPolicy}; pub use state::{AlignmentResult, StatePolicy, StateResult}; use super::SharedDatabase; +use crate::purgatory::Purgatory; +use std::sync::Arc; /// Shared context for all sub-policies #[derive(Clone)] @@ -23,6 +25,7 @@ pub struct PolicyContext { pub domain: String, pub database: SharedDatabase, pub git_data_path: std::path::PathBuf, + pub purgatory: Arc, } impl PolicyContext { @@ -30,11 +33,13 @@ impl PolicyContext { domain: impl Into, database: SharedDatabase, git_data_path: impl Into, + purgatory: Arc, ) -> Self { Self { domain: domain.into(), database, git_data_path: git_data_path.into(), + purgatory, } } } diff --git a/src/nostr/policy/pr_event.rs b/src/nostr/policy/pr_event.rs index 53da369..c7602b0 100644 --- a/src/nostr/policy/pr_event.rs +++ b/src/nostr/policy/pr_event.rs @@ -19,6 +19,155 @@ impl PrEventPolicy { Self { ctx } } + /// Check if git data exists for a PR event + /// + /// This checks: + /// 1. If a placeholder exists (git-data-first scenario) + /// 2. If the commit exists in any relevant repository + /// + /// # Returns + /// - `Ok(true)` if git data ready (either placeholder found or commit exists) + /// - `Ok(false)` if git data missing (should add to purgatory) + /// - `Err(msg)` on errors + pub async fn check_git_data_exists(&self, event: &Event) -> Result { + let event_id = event.id.to_hex(); + + // Extract the `c` tag (commit hash) from the PR event + let commit = event.tags.iter().find_map(|tag| { + let tag_vec = tag.clone().to_vec(); + if tag_vec.len() >= 2 && tag_vec[0] == "c" { + Some(tag_vec[1].clone()) + } else { + None + } + }); + + let commit = match commit { + Some(c) => c, + None => { + return Err(format!("PR event {} has no 'c' tag", event_id)); + } + }; + + // Check for placeholder first (git-data-first scenario) + if let Some(placeholder_commit) = self.ctx.purgatory.find_pr_placeholder(&event_id) { + if placeholder_commit == commit { + // Perfect match - git data arrived first with matching commit + tracing::debug!( + "Found matching placeholder for PR event {} with commit {}", + event_id, + commit + ); + // Remove placeholder - event processing will continue normally + self.ctx.purgatory.remove_pr(&event_id); + return Ok(true); + } else { + // Placeholder has different commit - incoming event supersedes + tracing::info!( + "PR event {} supersedes placeholder: event expects commit {}, placeholder has {}", + event_id, + commit, + placeholder_commit + ); + // Remove placeholder with old commit data + self.ctx.purgatory.remove_pr(&event_id); + // TODO: Also remove git data (refs/nostr/) - Phase 5 + // Fall through to check if new commit exists + } + } + + // Check if commit exists in any repository referenced by this PR + // Extract ALL `a` tags (repository references) from the PR event + let repo_refs: Vec = event + .tags + .iter() + .filter_map(|tag| { + let tag_vec = tag.clone().to_vec(); + if tag_vec.len() >= 2 && tag_vec[0] == "a" && tag_vec[1].starts_with("30617:") { + Some(tag_vec[1].clone()) + } else { + None + } + }) + .collect(); + + if repo_refs.is_empty() { + // No repo references - cannot check git data + // This is unusual but let it through (other validation will catch issues) + return Ok(true); + } + + // Check each repository to see if commit exists + for repo_ref in repo_refs { + // Parse the repo reference: 30617:: + let parts: Vec<&str> = repo_ref.split(':').collect(); + if parts.len() < 3 { + continue; + } + + let repo_pubkey = match PublicKey::from_hex(parts[1]) { + Ok(pk) => pk, + Err(_) => continue, + }; + let identifier = parts[2]; + + // Look up repository announcement to get the npub for path + let filter = Filter::new() + .kind(Kind::from(KIND_REPOSITORY_ANNOUNCEMENT)) + .author(repo_pubkey) + .custom_tag( + SingleLetterTag::lowercase(Alphabet::D), + identifier.to_string(), + ); + + let announcements: Vec = match self.ctx.database.query(filter).await { + Ok(events) => events.into_iter().collect(), + Err(e) => { + tracing::warn!( + "Failed to query for repository announcement for PR {}: {}", + event_id, + e + ); + continue; + } + }; + + if announcements.is_empty() { + continue; + } + + // Check each matching announcement + for announcement_event in announcements { + let announcement = match RepositoryAnnouncement::from_event(announcement_event) { + Ok(a) => a, + Err(_) => continue, + }; + + // Build repository path + let repo_path = self.ctx.git_data_path.join(announcement.repo_path()); + + // Check if commit exists + if git::commit_exists(&repo_path, &commit) { + tracing::debug!( + "Found commit {} for PR event {} in repository {}", + commit, + event_id, + repo_path.display() + ); + return Ok(true); + } + } + } + + // No git data found - should add to purgatory + tracing::debug!( + "No git data found for PR event {} with commit {}", + event_id, + commit + ); + Ok(false) + } + /// Validate refs/nostr/ ref against a PR or PR Update event's `c` tag /// /// When a PR event (kind 1618) or PR Update event (kind 1619) is received, diff --git a/src/nostr/policy/state.rs b/src/nostr/policy/state.rs index 43349e2..5e749ed 100644 --- a/src/nostr/policy/state.rs +++ b/src/nostr/policy/state.rs @@ -66,6 +66,24 @@ impl StatePolicy { let state = RepositoryState::from_event(event.clone()) .map_err(|e| format!("Failed to parse state: {}", e))?; + // Check if ANY git repositories exist for this identifier (regardless of authorization) + // This helps us distinguish "no git data yet" from "not authorized" or "not latest" + let has_any_git_data = self.has_git_data_for_identifier(&state.identifier); + + if !has_any_git_data { + // No git data exists yet - add to purgatory + tracing::debug!( + "No git data found for identifier {}, adding state event {} to purgatory", + state.identifier, + event.id.to_hex() + ); + self.ctx + .purgatory + .add_state(event.clone(), state.identifier.clone(), event.pubkey); + // Return 0 repos aligned, but this is not an error + return Ok(0); + } + // Identify owner repositories for which this is the latest authorized state let owner_repos = self.identify_owner_repositories(&state).await?; let repo_count = owner_repos.len(); @@ -97,13 +115,48 @@ impl StatePolicy { ); } else { tracing::debug!( - "No owner repos to align for state - git data not available yet or not latest" + "No owner repos to align for state - git data exists but author not authorized or not latest" ); } Ok(total_aligned) } + /// Check if any git repositories exist for the given identifier + /// + /// Scans the git_data_path for any directories matching the pattern: + /// `/.git` + /// + /// This is used to distinguish "no git data yet" from "not authorized". + fn has_git_data_for_identifier(&self, identifier: &str) -> bool { + let git_data_path = &self.ctx.git_data_path; + + // Check if git_data_path exists + if !git_data_path.exists() { + return false; + } + + // Scan for any npub directories + let read_dir = match std::fs::read_dir(git_data_path) { + Ok(dir) => dir, + Err(_) => return false, + }; + + for entry in read_dir.flatten() { + if let Ok(file_type) = entry.file_type() { + if file_type.is_dir() { + // Check if /.git exists + let repo_path = entry.path().join(format!("{}.git", identifier)); + if repo_path.exists() { + return true; + } + } + } + } + + false + } + /// Check if this state event is the latest for its identifier among authorized authors /// /// A state is considered "latest" if no other state event in the database -- cgit v1.2.3