path: root/src/sync/connection.rs

//! WebSocket connection handling for sync
//!
//! Manages the connection to a source relay, subscribes to events using
//! the three-layer filter strategy, and passes them through validation.
//!
//! ## Phase 2 Features
//!
//! - Three-layer filter subscriptions:
//!   1. Layer 1: kinds 30617 + 30618 (announcements)
//!   2. Layer 2: A/a tags for repository events
//!   3. Layer 3: E/e tags for related events (PRs, Issues, etc.)
//!
//! ## Phase 3 Features
//!
//! - Health tracking with success/failure reporting
//! - Exponential backoff with health-aware delays
//! - Dead relay detection and minimal retry

use std::sync::Arc;
use std::time::Duration;

use nostr_sdk::prelude::*;
use tokio::sync::mpsc;

use super::filter::FilterService;
use super::health::RelayHealthTracker;

/// 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,
    filter_service: Arc<FilterService>,
    remote_domain: String,
}

impl SyncConnection {
    /// Create a new sync connection to the given relay URL
    pub async fn new(
        url: &str,
        filter_service: Arc<FilterService>,
        remote_domain: &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,
            filter_service,
            remote_domain: remote_domain.to_string(),
        })
    }

    /// Start receiving events and send them through the channel
    ///
    /// This method runs indefinitely, handling events from all three filter layers.
    pub async fn run(self, tx: mpsc::Sender<SyncedEvent>) {
        // Subscribe to all three filter layers

        // Layer 1: Announcement discovery (kinds 30617 + 30618)
        let layer1_filters = self.filter_service.get_layer1_filters();
        for filter in &layer1_filters {
            match self.client.subscribe(filter.clone(), None).await {
                Ok(output) => {
                    tracing::info!(
                        "Subscribed to Layer 1 (announcements) on {} (subscription: {})",
                        self.url,
                        output.id()
                    );
                }
                Err(e) => {
                    tracing::error!("Failed to subscribe Layer 1 on {}: {}", self.url, e);
                }
            }
        }

        // Layer 2: Repository events (A/a tags)
        let layer2_filters = self
            .filter_service
            .get_layer2_filters(&self.remote_domain)
            .await;
        for filter in &layer2_filters {
            match self.client.subscribe(filter.clone(), None).await {
                Ok(output) => {
                    tracing::info!(
                        "Subscribed to Layer 2 (repo events) on {} (subscription: {})",
                        self.url,
                        output.id()
                    );
                }
                Err(e) => {
                    tracing::error!("Failed to subscribe Layer 2 on {}: {}", self.url, e);
                }
            }
        }

        // Layer 3: Related events (E/e tags)
        let layer3_filters = self.filter_service.get_layer3_filters().await;
        for filter in &layer3_filters {
            match self.client.subscribe(filter.clone(), None).await {
                Ok(output) => {
                    tracing::info!(
                        "Subscribed to Layer 3 (related events) on {} (subscription: {})",
                        self.url,
                        output.id()
                    );
                }
                Err(e) => {
                    tracing::error!("Failed to subscribe Layer 3 on {}: {}", self.url, e);
                }
            }
        }

        tracing::info!(
            "Sync subscriptions active on {} (L1: {}, L2: {}, L3: {})",
            self.url,
            layer1_filters.len(),
            layer2_filters.len(),
            layer3_filters.len()
        );

        // 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 health-aware exponential backoff
///
/// This function manages the connection lifecycle with health tracking:
/// - Checks health state before attempting connections
/// - Reports success/failure to the health tracker
/// - Respects backoff delays from the health tracker
/// - Handles dead relay detection (24h+ failures)
///
/// # Arguments
/// * `url` - The relay URL to connect to
/// * `tx` - Channel sender for synced events
/// * `filter_service` - FilterService for building subscriptions
/// * `our_domain` - Our relay's domain (used to extract remote domain)
/// * `health_tracker` - Health tracker for managing connection state
pub async fn connect_with_retry(
    url: &str,
    tx: mpsc::Sender<SyncedEvent>,
    filter_service: Arc<FilterService>,
    _our_domain: &str,
    health_tracker: Arc<RelayHealthTracker>,
) {
    // Extract remote domain from URL
    let remote_domain = extract_domain_from_url(url).unwrap_or_else(|| url.to_string());

    loop {
        // Check if we should attempt connection based on health state
        if !health_tracker.should_attempt_connection(url) {
            // Wait for remaining backoff
            if let Some(remaining) = health_tracker.get_remaining_backoff(url) {
                tracing::debug!(
                    "Relay {} in backoff, waiting {:?} before retry",
                    url,
                    remaining
                );
                tokio::time::sleep(remaining).await;
                continue;
            }
        }

        // Log current health state for dead relays
        if health_tracker.is_dead(url) {
            tracing::info!(
                "Attempting reconnection to dead relay {} (daily retry)",
                url
            );
        }

        match SyncConnection::new(url, filter_service.clone(), &remote_domain).await {
            Ok(conn) => {
                // Record successful connection
                health_tracker.record_success(url);
                tracing::info!("Sync connection established to {}", url);

                // Run the connection (this blocks until disconnection)
                conn.run(tx.clone()).await;

                // Connection ended - record as failure for reconnection backoff
                // (The connection ending is considered a failure even if it worked for a while)
                health_tracker.record_failure(url);
                tracing::warn!("Sync connection to {} ended, will reconnect", url);
            }
            Err(e) => {
                // Record connection failure
                health_tracker.record_failure(url);

                let failure_count = health_tracker.get_failure_count(url);
                let state = health_tracker.get_state(url);

                tracing::error!(
                    "Failed to connect to sync relay {} (attempt #{}, state: {}): {}",
                    url,
                    failure_count,
                    state,
                    e
                );
            }
        }

        // Get the backoff duration from health tracker
        // If the health tracker has no backoff set (shouldn't happen), use a small default
        let wait_duration = health_tracker
            .get_remaining_backoff(url)
            .unwrap_or(Duration::from_secs(5));

        tracing::debug!(
            "Waiting {:?} before reconnecting to {}",
            wait_duration,
            url
        );
        tokio::time::sleep(wait_duration).await;
    }
}

/// Extract domain from a URL
fn extract_domain_from_url(url: &str) -> Option<String> {
    let url = url
        .trim_start_matches("ws://")
        .trim_start_matches("wss://")
        .trim_start_matches("http://")
        .trim_start_matches("https://");

    // Remove path
    let domain = url.split('/').next()?;

    Some(domain.to_string())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_extract_domain() {
        assert_eq!(
            extract_domain_from_url("ws://127.0.0.1:8080"),
            Some("127.0.0.1:8080".to_string())
        );
        assert_eq!(
            extract_domain_from_url("wss://relay.example.com/path"),
            Some("relay.example.com".to_string())
        );
    }
}