feat(sync): Phase 1 MVP - single relay proactive sync

- Add src/sync/ module with SyncManager
- Add NGIT_SYNC_RELAY_URL config option
- Subscribe to kind 30617 on configured relay
- Validate synced events through Nip34WritePolicy
- Integration test with two TestRelay instances

src/config.rs

@@ -83,6 +83,10 @@ pub struct Config {
     /// Number of top bandwidth repos to track in metrics
     #[arg(long = "metrics-top-n-repos", env = "NGIT_METRICS_TOP_N_REPOS", default_value_t = 10)]
     pub metrics_top_n_repos: usize,
+
+    /// URL of relay to sync kind 30617 events from (optional, enables proactive sync)
+    #[arg(long, env = "NGIT_SYNC_RELAY_URL")]
+    pub sync_relay_url: Option<String>,
 }
 
 impl Config {
@@ -138,6 +142,7 @@ impl Config {
             metrics_enabled: true,
             metrics_connection_per_ip_abuse_threshold: 10,
             metrics_top_n_repos: 10,
+            sync_relay_url: None,
         }
     }
 }

src/http/nip11.rs

@@ -105,6 +105,7 @@ mod tests {
             metrics_enabled: true,
             metrics_connection_per_ip_abuse_threshold: 10,
             metrics_top_n_repos: 10,
+            sync_relay_url: None,
         };
 
         let doc = RelayInformationDocument::from_config(&config);
@@ -139,6 +140,7 @@ mod tests {
             metrics_enabled: true,
             metrics_connection_per_ip_abuse_threshold: 10,
             metrics_top_n_repos: 10,
+            sync_relay_url: None,
         };
 
         let doc = RelayInformationDocument::from_config(&config);

src/lib.rs

@@ -3,3 +3,4 @@ pub mod git;
 pub mod http;
 pub mod metrics;
 pub mod nostr;
+pub mod sync;

src/main.rs

@@ -9,6 +9,7 @@ use ngit_grasp::{
     http,
     metrics::Metrics,
     nostr,
+    sync::SyncManager,
 };
 
 #[tokio::main]
@@ -50,6 +51,21 @@ async fn main() -> Result<()> {
             config.domain
         );
 
+        // Start SyncManager if sync_relay_url is configured
+        if let Some(ref sync_url) = config.sync_relay_url {
+            info!("Starting proactive sync from: {}", sync_url);
+            let sync_manager = SyncManager::new(
+                sync_url.clone(),
+                relay_with_db.database.clone(),
+                relay_with_db.write_policy.clone(),
+            );
+            tokio::spawn(async move {
+                sync_manager.run().await;
+            });
+        } else {
+            info!("Proactive sync disabled (no NGIT_SYNC_RELAY_URL configured)");
+        }
+
         // Start HTTP server with integrated relay and database
         info!("Starting HTTP server on {}", config.bind_address);
         http::run_server(config, relay_with_db.relay, relay_with_db.database, metrics).await?;

src/nostr/builder.rs

@@ -269,12 +269,14 @@ impl WritePolicy for Nip34WritePolicy {
     }
 }
 
-/// Result of creating a relay - includes both the relay and database
+/// Result of creating a relay - includes relay, database, and write policy
 pub struct RelayWithDatabase {
     /// The local relay instance
     pub relay: LocalRelay,
     /// The database Arc that can be used for direct queries
     pub database: SharedDatabase,
+    /// The write policy used for event validation
+    pub write_policy: Nip34WritePolicy,
 }
 
 /// Create a configured LocalRelay with full GRASP-01 validation
@@ -330,13 +332,11 @@ pub fn create_relay(config: &Config) -> Result<RelayWithDatabase> {
     // 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);
+
     let builder = RelayBuilder::default()
         .database(database.clone())
-        .write_policy(Nip34WritePolicy::new(
-            &config.domain,
-            database.clone(),
-            &git_data_path,
-        ));
+        .write_policy(write_policy.clone());
 
     tracing::info!(
         "Relay configured with GRASP-01 validation for domain: {}",
@@ -346,5 +346,6 @@ pub fn create_relay(config: &Config) -> Result<RelayWithDatabase> {
     Ok(RelayWithDatabase {
         relay: LocalRelay::new(builder),
         database,
+        write_policy,
     })
 }

src/sync/connection.rs

@@ -0,0 +1,143 @@
+//! WebSocket connection handling for sync
+//!
+//! Manages the connection to a source relay, subscribes to kind 30617 events,
+//! and passes them through validation.
+
+use std::time::Duration;
+
+use nostr_sdk::prelude::*;
+use tokio::sync::mpsc;
+
+use super::KIND_REPOSITORY_STATE;
+
+/// Event received from the sync connection
+#[derive(Debug, Clone)]
+pub struct SyncedEvent {
+    pub event: Event,
+    pub source_url: String,
+}
+
+/// Manages a WebSocket connection to a single relay for syncing
+pub struct SyncConnection {
+    url: String,
+    client: Client,
+}
+
+impl SyncConnection {
+    /// Create a new sync connection to the given relay URL
+    pub async fn new(url: &str) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
+        let client = Client::default();
+
+        // Add the relay
+        client.add_relay(url).await?;
+
+        // Connect to the relay
+        client.connect().await;
+
+        tracing::info!("Sync connection established to {}", url);
+
+        Ok(Self {
+            url: url.to_string(),
+            client,
+        })
+    }
+
+    /// Start receiving events and send them through the channel
+    ///
+    /// This method runs indefinitely, reconnecting as needed.
+    pub async fn run(self, tx: mpsc::Sender<SyncedEvent>) {
+        // Create filter for kind 30617 (repository state) events
+        let filter = Filter::new().kind(Kind::Custom(KIND_REPOSITORY_STATE));
+
+        // Subscribe to events
+        match self.client.subscribe(filter, None).await {
+            Ok(output) => {
+                tracing::info!(
+                    "Subscribed to kind {} events on {} (subscription: {})",
+                    KIND_REPOSITORY_STATE,
+                    self.url,
+                    output.id()
+                );
+            }
+            Err(e) => {
+                tracing::error!("Failed to subscribe on {}: {}", self.url, e);
+                return;
+            }
+        };
+
+        // Handle incoming notifications
+        let url = self.url.clone();
+        self.client
+            .handle_notifications(|notification| {
+                let tx = tx.clone();
+                let url = url.clone();
+                async move {
+                    match notification {
+                        RelayPoolNotification::Event { event, .. } => {
+                            tracing::debug!(
+                                "Received event {} from {} (kind {})",
+                                event.id,
+                                url,
+                                event.kind.as_u16()
+                            );
+
+                            // Send the event to the manager for processing
+                            let synced = SyncedEvent {
+                                event: (*event).clone(),
+                                source_url: url.clone(),
+                            };
+
+                            if let Err(e) = tx.send(synced).await {
+                                tracing::warn!("Failed to send synced event: {}", e);
+                                return Ok(true); // Stop if channel is closed
+                            }
+                        }
+                        RelayPoolNotification::Shutdown => {
+                            tracing::warn!("Relay connection shutdown for {}", url);
+                            return Ok(true); // Stop on shutdown
+                        }
+                        RelayPoolNotification::Message { message, .. } => {
+                            tracing::trace!("Received message from {}: {:?}", url, message);
+                        }
+                    }
+                    Ok(false) // Continue processing
+                }
+            })
+            .await
+            .ok();
+    }
+
+}
+
+/// Reconnect loop with exponential backoff
+pub async fn connect_with_retry(
+    url: &str,
+    tx: mpsc::Sender<SyncedEvent>,
+) {
+    let mut backoff = Duration::from_secs(1);
+    let max_backoff = Duration::from_secs(60);
+
+    loop {
+        match SyncConnection::new(url).await {
+            Ok(conn) => {
+                backoff = Duration::from_secs(1); // Reset backoff on successful connection
+                conn.run(tx.clone()).await;
+                tracing::warn!("Sync connection to {} ended, will reconnect", url);
+            }
+            Err(e) => {
+                tracing::error!(
+                    "Failed to connect to sync relay {}: {} (retrying in {:?})",
+                    url,
+                    e,
+                    backoff
+                );
+            }
+        }
+
+        // Wait before reconnecting
+        tokio::time::sleep(backoff).await;
+
+        // Exponential backoff
+        backoff = std::cmp::min(backoff * 2, max_backoff);
+    }
+}

src/sync/manager.rs

@@ -0,0 +1,101 @@
+//! SyncManager - Coordinates proactive sync operations
+//!
+//! The SyncManager spawns connections to configured relays, receives events,
+//! validates them through the write policy, and stores accepted events.
+
+use nostr_relay_builder::prelude::*;
+use tokio::sync::mpsc;
+
+use super::connection::{connect_with_retry, SyncedEvent};
+use super::SYNC_SOURCE_ADDR;
+use crate::nostr::builder::{Nip34WritePolicy, SharedDatabase};
+
+/// Coordinates proactive sync from configured relays
+pub struct SyncManager {
+    /// URL of the relay to sync from
+    sync_relay_url: String,
+    /// Database for storing accepted events
+    database: SharedDatabase,
+    /// Write policy for validating events
+    write_policy: Nip34WritePolicy,
+}
+
+impl SyncManager {
+    /// Create a new SyncManager
+    pub fn new(
+        sync_relay_url: String,
+        database: SharedDatabase,
+        write_policy: Nip34WritePolicy,
+    ) -> Self {
+        Self {
+            sync_relay_url,
+            database,
+            write_policy,
+        }
+    }
+
+    /// Run the sync manager
+    ///
+    /// This spawns a connection task and processes incoming events.
+    /// Runs indefinitely until the task is cancelled.
+    pub async fn run(self) {
+        tracing::info!("Starting SyncManager for relay: {}", self.sync_relay_url);
+
+        // Create channel for receiving events from connection
+        let (tx, mut rx) = mpsc::channel::<SyncedEvent>(100);
+
+        // Spawn connection task with auto-reconnect
+        let url = self.sync_relay_url.clone();
+        tokio::spawn(async move {
+            connect_with_retry(&url, tx).await;
+        });
+
+        // Process incoming events
+        while let Some(synced_event) = rx.recv().await {
+            self.process_event(synced_event).await;
+        }
+
+        tracing::warn!("SyncManager event channel closed, shutting down");
+    }
+
+    /// Process a single synced event
+    async fn process_event(&self, synced_event: SyncedEvent) {
+        let event = &synced_event.event;
+        let event_id = event.id.to_hex();
+
+        tracing::debug!(
+            "Processing synced event {} (kind {}) from {}",
+            event_id,
+            event.kind.as_u16(),
+            synced_event.source_url
+        );
+
+        // Validate through write policy using SYNC_SOURCE_ADDR
+        let result = self.write_policy.admit_event(event, &SYNC_SOURCE_ADDR).await;
+
+        match result {
+            PolicyResult::Accept => {
+                tracing::info!(
+                    "Synced event {} (kind {}) accepted, storing",
+                    event_id,
+                    event.kind.as_u16()
+                );
+
+                // Store the event in the database
+                if let Err(e) = self.database.save_event(event).await {
+                    tracing::error!("Failed to store synced event {}: {}", event_id, e);
+                } else {
+                    tracing::debug!("Synced event {} stored successfully", event_id);
+                }
+            }
+            PolicyResult::Reject(reason) => {
+                tracing::info!(
+                    "Synced event {} (kind {}) rejected: {}",
+                    event_id,
+                    event.kind.as_u16(),
+                    reason
+                );
+            }
+        }
+    }
+}

src/sync/mod.rs

@@ -0,0 +1,22 @@
+//! Proactive Sync Module for GRASP-02
+//!
+//! This module implements proactive synchronization of kind 30617 (repository state)
+//! events from configured relay(s). Events are validated through the same write policy
+//! as directly-submitted events.
+
+mod connection;
+mod manager;
+
+pub use manager::SyncManager;
+
+use std::net::SocketAddr;
+
+/// Synthetic source address used for synced events
+///
+/// This distinguishes synced events from directly-submitted events in logs and metrics.
+/// Uses 127.0.0.2:0 as a recognizable "synced event" marker.
+pub const SYNC_SOURCE_ADDR: SocketAddr =
+    SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::new(127, 0, 0, 2)), 0);
+
+/// Kind for repository state events (NIP-34)
+pub const KIND_REPOSITORY_STATE: u16 = 30617;

tests/common/relay.rs

@@ -33,11 +33,36 @@ impl TestRelay {
     /// }
     /// ```
     pub async fn start() -> Self {
-        Self::start_with_port(Self::find_free_port()).await
+        Self::start_with_options(Self::find_free_port(), None).await
     }
 
     /// Start relay on a specific port
     pub async fn start_with_port(port: u16) -> Self {
+        Self::start_with_options(port, None).await
+    }
+
+    /// Start relay with sync from another relay
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// use common::TestRelay;
+    ///
+    /// #[tokio::test]
+    /// async fn test_sync() {
+    ///     let source = TestRelay::start().await;
+    ///     let syncing = TestRelay::start_with_sync(source.url()).await;
+    ///     // ... test sync behavior ...
+    ///     syncing.stop().await;
+    ///     source.stop().await;
+    /// }
+    /// ```
+    pub async fn start_with_sync(sync_relay_url: &str) -> Self {
+        Self::start_with_options(Self::find_free_port(), Some(sync_relay_url.to_string())).await
+    }
+
+    /// Start relay with options
+    async fn start_with_options(port: u16, sync_relay_url: Option<String>) -> Self {
         let bind_address = format!("127.0.0.1:{}", port);
         let url = format!("ws://127.0.0.1:{}", port);
 
@@ -62,16 +87,21 @@ impl TestRelay {
             .expect("Failed to generate test npub");
 
         // Start the relay process
-        let process = Command::new(&binary_path)
-            .env("NGIT_BIND_ADDRESS", &bind_address)
+        let mut cmd = Command::new(&binary_path);
+        cmd.env("NGIT_BIND_ADDRESS", &bind_address)
             .env("NGIT_DOMAIN", &bind_address) // Set domain to match bind address
             .env("NGIT_GIT_DATA_PATH", git_data_dir.path())
             .env("NGIT_OWNER_NPUB", &test_npub)
             .env("RUST_LOG", "warn") // Less logging during tests
             .stdout(Stdio::null())
-            .stderr(Stdio::null())
-            .spawn()
-            .expect("Failed to start relay process");
+            .stderr(Stdio::null());
+
+        // Add sync relay URL if provided
+        if let Some(ref sync_url) = sync_relay_url {
+            cmd.env("NGIT_SYNC_RELAY_URL", sync_url);
+        }
+
+        let process = cmd.spawn().expect("Failed to start relay process");
 
         let relay = Self { process, url, port };
 

tests/proactive_sync_basic.rs

@@ -0,0 +1,262 @@
+//! GRASP-02 Phase 1: Proactive Sync Basic Integration Tests
+//!
+//! Tests the basic proactive sync functionality using two TestRelay instances:
+//! - relay_a: Source relay with events
+//! - relay_b: Sync relay configured to sync from relay_a
+//!
+//! # Running Tests
+//!
+//! ```bash
+//! cargo test --test proactive_sync_basic
+//! cargo test --test proactive_sync_basic -- --nocapture
+//! ```
+
+mod common;
+
+use std::time::Duration;
+
+use common::TestRelay;
+use nostr_sdk::prelude::*;
+
+/// Kind 30617 - Repository State (NIP-34)
+const KIND_REPOSITORY_STATE: u16 = 30617;
+
+/// Create a valid repository announcement event for testing
+///
+/// This creates a kind 30617 event with required clone and relays tags
+fn create_valid_repo_announcement(
+    keys: &Keys,
+    domain: &str,
+    identifier: &str,
+) -> Event {
+    // Build tags for repository announcement
+    let tags = vec![
+        Tag::identifier(identifier),
+        Tag::custom(
+            TagKind::custom("clone"),
+            vec![format!("http://{}/{}", domain, identifier)],
+        ),
+        Tag::custom(
+            TagKind::custom("relays"),
+            vec![format!("ws://{}", domain)],
+        ),
+    ];
+
+    EventBuilder::new(Kind::Custom(KIND_REPOSITORY_STATE), "Repository state")
+        .tags(tags)
+        .sign_with_keys(keys)
+        .expect("Failed to sign event")
+}
+
+/// Test that syncing relay connects to source relay
+#[tokio::test]
+async fn test_sync_relay_connects_to_source() {
+    // Start source relay (relay_a)
+    let relay_a = TestRelay::start().await;
+
+    // Start syncing relay (relay_b) configured to sync from relay_a
+    let relay_b = TestRelay::start_with_sync(relay_a.url()).await;
+
+    // Give some time for connection to establish
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    // If we got here without panicking, the relays started successfully
+    // The sync connection happens in the background
+
+    relay_b.stop().await;
+    relay_a.stop().await;
+}
+
+/// Test that valid events sync from source to syncing relay
+#[tokio::test]
+async fn test_valid_event_syncs_to_relay() {
+    // Start source relay (relay_a)
+    let relay_a = TestRelay::start().await;
+
+    // Give relay_a time to start
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Start syncing relay (relay_b) configured to sync from relay_a
+    let relay_b = TestRelay::start_with_sync(relay_a.url()).await;
+
+    // Create test keys
+    let keys = Keys::generate();
+
+    // Create and submit a valid repository announcement to relay_a
+    let event = create_valid_repo_announcement(&keys, &relay_a.domain(), "test-repo");
+    let event_id = event.id;
+
+    // Submit event to relay_a
+    let client_a = Client::default();
+    client_a.add_relay(relay_a.url()).await.expect("Failed to add relay_a");
+    client_a.connect().await;
+
+    let send_result = client_a.send_event(&event).await;
+    assert!(send_result.is_ok(), "Failed to send event to relay_a: {:?}", send_result.err());
+
+    // Wait for sync to occur
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    // Query relay_b to verify the event was synced
+    let client_b = Client::default();
+    client_b.add_relay(relay_b.url()).await.expect("Failed to add relay_b");
+    client_b.connect().await;
+
+    // Create filter to find our event
+    let filter = Filter::new()
+        .kind(Kind::Custom(KIND_REPOSITORY_STATE))
+        .author(keys.public_key());
+
+    let events = client_b
+        .fetch_events(filter, Duration::from_secs(5))
+        .await
+        .expect("Failed to fetch events from relay_b");
+
+    // Check if our event was synced
+    let found = events.iter().any(|e| e.id == event_id);
+
+    // Clean up
+    client_a.disconnect().await;
+    client_b.disconnect().await;
+    relay_b.stop().await;
+    relay_a.stop().await;
+
+    assert!(
+        found,
+        "Event {} was not synced to relay_b. Found {} events",
+        event_id,
+        events.len()
+    );
+}
+
+/// Test that invalid events are rejected by syncing relay validation
+#[tokio::test]
+async fn test_invalid_event_rejected_by_sync_validation() {
+    // Start source relay (relay_a) - this is a simple relay without GRASP validation
+    // For this test, we'll use a second ngit-grasp relay, but the key insight is that
+    // the syncing relay should reject events that don't pass its own validation
+
+    let relay_a = TestRelay::start().await;
+    let relay_b = TestRelay::start_with_sync(relay_a.url()).await;
+
+    // Give time for connection
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    // Create test keys
+    let keys = Keys::generate();
+
+    // Create an INVALID repository announcement (missing clone tag)
+    let tags = vec![
+        Tag::identifier("test-invalid-repo"),
+        // Missing required "clone" tag!
+        Tag::custom(
+            TagKind::custom("relays"),
+            vec![format!("ws://{}", relay_a.domain())],
+        ),
+    ];
+
+    let invalid_event = EventBuilder::new(Kind::Custom(KIND_REPOSITORY_STATE), "Invalid repo")
+        .tags(tags)
+        .sign_with_keys(&keys)
+        .expect("Failed to sign event");
+
+    let invalid_event_id = invalid_event.id;
+
+    // Submit invalid event to relay_a
+    // Note: relay_a will also reject it due to GRASP validation
+    let client_a = Client::default();
+    client_a.add_relay(relay_a.url()).await.expect("Failed to add relay_a");
+    client_a.connect().await;
+
+    // This will likely fail since relay_a also validates, but let's try
+    let _ = client_a.send_event(&invalid_event).await;
+
+    // Wait for potential sync
+    tokio::time::sleep(Duration::from_secs(1)).await;
+
+    // Query relay_b - the event should NOT be present
+    let client_b = Client::default();
+    client_b.add_relay(relay_b.url()).await.expect("Failed to add relay_b");
+    client_b.connect().await;
+
+    let filter = Filter::new()
+        .kind(Kind::Custom(KIND_REPOSITORY_STATE))
+        .author(keys.public_key());
+
+    let events = client_b
+        .fetch_events(filter, Duration::from_secs(3))
+        .await
+        .expect("Failed to fetch events from relay_b");
+
+    let found = events.iter().any(|e| e.id == invalid_event_id);
+
+    // Clean up
+    client_a.disconnect().await;
+    client_b.disconnect().await;
+    relay_b.stop().await;
+    relay_a.stop().await;
+
+    assert!(
+        !found,
+        "Invalid event {} should NOT have been synced to relay_b",
+        invalid_event_id
+    );
+}
+
+/// Test that syncing relay maintains its own validation policy
+#[tokio::test]
+async fn test_sync_respects_local_validation() {
+    // This test verifies that synced events go through the local Nip34WritePolicy
+    // by testing that orphan events (events referencing non-existent repos) are rejected
+
+    let relay_a = TestRelay::start().await;
+    let relay_b = TestRelay::start_with_sync(relay_a.url()).await;
+
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let keys = Keys::generate();
+
+    // First, create a VALID repository announcement and submit it
+    let valid_event = create_valid_repo_announcement(&keys, &relay_a.domain(), "valid-repo");
+    let valid_event_id = valid_event.id;
+
+    let client_a = Client::default();
+    client_a.add_relay(relay_a.url()).await.expect("Failed to add relay_a");
+    client_a.connect().await;
+
+    client_a
+        .send_event(&valid_event)
+        .await
+        .expect("Failed to send valid event");
+
+    // Wait for sync
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    // Query relay_b to verify the valid event was synced
+    let client_b = Client::default();
+    client_b.add_relay(relay_b.url()).await.expect("Failed to add relay_b");
+    client_b.connect().await;
+
+    let filter = Filter::new()
+        .kind(Kind::Custom(KIND_REPOSITORY_STATE))
+        .author(keys.public_key());
+
+    let events = client_b
+        .fetch_events(filter, Duration::from_secs(5))
+        .await
+        .expect("Failed to fetch events from relay_b");
+
+    let found = events.iter().any(|e| e.id == valid_event_id);
+
+    // Clean up
+    client_a.disconnect().await;
+    client_b.disconnect().await;
+    relay_b.stop().await;
+    relay_a.stop().await;
+
+    assert!(
+        found,
+        "Valid event {} should have been synced to relay_b",
+        valid_event_id
+    );
+}