eidetica/sync/

peer.rs

//! Peer management, sync relationships, and address handling for the sync system.

use std::time::{Duration, SystemTime, UNIX_EPOCH};

use tokio::sync::oneshot;
use tracing::info;

use super::{
    Address, ConnectionState, PeerId, PeerInfo, PeerStatus, Sync, SyncError, SyncHandle,
    SyncPeerInfo, SyncStatus, background::SyncCommand, peer_manager::PeerManager,
};
use crate::{Result, auth::crypto::PublicKey, entry::ID};

impl Sync {
    // === Peer Management Methods ===

    /// Register a new remote peer in the sync network.
    ///
    /// # Arguments
    /// * `pubkey` - The peer's public key
    /// * `display_name` - Optional human-readable name for the peer
    ///
    /// # Returns
    /// A Result indicating success or an error.
    pub async fn register_peer(
        &self,
        pubkey: &PublicKey,
        display_name: Option<&str>,
    ) -> Result<()> {
        // Store in sync tree via PeerManager
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn)
            .register_peer(pubkey, display_name)
            .await?;
        txn.commit().await?;

        // Background sync will read peer info directly from sync tree when needed
        Ok(())
    }

    /// Update the status of a registered peer.
    ///
    /// # Arguments
    /// * `pubkey` - The peer's public key
    /// * `status` - The new status for the peer
    ///
    /// # Returns
    /// A Result indicating success or an error.
    pub async fn update_peer_status(&self, pubkey: &PublicKey, status: PeerStatus) -> Result<()> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn)
            .update_peer_status(pubkey, status)
            .await?;
        txn.commit().await?;
        Ok(())
    }

    /// Get information about a registered peer.
    ///
    /// # Arguments
    /// * `pubkey` - The peer's public key
    ///
    /// # Returns
    /// The peer information if found, None otherwise.
    pub async fn get_peer_info(&self, pubkey: &PublicKey) -> Result<Option<PeerInfo>> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn).get_peer_info(pubkey).await
        // No commit - just reading
    }

    /// List all registered peers.
    ///
    /// # Returns
    /// A vector of all registered peer information.
    pub async fn list_peers(&self) -> Result<Vec<PeerInfo>> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn).list_peers().await
        // No commit - just reading
    }

    /// Remove a peer from the sync network.
    ///
    /// This removes the peer entry and all associated sync relationships and transport info.
    ///
    /// # Arguments
    /// * `pubkey` - The peer's public key
    ///
    /// # Returns
    /// A Result indicating success or an error.
    pub async fn remove_peer(&self, pubkey: &PublicKey) -> Result<()> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn).remove_peer(pubkey).await?;
        txn.commit().await?;
        Ok(())
    }

    // === Declarative Sync API ===

    /// Register a peer for syncing (declarative API).
    ///
    /// This is the recommended way to set up syncing. It immediately registers
    /// the peer and tree/peer relationship, then the background sync engine
    /// handles the actual data synchronization.
    ///
    /// # Arguments
    /// * `info` - Information about the peer and sync configuration
    ///
    /// # Returns
    /// A handle for tracking sync status and adding more address hints.
    ///
    /// # Example
    /// ```no_run
    /// # use eidetica::*;
    /// # use eidetica::sync::{SyncPeerInfo, Address, AuthParams};
    /// # use eidetica::auth::PublicKey;
    /// # async fn example(sync: sync::Sync, peer_pubkey: PublicKey, tree_id: entry::ID) -> Result<()> {
    /// // Register peer for syncing
    /// let handle = sync.register_sync_peer(SyncPeerInfo {
    ///     peer_pubkey,
    ///     tree_id,
    ///     addresses: vec![Address {
    ///         transport_type: "http".to_string(),
    ///         address: "http://localhost:8080".to_string(),
    ///     }],
    ///     auth: None,
    ///     display_name: Some("My Peer".to_string()),
    /// }).await?;
    ///
    /// // Optionally wait for initial sync
    /// handle.wait_for_initial_sync().await?;
    ///
    /// // Check status anytime
    /// let status = handle.status().await?;
    /// println!("Has local data: {}", status.has_local_data);
    /// # Ok(())
    /// # }
    /// ```
    pub async fn register_sync_peer(&self, info: SyncPeerInfo) -> Result<SyncHandle> {
        let txn = self.sync_tree.new_transaction().await?;
        let peer_mgr = PeerManager::new(&txn);

        // Register peer if it doesn't exist
        if peer_mgr.get_peer_info(&info.peer_pubkey).await?.is_none() {
            peer_mgr
                .register_peer(&info.peer_pubkey, info.display_name.as_deref())
                .await?;
        }

        // Add all address hints
        for addr in &info.addresses {
            peer_mgr
                .add_address(&info.peer_pubkey, addr.clone())
                .await?;
        }

        // Register the tree/peer relationship
        peer_mgr
            .add_tree_sync(&info.peer_pubkey, &info.tree_id)
            .await?;

        // TODO: Store auth params if provided for bootstrap
        // For now, auth is passed during the actual sync handshake via on_local_write callback

        txn.commit().await?;

        info!(
            peer = %info.peer_pubkey,
            tree = %info.tree_id,
            address_count = info.addresses.len(),
            "Registered peer for syncing"
        );

        Ok(SyncHandle {
            tree_id: info.tree_id,
            peer_pubkey: info.peer_pubkey,
            sync: self.clone(),
        })
    }

    /// Get the current sync status for a tree/peer pair.
    ///
    /// # Arguments
    /// * `tree_id` - The tree to check
    /// * `peer_pubkey` - The peer public key
    ///
    /// # Returns
    /// Current sync status including whether we have local data.
    pub async fn get_sync_status(
        &self,
        tree_id: &ID,
        peer_pubkey: &PublicKey,
    ) -> Result<SyncStatus> {
        // Check if we have local data for this tree
        let backend = self.backend()?;
        let our_snapshot = backend.snapshot(tree_id).await.unwrap_or_default();

        // TODO: Track last_error in sync tree
        Ok(SyncStatus {
            has_local_data: !our_snapshot.is_empty(),
            last_sync: self.peer_last_sync(peer_pubkey),
            last_error: None,
        })
    }

    /// When the background engine last synced with a peer.
    ///
    /// Reads the shared liveness state directly rather than going through the
    /// command channel, so it does not wait on an in-flight sync round. `None`
    /// when no successful round is on record, including when the engine has not
    /// started — both mean the same thing to a caller.
    fn peer_last_sync(&self, peer_pubkey: &PublicKey) -> Option<SystemTime> {
        let millis = self
            .peer_state
            .last_success_ms(&PeerId::from(peer_pubkey))?;
        Some(UNIX_EPOCH + Duration::from_millis(millis))
    }

    // === Database Sync Relationship Methods ===

    /// Add a tree to the sync relationship with a peer.
    ///
    /// # Arguments
    /// * `peer_pubkey` - The peer's public key
    /// * `tree_root_id` - The root ID of the tree to sync
    ///
    /// # Returns
    /// A Result indicating success or an error.
    pub async fn add_tree_sync(&self, peer_pubkey: &PublicKey, tree_root_id: &ID) -> Result<()> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn)
            .add_tree_sync(peer_pubkey, tree_root_id)
            .await?;
        txn.commit().await?;
        Ok(())
    }

    /// Remove a tree from the sync relationship with a peer.
    ///
    /// # Arguments
    /// * `peer_pubkey` - The peer's public key
    /// * `tree_root_id` - The root ID of the tree to stop syncing
    ///
    /// # Returns
    /// A Result indicating success or an error.
    pub async fn remove_tree_sync(&self, peer_pubkey: &PublicKey, tree_root_id: &ID) -> Result<()> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn)
            .remove_tree_sync(peer_pubkey, tree_root_id)
            .await?;
        txn.commit().await?;
        Ok(())
    }

    /// Get the list of trees synced with a peer.
    ///
    /// # Arguments
    /// * `peer_pubkey` - The peer's public key
    ///
    /// # Returns
    /// A vector of tree root IDs synced with this peer.
    pub async fn get_peer_trees(&self, peer_pubkey: &PublicKey) -> Result<Vec<ID>> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn).get_peer_trees(peer_pubkey).await
        // No commit - just reading
    }

    /// Get all peers that sync a specific tree.
    ///
    /// # Arguments
    /// * `tree_root_id` - The root ID of the tree
    ///
    /// # Returns
    /// A vector of peer IDs that sync this tree.
    pub async fn get_tree_peers(&self, tree_root_id: &ID) -> Result<Vec<PeerId>> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn).get_tree_peers(tree_root_id).await
        // No commit - just reading
    }

    /// Connect to a remote peer and perform handshake.
    ///
    /// This method initiates a connection to a peer, performs the handshake protocol,
    /// and automatically registers the peer if successful.
    ///
    /// # Arguments
    /// * `address` - The address of the peer to connect to
    ///
    /// # Returns
    /// A Result containing the peer's public key if successful.
    pub async fn connect_to_peer(&self, address: &Address) -> Result<PublicKey> {
        let (tx, rx) = oneshot::channel();

        self.background_tx
            .get()
            .ok_or(SyncError::NoTransportEnabled)?
            .send(SyncCommand::ConnectToPeer {
                address: address.clone(),
                response: tx,
            })
            .await
            .map_err(|e| SyncError::CommandSendError(e.to_string()))?;

        rx.await
            .map_err(|e| SyncError::Network(format!("Response channel error: {e}")))?
    }

    /// Update the connection state of a peer.
    ///
    /// # Arguments
    /// * `pubkey` - The peer's public key
    /// * `state` - The new connection state
    ///
    /// # Returns
    /// A Result indicating success or an error.
    pub async fn update_peer_connection_state(
        &self,
        pubkey: &PublicKey,
        state: ConnectionState,
    ) -> Result<()> {
        let txn = self.sync_tree.new_transaction().await?;
        let peer_manager = PeerManager::new(&txn);

        // Get current peer info
        let mut peer_info = match peer_manager.get_peer_info(pubkey).await? {
            Some(info) => info,
            None => return Err(SyncError::PeerNotFound(pubkey.to_string()).into()),
        };

        // Update connection state
        peer_info.connection_state = state;
        peer_info.touch_at(txn.now_rfc3339()?);

        // Save updated peer info
        peer_manager.update_peer_info(pubkey, peer_info).await?;
        txn.commit().await?;
        Ok(())
    }

    /// Check if a tree is synced with a specific peer.
    ///
    /// # Arguments
    /// * `peer_pubkey` - The peer's public key
    /// * `tree_root_id` - The root ID of the tree
    ///
    /// # Returns
    /// True if the tree is synced with the peer, false otherwise.
    pub async fn is_tree_synced_with_peer(
        &self,
        peer_pubkey: &PublicKey,
        tree_root_id: &ID,
    ) -> Result<bool> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn)
            .is_tree_synced_with_peer(peer_pubkey, tree_root_id)
            .await
        // No commit - just reading
    }

    // === Address Management Methods ===

    /// Add an address to a peer.
    ///
    /// # Arguments
    /// * `peer_pubkey` - The peer's public key
    /// * `address` - The address to add
    ///
    /// # Returns
    /// A Result indicating success or an error.
    pub async fn add_peer_address(&self, peer_pubkey: &PublicKey, address: Address) -> Result<()> {
        // Update sync tree via PeerManager
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn)
            .add_address(peer_pubkey, address)
            .await?;
        txn.commit().await?;

        // Background sync will read updated peer info directly from sync tree when needed
        Ok(())
    }

    /// Remove a specific address from a peer.
    ///
    /// # Arguments
    /// * `peer_pubkey` - The peer's public key
    /// * `address` - The address to remove
    ///
    /// # Returns
    /// A Result indicating success or an error (true if removed, false if not found).
    pub async fn remove_peer_address(
        &self,
        peer_pubkey: &PublicKey,
        address: &Address,
    ) -> Result<bool> {
        let txn = self.sync_tree.new_transaction().await?;
        let result = PeerManager::new(&txn)
            .remove_address(peer_pubkey, address)
            .await?;
        txn.commit().await?;
        Ok(result)
    }

    /// Get addresses for a peer, optionally filtered by transport type.
    ///
    /// # Arguments
    /// * `peer_pubkey` - The peer's public key
    /// * `transport_type` - Optional transport type filter
    ///
    /// # Returns
    /// A vector of addresses matching the criteria.
    pub async fn get_peer_addresses(
        &self,
        peer_pubkey: &PublicKey,
        transport_type: Option<&str>,
    ) -> Result<Vec<Address>> {
        let txn = self.sync_tree.new_transaction().await?;
        PeerManager::new(&txn)
            .get_addresses(peer_pubkey, transport_type)
            .await
        // No commit - just reading
    }
}