upleb.uk

Public git repos — served from a NIP-34 GRASP relay at git.upleb.uk

summaryrefslogtreecommitdiff
path: root/src/purgatory/mod.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/purgatory/mod.rs')
-rw-r--r--src/purgatory/mod.rs593
1 files changed, 593 insertions, 0 deletions
diff --git a/src/purgatory/mod.rs b/src/purgatory/mod.rs
new file mode 100644
index 0000000..18a55d5
--- /dev/null
+++ b/src/purgatory/mod.rs
@@ -0,0 +1,593 @@
1//! Purgatory: In-memory holding area for events awaiting git data.
2//!
3//! Solves the "which arrives first?" problem where either nostr events or git pushes
4//! can arrive in any order. Events and git data are held temporarily until their
5//! counterpart arrives, at which point they can be processed together.
6//!
7//! ## Architecture
8//!
9//! - **In-memory only**: Data is lost on restart (acceptable per spec)
10//! - **Thread-safe**: Uses DashMap for concurrent access from multiple handlers
11//! - **Automatic expiry**: Entries expire after 30 minutes by default
12//! - **Separate stores**: State events and PR events use different indexing strategies
13
14mod helpers;
15mod types;
16
17pub use helpers::{can_satisfy_state, extract_refs_from_state, get_unpushed_refs};
18pub use types::{PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry};
19
20use dashmap::DashMap;
21use nostr_sdk::prelude::*;
22use std::sync::Arc;
23use std::time::{Duration, Instant};
24
25/// Default expiry duration for purgatory entries (30 minutes)
26const DEFAULT_EXPIRY: Duration = Duration::from_secs(1800);
27
28/// Main purgatory structure holding events awaiting git data.
29///
30/// Provides thread-safe concurrent access to two separate stores:
31/// - State events indexed by repository identifier
32/// - PR events indexed by event ID
33#[derive(Clone)]
34pub struct Purgatory {
35 /// State events (kind 30618) indexed by repository identifier.
36 /// Multiple state events can wait for the same identifier (different maintainers).
37 state_events: Arc<DashMap<String, Vec<StatePurgatoryEntry>>>,
38
39 /// PR events (kind 1617/1618) or placeholders indexed by event ID (hex string).
40 /// Event ID is from the 'e' tag in the PR event itself.
41 pr_events: Arc<DashMap<String, PrPurgatoryEntry>>,
42}
43
44impl Purgatory {
45 /// Create a new empty purgatory.
46 pub fn new() -> Self {
47 Self {
48 state_events: Arc::new(DashMap::new()),
49 pr_events: Arc::new(DashMap::new()),
50 }
51 }
52
53 /// Add a state event to purgatory.
54 ///
55 /// The event will expire after the default duration unless matched with git data.
56 /// Multiple state events for the same identifier are allowed (from different authors).
57 ///
58 /// # Arguments
59 /// * `event` - The state event (kind 30618) to hold
60 /// * `identifier` - The repository identifier from the 'd' tag
61 /// * `author` - The event author's public key
62 pub fn add_state(&self, event: Event, identifier: String, author: PublicKey) {
63 let now = Instant::now();
64 let entry = StatePurgatoryEntry {
65 event,
66 identifier: identifier.clone(),
67 author,
68 created_at: now,
69 expires_at: now + DEFAULT_EXPIRY,
70 };
71
72 self.state_events.entry(identifier).or_default().push(entry);
73 }
74
75 /// Add a PR event to purgatory.
76 ///
77 /// The event will expire after the default duration unless matched with git data.
78 ///
79 /// # Arguments
80 /// * `event` - The PR event (kind 1617/1618) to hold
81 /// * `event_id` - The event ID (hex string) from the 'e' tag
82 /// * `commit` - The commit SHA from the 'c' tag
83 pub fn add_pr(&self, event: Event, event_id: String, commit: String) {
84 let now = Instant::now();
85 let entry = PrPurgatoryEntry {
86 event: Some(event),
87 commit,
88 created_at: now,
89 expires_at: now + DEFAULT_EXPIRY,
90 };
91
92 self.pr_events.insert(event_id, entry);
93 }
94
95 /// Add a PR placeholder (git data arrived before PR event).
96 ///
97 /// Creates a placeholder entry waiting for the corresponding PR event.
98 ///
99 /// # Arguments
100 /// * `event_id` - The expected event ID (from git ref name)
101 /// * `commit` - The commit SHA that was pushed
102 pub fn add_pr_placeholder(&self, event_id: String, commit: String) {
103 let now = Instant::now();
104 let entry = PrPurgatoryEntry {
105 event: None, // Placeholder - no event yet
106 commit,
107 created_at: now,
108 expires_at: now + DEFAULT_EXPIRY,
109 };
110
111 self.pr_events.insert(event_id, entry);
112 }
113
114 /// Find state events waiting for a specific repository identifier.
115 ///
116 /// Returns all state events (from all maintainers) waiting for git data
117 /// matching this identifier.
118 ///
119 /// # Arguments
120 /// * `identifier` - The repository identifier to search for
121 ///
122 /// # Returns
123 /// Vector of state events waiting for this identifier, or empty vec if none found
124 pub fn find_state(&self, identifier: &str) -> Vec<StatePurgatoryEntry> {
125 self.state_events
126 .get(identifier)
127 .map(|entries| entries.clone())
128 .unwrap_or_default()
129 }
130
131 /// Find a PR event or placeholder by event ID.
132 ///
133 /// # Arguments
134 /// * `event_id` - The event ID to search for
135 ///
136 /// # Returns
137 /// The PR entry if found, None otherwise
138 pub fn find_pr(&self, event_id: &str) -> Option<PrPurgatoryEntry> {
139 self.pr_events.get(event_id).map(|entry| entry.clone())
140 }
141
142 /// Find a PR placeholder specifically (git-data-first scenario).
143 ///
144 /// Returns the commit SHA only if a placeholder exists (entry with no event).
145 /// Used to distinguish placeholders from actual PR events.
146 ///
147 /// # Arguments
148 /// * `event_id` - The event ID to search for
149 ///
150 /// # Returns
151 /// Some(commit_sha) if a placeholder exists, None if no entry or entry has an event
152 pub fn find_pr_placeholder(&self, event_id: &str) -> Option<String> {
153 self.pr_events.get(event_id).and_then(|entry| {
154 if entry.event.is_none() {
155 Some(entry.commit.clone())
156 } else {
157 None
158 }
159 })
160 }
161
162 /// Remove a state event from purgatory.
163 ///
164 /// Removes all entries for the given identifier.
165 ///
166 /// # Arguments
167 /// * `identifier` - The repository identifier to remove
168 pub fn remove_state(&self, identifier: &str) {
169 self.state_events.remove(identifier);
170 }
171
172 /// Remove a specific state event by comparing the full event.
173 ///
174 /// This allows removing a single state event while leaving others
175 /// for the same identifier intact.
176 ///
177 /// # Arguments
178 /// * `identifier` - The repository identifier
179 /// * `event_id` - The specific event ID to remove
180 pub fn remove_state_event(&self, identifier: &str, event_id: &EventId) {
181 if let Some(mut entries) = self.state_events.get_mut(identifier) {
182 entries.retain(|entry| entry.event.id != *event_id);
183 if entries.is_empty() {
184 drop(entries); // Release lock before removal
185 self.state_events.remove(identifier);
186 }
187 }
188 }
189
190 /// Find state events that could be satisfied by ref updates.
191 ///
192 /// Returns state events waiting for this identifier where applying the
193 /// ref updates to local state results in exactly the declared state.
194 /// Uses late-binding ref extraction at git push time.
195 ///
196 /// # Arguments
197 /// * `identifier` - The repository identifier to search for
198 /// * `pushed_updates` - Ref updates in the current push operation
199 /// * `local_refs` - Refs already existing locally (ref_name -> SHA)
200 ///
201 /// # Returns
202 /// Vector of events that can be satisfied by the push
203 pub fn find_matching_states(
204 &self,
205 identifier: &str,
206 pushed_updates: &[RefUpdate],
207 local_refs: &std::collections::HashMap<String, String>,
208 ) -> Vec<Event> {
209 self.state_events
210 .get(identifier)
211 .map(|entries| {
212 entries
213 .iter()
214 .filter(|entry| {
215 helpers::can_satisfy_state(&entry.event, pushed_updates, local_refs)
216 })
217 .map(|entry| entry.event.clone())
218 .collect()
219 })
220 .unwrap_or_default()
221 }
222
223 /// Extend expiry for state events about to be processed.
224 ///
225 /// Ensures entries have at least `duration` remaining on their timer.
226 /// Sets expiry to max(current_expiry, now + duration).
227 ///
228 /// # Arguments
229 /// * `identifier` - The repository identifier
230 /// * `event_ids` - Event IDs to extend expiry for
231 /// * `duration` - Minimum duration to guarantee from now
232 pub fn extend_expiry(&self, identifier: &str, event_ids: &[EventId], duration: Duration) {
233 if let Some(mut entries) = self.state_events.get_mut(identifier) {
234 let now = Instant::now();
235 let new_expiry = now + duration;
236
237 for entry in entries.iter_mut() {
238 if event_ids.contains(&entry.event.id) {
239 // Set to max of current expiry and new expiry
240 if entry.expires_at < new_expiry {
241 entry.expires_at = new_expiry;
242 }
243 }
244 }
245 }
246 }
247
248 /// Remove a PR event or placeholder from purgatory.
249 ///
250 /// # Arguments
251 /// * `event_id` - The event ID to remove
252 pub fn remove_pr(&self, event_id: &str) {
253 self.pr_events.remove(event_id);
254 }
255
256 /// Get all PR placeholder event IDs (git-data-first entries without events).
257 ///
258 /// Returns event IDs for entries where git data arrived before the PR event.
259 /// These correspond to `refs/nostr/<event-id>` refs that should be cleaned up
260 /// on shutdown since they don't have corresponding events.
261 ///
262 /// # Returns
263 /// Vector of event IDs (hex strings) for placeholder entries
264 pub fn get_placeholder_event_ids(&self) -> Vec<String> {
265 self.pr_events
266 .iter()
267 .filter_map(|entry| {
268 if entry.value().event.is_none() {
269 Some(entry.key().clone())
270 } else {
271 None
272 }
273 })
274 .collect()
275 }
276
277 /// Remove expired entries from purgatory.
278 ///
279 /// Should be called periodically (every 60 seconds) by background task to clean up
280 /// entries that have exceeded their expiry deadline.
281 ///
282 /// # Returns
283 /// Tuple of (num_state_removed, num_pr_removed)
284 pub fn cleanup(&self) -> (usize, usize) {
285 let now = Instant::now();
286 let mut state_removed = 0;
287
288 // Remove expired state events
289 self.state_events.retain(|_, entries| {
290 let original_len = entries.len();
291 entries.retain(|entry| entry.expires_at > now);
292 state_removed += original_len - entries.len();
293 !entries.is_empty()
294 });
295
296 // Remove expired PR events
297 let expired_prs: Vec<String> = self
298 .pr_events
299 .iter()
300 .filter(|entry| entry.value().expires_at <= now)
301 .map(|entry| entry.key().clone())
302 .collect();
303
304 let pr_removed = expired_prs.len();
305 for event_id in expired_prs {
306 self.pr_events.remove(&event_id);
307 }
308
309 (state_removed, pr_removed)
310 }
311
312 /// Remove expired entries from purgatory (legacy method).
313 ///
314 /// # Returns
315 /// Total number of entries removed (state + PR events)
316 #[deprecated(since = "0.1.0", note = "Use cleanup() instead for separate counts")]
317 pub fn remove_expired(&self) -> usize {
318 let (state, pr) = self.cleanup();
319 state + pr
320 }
321
322 /// Get current count of entries in purgatory.
323 ///
324 /// # Returns
325 /// Tuple of (state_event_count, pr_event_count)
326 pub fn count(&self) -> (usize, usize) {
327 let state_count: usize = self.state_events.iter().map(|e| e.value().len()).sum();
328 let pr_count = self.pr_events.len();
329 (state_count, pr_count)
330 }
331
332 /// Clear all entries from purgatory (for testing).
333 #[cfg(test)]
334 pub fn clear(&self) {
335 self.state_events.clear();
336 self.pr_events.clear();
337 }
338}
339
340impl Default for Purgatory {
341 fn default() -> Self {
342 Self::new()
343 }
344}
345
346#[cfg(test)]
347mod tests {
348 use super::*;
349
350 #[test]
351 fn test_purgatory_creation() {
352 let purgatory = Purgatory::new();
353 let (state_count, pr_count) = purgatory.count();
354 assert_eq!(state_count, 0);
355 assert_eq!(pr_count, 0);
356 }
357
358 #[test]
359 fn test_purgatory_count() {
360 let purgatory = Purgatory::new();
361
362 // Add some test data
363 let keys = Keys::generate();
364 let event = EventBuilder::text_note("test")
365 .sign_with_keys(&keys)
366 .unwrap();
367
368 purgatory.add_state(event.clone(), "test-repo".to_string(), keys.public_key());
369 purgatory.add_pr(event, "test-event-id".to_string(), "abc123".to_string());
370
371 let (state_count, pr_count) = purgatory.count();
372 assert_eq!(state_count, 1);
373 assert_eq!(pr_count, 1);
374 }
375}
376
377#[test]
378fn test_pr_event_vs_placeholder() {
379 let purgatory = Purgatory::new();
380 let keys = Keys::generate();
381 let event = EventBuilder::text_note("test PR")
382 .sign_with_keys(&keys)
383 .unwrap();
384
385 // Add a PR event with actual event
386 purgatory.add_pr(
387 event.clone(),
388 "event-id-1".to_string(),
389 "commit-abc".to_string(),
390 );
391
392 // Add a placeholder (no event)
393 purgatory.add_pr_placeholder("event-id-2".to_string(), "commit-def".to_string());
394
395 // find_pr should find both
396 assert!(purgatory.find_pr("event-id-1").is_some());
397 assert!(purgatory.find_pr("event-id-2").is_some());
398
399 // find_pr_placeholder should only find the placeholder
400 assert!(purgatory.find_pr_placeholder("event-id-1").is_none());
401 assert_eq!(
402 purgatory.find_pr_placeholder("event-id-2"),
403 Some("commit-def".to_string())
404 );
405}
406
407#[test]
408fn test_pr_placeholder_creation_and_retrieval() {
409 let purgatory = Purgatory::new();
410
411 // Add a placeholder
412 purgatory.add_pr_placeholder("placeholder-id".to_string(), "commit-123".to_string());
413
414 // Should be findable by find_pr
415 let entry = purgatory.find_pr("placeholder-id");
416 assert!(entry.is_some());
417 let entry = entry.unwrap();
418 assert!(entry.event.is_none()); // No event yet
419 assert_eq!(entry.commit, "commit-123");
420
421 // Should be findable by find_pr_placeholder
422 let commit = purgatory.find_pr_placeholder("placeholder-id");
423 assert_eq!(commit, Some("commit-123".to_string()));
424}
425
426#[test]
427fn test_cleanup_removes_expired_entries() {
428 use std::time::Duration;
429
430 let purgatory = Purgatory::new();
431 let keys = Keys::generate();
432
433 // Create events
434 let state_event = EventBuilder::text_note("state event")
435 .sign_with_keys(&keys)
436 .unwrap();
437 let pr_event = EventBuilder::text_note("pr event")
438 .sign_with_keys(&keys)
439 .unwrap();
440
441 // Add entries to purgatory
442 purgatory.add_state(
443 state_event.clone(),
444 "test-repo".to_string(),
445 keys.public_key(),
446 );
447 purgatory.add_pr(pr_event, "pr-123".to_string(), "commit-abc".to_string());
448 purgatory.add_pr_placeholder("pr-456".to_string(), "commit-def".to_string());
449
450 // Verify entries are there
451 let (state_count, pr_count) = purgatory.count();
452 assert_eq!(state_count, 1);
453 assert_eq!(pr_count, 2);
454
455 // Manually expire entries by modifying their expiry time
456 // (This is a bit hacky but needed for testing without waiting 30 minutes)
457 if let Some(mut entries) = purgatory.state_events.get_mut("test-repo") {
458 for entry in entries.iter_mut() {
459 entry.expires_at = Instant::now() - Duration::from_secs(1);
460 }
461 }
462
463 // Expire PR events
464 for mut entry in purgatory.pr_events.iter_mut() {
465 entry.value_mut().expires_at = Instant::now() - Duration::from_secs(1);
466 }
467
468 // Run cleanup
469 let (state_removed, pr_removed) = purgatory.cleanup();
470
471 // Verify counts
472 assert_eq!(state_removed, 1);
473 assert_eq!(pr_removed, 2);
474
475 // Verify entries are gone
476 let (state_count, pr_count) = purgatory.count();
477 assert_eq!(state_count, 0);
478 assert_eq!(pr_count, 0);
479}
480
481#[test]
482fn test_cleanup_preserves_non_expired_entries() {
483 let purgatory = Purgatory::new();
484 let keys = Keys::generate();
485
486 let state_event = EventBuilder::text_note("state event")
487 .sign_with_keys(&keys)
488 .unwrap();
489 let pr_event = EventBuilder::text_note("pr event")
490 .sign_with_keys(&keys)
491 .unwrap();
492
493 // Add fresh entries
494 purgatory.add_state(state_event, "test-repo".to_string(), keys.public_key());
495 purgatory.add_pr(pr_event, "pr-123".to_string(), "commit-abc".to_string());
496
497 // Run cleanup
498 let (state_removed, pr_removed) = purgatory.cleanup();
499
500 // Nothing should be removed
501 assert_eq!(state_removed, 0);
502 assert_eq!(pr_removed, 0);
503
504 // Verify entries are still there
505 let (state_count, pr_count) = purgatory.count();
506 assert_eq!(state_count, 1);
507 assert_eq!(pr_count, 1);
508}
509
510#[test]
511fn test_cleanup_mixed_expired_and_fresh() {
512 use std::time::Duration;
513
514 let purgatory = Purgatory::new();
515 let keys = Keys::generate();
516
517 // Add multiple state events for same repo
518 let event1 = EventBuilder::text_note("event1")
519 .sign_with_keys(&keys)
520 .unwrap();
521 let event2 = EventBuilder::text_note("event2")
522 .sign_with_keys(&keys)
523 .unwrap();
524
525 purgatory.add_state(event1, "test-repo".to_string(), keys.public_key());
526 purgatory.add_state(event2, "test-repo".to_string(), keys.public_key());
527
528 // Expire only the first one
529 if let Some(mut entries) = purgatory.state_events.get_mut("test-repo") {
530 if let Some(entry) = entries.get_mut(0) {
531 entry.expires_at = Instant::now() - Duration::from_secs(1);
532 }
533 }
534
535 // Add PR events
536 let pr1 = EventBuilder::text_note("pr1")
537 .sign_with_keys(&keys)
538 .unwrap();
539 let pr2 = EventBuilder::text_note("pr2")
540 .sign_with_keys(&keys)
541 .unwrap();
542
543 purgatory.add_pr(pr1, "pr-1".to_string(), "commit-1".to_string());
544 purgatory.add_pr(pr2, "pr-2".to_string(), "commit-2".to_string());
545
546 // Expire only first PR
547 if let Some(mut entry) = purgatory.pr_events.get_mut("pr-1") {
548 entry.expires_at = Instant::now() - Duration::from_secs(1);
549 }
550
551 // Run cleanup
552 let (state_removed, pr_removed) = purgatory.cleanup();
553
554 // One of each should be removed
555 assert_eq!(state_removed, 1);
556 assert_eq!(pr_removed, 1);
557
558 // Verify remaining counts
559 let (state_count, pr_count) = purgatory.count();
560 assert_eq!(state_count, 1); // One state event remains
561 assert_eq!(pr_count, 1); // One PR event remains
562}
563
564#[test]
565fn test_remove_expired_legacy_method() {
566 use std::time::Duration;
567
568 let purgatory = Purgatory::new();
569 let keys = Keys::generate();
570
571 let state_event = EventBuilder::text_note("state")
572 .sign_with_keys(&keys)
573 .unwrap();
574 let pr_event = EventBuilder::text_note("pr").sign_with_keys(&keys).unwrap();
575
576 purgatory.add_state(state_event, "repo".to_string(), keys.public_key());
577 purgatory.add_pr(pr_event, "pr-id".to_string(), "commit".to_string());
578
579 // Expire both
580 if let Some(mut entries) = purgatory.state_events.get_mut("repo") {
581 for entry in entries.iter_mut() {
582 entry.expires_at = Instant::now() - Duration::from_secs(1);
583 }
584 }
585 for mut entry in purgatory.pr_events.iter_mut() {
586 entry.value_mut().expires_at = Instant::now() - Duration::from_secs(1);
587 }
588
589 // Test legacy method returns total
590 #[allow(deprecated)]
591 let total = purgatory.remove_expired();
592 assert_eq!(total, 2); // 1 state + 1 PR
593}