7 files changed, 64 insertions, 47 deletions

diff --git a/src/sync/algorithms.rs b/src/sync/algorithms.rs
index 7d87411..3063516 100644
--- a/src/sync/algorithms.rs
+++ b/src/sync/algorithms.rs
@@ -65,9 +65,7 @@ pub fn derive_relay_targets(
 
     for (repo_id, needs) in repo_index {
         for relay_url in &needs.relays {
-            let entry = relay_targets
-                .entry(relay_url.clone())
-                .or_insert_with(RelaySyncNeeds::default);
+            let entry = relay_targets.entry(relay_url.clone()).or_default();
 
             entry.repos.insert(repo_id.clone());
             entry.root_events.extend(needs.root_events.iter().cloned());
@@ -586,4 +584,4 @@ mod tests {
         );
         assert_eq!(actions[0].relay_url, "wss://new-relay.com");
     }
-}
\ No newline at end of file
+}
diff --git a/src/sync/filters.rs b/src/sync/filters.rs
index 02d580e..24e9bb2 100644
--- a/src/sync/filters.rs
+++ b/src/sync/filters.rs
@@ -337,4 +337,4 @@ mod tests {
 
         assert_eq!(filters.len(), 6);
     }
-}
\ No newline at end of file
+}
diff --git a/src/sync/health.rs b/src/sync/health.rs
index f9a5f3a..0ae7dee 100644
--- a/src/sync/health.rs
+++ b/src/sync/health.rs
@@ -206,11 +206,7 @@ impl RelayHealthTracker {
                 health.next_retry_at = Some(now + backoff);
 
                 if old_state != HealthState::Degraded {
-                    tracing::warn!(
-                        "Relay {} degraded, backoff {:?}",
-                        relay_url,
-                        backoff
-                    );
+                    tracing::warn!("Relay {} degraded, backoff {:?}", relay_url, backoff);
                 } else {
                     tracing::debug!(
                         "Relay {} failure #{}, backoff {:?}",
@@ -308,12 +304,17 @@ impl RelayHealthTracker {
 
     /// Get all tracked relay URLs
     pub fn get_tracked_relays(&self) -> Vec<String> {
-        self.health.iter().map(|entry| entry.key().clone()).collect()
+        self.health
+            .iter()
+            .map(|entry| entry.key().clone())
+            .collect()
     }
 
     /// Get a clone of the health info for a relay
     pub fn get_health(&self, relay_url: &str) -> Option<RelayHealth> {
-        self.health.get(relay_url).map(|entry| entry.value().clone())
+        self.health
+            .get(relay_url)
+            .map(|entry| entry.value().clone())
     }
 }
 
@@ -369,7 +370,7 @@ mod tests {
     fn test_backoff_increases_exponentially() {
         let base = DEFAULT_BASE_BACKOFF_SECS; // 5 seconds
         let max = 3600u64;
-        
+
         // failure 1: 5s (base * 2^0 = 5)
         assert_eq!(
             RelayHealthTracker::get_backoff_duration(1, base, max),
@@ -498,4 +499,4 @@ mod tests {
         let health = tracker.get_health("wss://nonexistent.example.com");
         assert!(health.is_none());
     }
-}
\ No newline at end of file
+}
diff --git a/src/sync/metrics.rs b/src/sync/metrics.rs
index 411ff63..d917dc0 100644
--- a/src/sync/metrics.rs
+++ b/src/sync/metrics.rs
@@ -207,7 +207,9 @@ impl SyncMetrics {
             HealthState::Degraded => 2,
             HealthState::Dead => 3,
         };
-        self.relay_status.with_label_values(&[relay]).set(state_value);
+        self.relay_status
+            .with_label_values(&[relay])
+            .set(state_value);
     }
 
     /// Record relay failure count.
@@ -259,9 +261,7 @@ impl SyncMetrics {
     /// * `source` - The event source type (see [`record_event`](Self::record_event))
     /// * `count` - Number of events to record
     pub fn record_events(&self, source: &str, count: u64) {
-        self.events_total
-            .with_label_values(&[source])
-            .inc_by(count);
+        self.events_total.with_label_values(&[source]).inc_by(count);
     }
 
     /// Record a gap event filled during catchup.
@@ -451,4 +451,4 @@ mod tests {
         let metrics2 = SyncMetrics::register(&registry);
         assert!(metrics2.is_err());
     }
-}
\ No newline at end of file
+}
diff --git a/src/sync/mod.rs b/src/sync/mod.rs
index c4c3c7f..fb59b3c 100644
--- a/src/sync/mod.rs
+++ b/src/sync/mod.rs
@@ -512,8 +512,8 @@ impl SyncManager {
         };
 
         // Check if relay supports NIP-77 negentropy AND negentropy is not disabled
-        let use_negentropy = !self.config.sync_disable_negentropy
-            && connection.supports_negentropy().await;
+        let use_negentropy =
+            !self.config.sync_disable_negentropy && connection.supports_negentropy().await;
 
         // Unsubscribe all current subscriptions
         connection.unsubscribe_all().await;
@@ -1657,12 +1657,12 @@ impl SyncManager {
 
                 let layer1_filters = 1;
                 let layer2_filters = if repo_count > 0 {
-                    ((repo_count + 99) / 100) * 3
+                    repo_count.div_ceil(100) * 3
                 } else {
                     0
                 };
                 let layer3_filters = if event_count > 0 {
-                    ((event_count + 99) / 100) * 3
+                    event_count.div_ceil(100) * 3
                 } else {
                     0
                 };
diff --git a/src/sync/relay_connection.rs b/src/sync/relay_connection.rs
index fae179b..4167a0c 100644
--- a/src/sync/relay_connection.rs
+++ b/src/sync/relay_connection.rs
@@ -150,17 +150,21 @@ impl RelayConnection {
         //
         // See: nostr-sdk-0.44 Client::try_connect_relay documentation
         self.client
-            .try_connect_relay(&self.url, std::time::Duration::from_secs(connection_timeout_secs))
+            .try_connect_relay(
+                &self.url,
+                std::time::Duration::from_secs(connection_timeout_secs),
+            )
             .await
             .map_err(|e| format!("Failed to connect to relay {}: {}", self.url, e))?;
 
         // Subscribe to Layer 1 (announcements)
         let filter = build_announcement_filter(since);
-        let output = self
-            .client
-            .subscribe(filter, None)
-            .await
-            .map_err(|e| format!("Failed to subscribe to announcements on {}: {}", self.url, e))?;
+        let output = self.client.subscribe(filter, None).await.map_err(|e| {
+            format!(
+                "Failed to subscribe to announcements on {}: {}",
+                self.url, e
+            )
+        })?;
 
         tracing::info!(url = %self.url, sub_id = %output.val, "Connected and subscribed to Layer 1 (announcements)");
         Ok(output.val)
@@ -250,7 +254,8 @@ impl RelayConnection {
                             }
                             RelayMessage::Closed { message: msg, .. } => {
                                 tracing::info!(relay = %url, message = %msg, "Relay closed subscription");
-                                let _ = event_sender.send(RelayEvent::Closed(msg.to_string())).await;
+                                let _ =
+                                    event_sender.send(RelayEvent::Closed(msg.to_string())).await;
                                 break;
                             }
                             _ => {}
@@ -421,7 +426,10 @@ impl RelayConnection {
     /// - Relay doesn't actually support NIP-77 (despite claiming to)
     /// - Network errors during reconciliation
     /// - Timeout during sync
-    pub async fn negentropy_sync_filter(&self, filter: Filter) -> Result<NegentropySyncResult, String> {
+    pub async fn negentropy_sync_filter(
+        &self,
+        filter: Filter,
+    ) -> Result<NegentropySyncResult, String> {
         // Use nostr-sdk's sync method which handles the NEG-OPEN/NEG-MSG exchange
         let sync_opts = SyncOptions::default();
 
diff --git a/src/sync/self_subscriber.rs b/src/sync/self_subscriber.rs
index f83b081..e29e45b 100644
--- a/src/sync/self_subscriber.rs
+++ b/src/sync/self_subscriber.rs
@@ -49,7 +49,12 @@ impl PendingUpdates {
     }
 
     /// Add or update a repo with its relays and root events
-    fn add_repo(&mut self, repo_id: String, relays: HashSet<String>, root_events: HashSet<EventId>) {
+    fn add_repo(
+        &mut self,
+        repo_id: String,
+        relays: HashSet<String>,
+        root_events: HashSet<EventId>,
+    ) {
         let entry = self.repos.entry(repo_id).or_insert_with(|| RepoSyncNeeds {
             relays: HashSet::new(),
             root_events: HashSet::new(),
@@ -251,9 +256,9 @@ impl SelfSubscriber {
     ///
     /// Returns true if any extracted relay URL contains our domain
     fn lists_our_relay(&self, event: &Event) -> bool {
-        Self::extract_relay_urls(event).iter().any(|url| {
-            url.contains(&self.relay_domain) || url == &self.own_relay_url
-        })
+        Self::extract_relay_urls(event)
+            .iter()
+            .any(|url| url.contains(&self.relay_domain) || url == &self.own_relay_url)
     }
 
     /// Main run loop
@@ -413,21 +418,21 @@ impl SelfSubscriber {
         if let Some(repo_sync) = index.get_mut(&repo_ref) {
             // Add event.id to root_events set in the index (immediate availability)
             repo_sync.root_events.insert(event.id);
-            
+
             // Clone the relays before releasing the lock - Layer 3 filters need to be
             // sent to the same relays as Layer 2 filters for this repo
             let relays = repo_sync.relays.clone();
-            
+
             // Release lock before modifying pending
             drop(index);
-            
+
             // Also add root event to pending - this ensures batch processing runs
             // and creates Layer 3 filters for events referencing this root event.
             // CRITICAL: Include relays so derive_relay_targets knows where to send filters!
             let mut root_events = HashSet::new();
             root_events.insert(event.id);
             pending.add_repo(repo_ref.clone(), relays.clone(), root_events);
-            
+
             tracing::debug!(
                 event_id = %event.id,
                 repo_ref = %repo_ref,
@@ -475,10 +480,12 @@ impl SelfSubscriber {
 
         for (repo_id, needs) in updates {
             // Merge with existing entry or insert new
-            let entry = index.entry(repo_id.clone()).or_insert_with(|| RepoSyncNeeds {
-                relays: HashSet::new(),
-                root_events: HashSet::new(),
-            });
+            let entry = index
+                .entry(repo_id.clone())
+                .or_insert_with(|| RepoSyncNeeds {
+                    relays: HashSet::new(),
+                    root_events: HashSet::new(),
+                });
             entry.relays.extend(needs.relays);
             entry.root_events.extend(needs.root_events);
 
@@ -556,7 +563,7 @@ fn clone_url_to_relay_url(clone_url: &str) -> Option<String> {
     } else {
         return None;
     };
-    
+
     // Extract just the host:port part (everything before the first /)
     let host_port = rest.split('/').next()?;
     Some(format!("{}{}", ws_scheme, host_port))
@@ -581,7 +588,7 @@ mod tests {
             Some("ws://localhost:3000".to_string())
         );
     }
-    
+
     #[test]
     fn test_clone_url_to_relay_url_with_port() {
         assert_eq!(
@@ -593,6 +600,9 @@ mod tests {
     #[test]
     fn test_clone_url_to_relay_url_unsupported() {
         assert_eq!(clone_url_to_relay_url("git://example.com/repo.git"), None);
-        assert_eq!(clone_url_to_relay_url("ssh://git@example.com/repo.git"), None);
+        assert_eq!(
+            clone_url_to_relay_url("ssh://git@example.com/repo.git"),
+            None
+        );
     }
-}
\ No newline at end of file
+}