purgatory: fix state event receive code

4 files changed, 149 insertions, 91 deletions

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<usize, String> {
+    /// Returns the true if git data already availale or false if added to purgatory
+    pub async fn process_state_event(&self, event: &Event) -> Result<WritePolicyResult> {
         // 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<PathBuf> {
+    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
+}