From 3d6901831904141166d9ed8f47813c45cba109b6 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Wed, 31 Dec 2025 09:17:49 +0000 Subject: purgatory: fix state event receive code --- src/git/authorization.rs | 1 - src/git/mod.rs | 24 +++++++ src/nostr/builder.rs | 47 ++----------- src/nostr/policy/state.rs | 168 ++++++++++++++++++++++++++++++++-------------- 4 files changed, 149 insertions(+), 91 deletions(-) (limited to 'src') diff --git a/src/git/authorization.rs b/src/git/authorization.rs index 9bcbdf8..6b997d8 100644 --- a/src/git/authorization.rs +++ b/src/git/authorization.rs @@ -149,7 +149,6 @@ pub async fn authorize_push( "Found {} non-refs/nostr/ refs - checking state authorization", state_refs.len() ); - let auth_result = get_state_authorization_for_specific_owner_repo( database, identifier, diff --git a/src/git/mod.rs b/src/git/mod.rs index 5c99b3e..1847c8c 100644 --- a/src/git/mod.rs +++ b/src/git/mod.rs @@ -74,6 +74,30 @@ pub fn commit_exists(repo_path: &Path, commit_hash: &str) -> bool { } } +/// Check if a oid exists in the repository +/// +/// # Arguments +/// * `repo_path` - Path to the bare git repository +/// * `oid` - The commit hash to check +/// +/// # Returns +/// True if the commit exists in the repository, false otherwise +pub fn oid_exists(repo_path: &Path, oid: &str) -> bool { + let output = Command::new("git") + .args(["cat-file", "-e", oid]) + .current_dir(repo_path) + .output(); + + match output { + Ok(result) => result.status.success(), + Err(_) => false, + } +} + +pub fn is_valid_oid(oid: &str) -> bool { + oid.len() >= 5 && oid.len() <= 40 && oid.chars().all(|c| c.is_digit(16)) +} + /// Set the repository HEAD to point to a branch /// /// This updates the HEAD symbolic ref to point to the specified branch. diff --git a/src/nostr/builder.rs b/src/nostr/builder.rs index 2b4d524..37fa025 100644 --- a/src/nostr/builder.rs +++ b/src/nostr/builder.rs @@ -144,51 +144,16 @@ impl Nip34WritePolicy { match self.state_policy.validate(event) { StateResult::Accept => { - // 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()); - // 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 - } + Ok(poilicy_result) => poilicy_result, Err(e) => { tracing::warn!("Failed to process state event {}: {}", event_id_str, e); - // Still accept the event even if processing failed - WritePolicyResult::Accept + // reject if processing failed + WritePolicyResult::Reject { + status: false, + message: format!("error: {e}").into(), + } } } } diff --git a/src/nostr/policy/state.rs b/src/nostr/policy/state.rs index 5e749ed..13f2549 100644 --- a/src/nostr/policy/state.rs +++ b/src/nostr/policy/state.rs @@ -1,3 +1,7 @@ +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use nostr_relay_builder::builder::WritePolicyResult; /// State Policy - State event validation + ref alignment /// /// Handles validation of NIP-34 repository state events (kind 30618) @@ -5,7 +9,8 @@ use nostr_relay_builder::prelude::{Alphabet, Event, Filter, Kind, PublicKey, SingleLetterTag}; use super::PolicyContext; -use crate::git; +use crate::git::authorization::{collect_authorized_maintainers, fetch_repository_data}; +use crate::git::{self}; use crate::nostr::events::{ validate_state, RepositoryAnnouncement, RepositoryState, KIND_REPOSITORY_ANNOUNCEMENT, KIND_REPOSITORY_STATE, @@ -60,66 +65,107 @@ impl StatePolicy { /// Process a state event: validate and align owner repositories /// - /// Returns the number of repositories aligned if successful. - pub async fn process_state_event(&self, event: &Event) -> Result { + /// Returns the true if git data already availale or false if added to purgatory + pub async fn process_state_event(&self, event: &Event) -> Result { // Parse state to get HEAD and branch info - 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 + let state = + RepositoryState::from_event(event.clone()).context("Failed to parse state event")?; + + // duplicate check in purgatory + if self + .ctx + .purgatory + .find_state(&state.identifier) + .iter() + .any(|e| e.event.id.eq(&event.id)) + { tracing::debug!( - "No git data found for identifier {}, adding state event {} to purgatory", - state.identifier, - event.id.to_hex() + "processed state event duplicate (already in purgatory): {}", + event.id, ); - 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); + return Ok(WritePolicyResult::Reject { + status: true, // Client sees OK + message: "duplicate: in purgatory".into(), + }); + } + // get all repositories and state events from db with identifier + let db_repo_data = fetch_repository_data(&self.ctx.database, &state.identifier).await?; + + // duplicate check in db + if db_repo_data.states.iter().any(|e| e.event.id.eq(&event.id)) { + tracing::debug!("processed state event duplicate (in db): {}", event.id,); + return Ok(WritePolicyResult::Reject { + status: true, // Client sees OK + message: "duplicate".into(), + }); } - // 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(); - let mut total_aligned = 0; - - // Align each owner repository with the authorized state - for (_announcement, repo_path) in owner_repos { - let result = self.align_repository_with_state(&repo_path, &state); - - if result.has_changes() { - tracing::info!( - "Aligned {} with state: created={}, updated={}, deleted={}, head_set={}", - repo_path.display(), - result.refs_created, - result.refs_updated, - result.refs_deleted, - result.head_set - ); - total_aligned += 1; + // check if git data is avialable + if let Some(repo_with_git_data) = + find_repo_with_git_data(&db_repo_data.announcements, &state, &self.ctx.git_data_path) + { + tracing::debug!( + "processing state event git as data already available: {}", + event.id, + ); + // find repos for which this state is authorised and align the git refs to this state + let by_owner = collect_authorized_maintainers(&db_repo_data.announcements); + let mut repo_count = 0; + for (owner, maintainers) in by_owner { + if maintainers.contains(&event.pubkey.to_string()) { + if let Some(previous_state) = db_repo_data + .states + .iter() + .filter(|e| maintainers.contains(&e.event.pubkey.to_string())) + .max_by_key(|e| e.event.created_at) + { + // TODO in event of a tie the event with the biggest event id wins + if state.event.created_at > previous_state.event.created_at { + if let Some(annoucement) = db_repo_data + .announcements + .iter() + .find(|a| a.event.pubkey.to_string().eq(&owner)) + { + let repo_path = + self.ctx.git_data_path.join(annoucement.repo_path().clone()); + // TODO - if repo_path != repo_with_git_data, pass as a datasource for missing data? + let result = self.align_repository_with_state(&repo_path, &state); + repo_count += 1; + tracing::info!( + "Aligned {} with state: created={}, updated={}, deleted={}, head_set={}", + repo_path.display(), + result.refs_created, + result.refs_updated, + result.refs_deleted, + result.head_set + ); + } + } + } + } } - } - if repo_count > 0 { tracing::info!( - "Processed state event for {} repo(s) ({} aligned) with identifier {}", - repo_count, - total_aligned, - state.identifier + "immediately accepting state event. Was latest authorised state and git data updated for {repo_count} repositories: eventid: {}", + state.event.id, ); + // immediately accept the event, bypassing purgatory + Ok(WritePolicyResult::Accept) // event should be saved and broadcast } else { - tracing::debug!( - "No owner repos to align for state - git data exists but author not authorized or not latest" + // if no git data - add to purgatory + self.ctx + .purgatory + .add_state(event.clone(), state.identifier.clone(), event.pubkey); + tracing::info!( + "state event added to purgatory: eventid: {}, identifier: {}", + state.event.id, + state.identifier, ); + Ok(WritePolicyResult::Reject { + status: true, // Client sees OK + message: "purgatory: won't be served until git data arrives".into(), + }) } - - Ok(total_aligned) } /// Check if any git repositories exist for the given identifier @@ -473,3 +519,27 @@ impl StatePolicy { result } } + +fn find_repo_with_git_data( + announcements: &[RepositoryAnnouncement], + state: &RepositoryState, + git_data_path: &Path, +) -> Option { + for announcement in announcements { + let repo_path = git_data_path.join(announcement.repo_path().clone()); + if state.branches.iter().all(|branch_state| { + if branch_state.commit.starts_with("ref: ") { + true // ignore symlinks + } else { + git::oid_exists(&repo_path, &branch_state.commit) + } + }) && state + .tags + .iter() + .all(|tag_state| git::oid_exists(&repo_path, &tag_state.commit)) + { + return Some(repo_path); + } + } + None +} -- cgit v1.2.3