syncweb-core 0.1.1-alpha.11

use std::{
    collections::{HashMap, HashSet},
    path::Path,
    pin::Pin,
    sync::Arc,
    time::{Duration, Instant},
};

use iroh_blobs::Hash;
use iroh_docs::{NamespaceId, engine::LiveEvent};
use n0_future::StreamExt;
use tokio::sync::mpsc;

use crate::{
    error::Result,
    filter::{FilterAction, FilterEngine, FilterEntry},
    folder::FolderManager,
    node::discovery::TopicTracker,
    node::iroh_node::IrohNode,
    node::{blob_store::BlobStore, docs_engine::DocsEngine},
    storage::node_db::NodeDatabase,
};

use super::checkpoint::SyncCheckpoint;
use super::{FetchStrategy, IntentHandle, SessionMode, SubscribeParams, SyncCommand, SyncEvent};

type IntentSetup = (
    IntentState,
    Pin<Box<tokio::time::Sleep>>,
    Pin<Box<tokio::time::Sleep>>,
    Option<SyncCheckpoint>,
    HashSet<Vec<u8>>,
);

/// Current aggregate statistics for a synchronization intent.
#[derive(Clone, Debug, Default, Eq, PartialEq)]
#[non_exhaustive]
pub struct TransferStats {
    pub bytes_transferred: u64,
    pub bytes_total: Option<u64>,
    pub bytes_per_second: u64,
    pub peer_count: usize,
    pub eta: Option<Duration>,
}

impl TransferStats {
    #[must_use]
    pub fn from_progress(
        bytes_transferred: u64,
        bytes_total: Option<u64>,
        elapsed: Duration,
        peer_count: usize,
    ) -> Self {
        let elapsed_millis = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX);
        let bytes_per_second = bytes_transferred
            .saturating_mul(1_000)
            .checked_div(elapsed_millis)
            .unwrap_or(0);
        let eta = bytes_total.and_then(|total| {
            let remaining = total.saturating_sub(bytes_transferred);
            remaining.checked_div(bytes_per_second).map(Duration::from_secs)
        });
        Self {
            bytes_transferred,
            bytes_total,
            bytes_per_second,
            peer_count,
            eta,
        }
    }
}

/// Coordinates document reconciliation and blob availability for a folder.
#[derive(Clone)]
pub struct SyncEngine {
    folder_manager: FolderManager,
    blob_store: BlobStore,
    docs_engine: DocsEngine,
    node_db: Option<NodeDatabase>,
    topic_tracker: Option<TopicTracker>,
    /// Optional reconciler invoked after live synchronization ingests remote
    /// entries. The daemon uses it to materialize writable folders.
    reconciler: Option<Arc<dyn Fn(NamespaceId) + Send + Sync>>,
}

impl SyncEngine {
    #[must_use]
    pub const fn new(
        folder_manager: FolderManager,
        blob_store: BlobStore,
        docs_engine: DocsEngine,
        topic_tracker: Option<TopicTracker>,
    ) -> Self {
        Self {
            folder_manager,
            blob_store,
            docs_engine,
            node_db: None,
            topic_tracker,
            reconciler: None,
        }
    }

    /// Set the node database for checkpoint tracking.
    #[must_use]
    pub fn with_node_db(mut self, node_db: NodeDatabase) -> Self {
        self.node_db = Some(node_db);
        self
    }

    /// Attach a reconciliation hook called after remote entries are ingested.
    #[must_use]
    pub fn with_reconciler(mut self, reconciler: Arc<dyn Fn(NamespaceId) + Send + Sync>) -> Self {
        self.reconciler = Some(reconciler);
        self
    }

    #[must_use]
    pub fn from_node(node: &IrohNode, folder_manager: FolderManager) -> Self {
        Self::new(
            folder_manager,
            node.blob_store().clone(),
            node.docs_engine().clone(),
            Some(node.topic_tracker().clone()),
        )
    }

    /// Start reconciling a managed folder.
    ///
    /// Iroh Docs performs network reconciliation itself. This intent provides
    /// lifecycle, control, and aggregate transfer events around that process.
    ///
    /// # Errors
    ///
    /// Returns an error if the folder is not managed locally.
    pub async fn sync(&self, folder_id: NamespaceId, mode: SessionMode) -> Result<IntentHandle> {
        self.sync_with_params(folder_id, mode, SubscribeParams::default()).await
    }

    /// Start a synchronization intent with subscription filtering.
    ///
    /// # Errors
    ///
    /// Returns an error if the folder is not managed locally or live sync
    /// cannot be started.
    pub async fn sync_with_params(
        &self,
        folder_id: NamespaceId,
        mode: SessionMode,
        params: SubscribeParams,
    ) -> Result<IntentHandle> {
        self.start(folder_id, mode, params, None).await
    }

    /// Subscribe to live changes in a folder.
    ///
    /// # Errors
    ///
    /// Returns an error if the folder is not managed locally or live sync
    /// cannot be started.
    pub async fn subscribe(&self, folder_id: NamespaceId, params: SubscribeParams) -> Result<IntentHandle> {
        self.sync_with_params(folder_id, SessionMode::Continuous, params).await
    }

    /// Fetch folder content using a full or filtered reconciliation strategy.
    ///
    /// Peer-count filters are applied by callers that have availability
    /// observations. The document subscription still carries path, size, and
    /// count limits so the remote reconciliation remains bounded.
    ///
    /// # Errors
    ///
    /// Returns an error if the folder is not managed locally or synchronization
    /// cannot be started.
    pub async fn fetch(&self, folder_id: NamespaceId, strategy: FetchStrategy) -> Result<IntentHandle> {
        let params = Self::fetch_params(strategy);
        self.sync_with_params(folder_id, SessionMode::ReconcileOnce, params)
            .await
    }

    /// Start a synchronization intent and filter emitted entries.
    ///
    /// The document reconciliation remains content-addressed and is handled
    /// by iroh-docs; the filter controls which entries are reported to the
    /// automatic synchronization consumer.
    ///
    /// # Errors
    ///
    /// Returns an error if the folder is not managed locally or live sync
    /// cannot be started.
    pub async fn sync_with_filter(
        &self,
        folder_id: NamespaceId,
        mode: SessionMode,
        params: SubscribeParams,
        filter: FilterEngine,
    ) -> Result<IntentHandle> {
        self.start(folder_id, mode, params, Some(filter)).await
    }

    fn fetch_params(strategy: FetchStrategy) -> SubscribeParams {
        let FetchStrategy::Filter(filter) = strategy else {
            return SubscribeParams::default();
        };
        let mut params = SubscribeParams::default();
        if let Some(paths) = filter.paths
            && let Some(path) = paths.first()
        {
            params = params.with_area(super::AreaFilter::Prefix(path.clone()));
        }
        if filter.max_size.is_some() || filter.max_count.is_some() {
            let area = params.area_filter.clone().unwrap_or(super::AreaFilter::All);
            let limits = super::AreaOfInterest::with_limits(
                area,
                filter.max_size.unwrap_or(0),
                u64::try_from(filter.max_count.unwrap_or(0)).unwrap_or(u64::MAX),
            );
            params = params.with_limits(limits);
        }
        params
    }

    async fn start(
        &self,
        folder_id: NamespaceId,
        mode: SessionMode,
        params: SubscribeParams,
        filter: Option<FilterEngine>,
    ) -> Result<IntentHandle> {
        let folder = self.folder_manager.get(folder_id).await?;
        // Feed peers discovered over the topic tracker straight into the
        // document sync below. Without this the tracker only logged the peers
        // it found and the intent synced with nobody: the first local change
        // after the initial join then never propagated, and a creator could
        // not push to a joiner it had learned about out of band.
        let mut discovered = Vec::new();
        if let Some(topic_tracker) = &self.topic_tracker {
            match topic_tracker.find_peers(folder_id).await {
                Ok(peers) if !peers.is_empty() => {
                    tracing::debug!(%folder_id, peer_count = peers.len(), "discovered peers via topic tracker");
                    discovered = peers;
                }
                Ok(_) => {}
                Err(error) => {
                    tracing::debug!(%folder_id, %error, "failed to discover peers via topic tracker");
                }
            }
        }
        let live_events = self.docs_engine.watch(folder.doc()).await?;
        self.docs_engine.start_sync(folder.doc(), discovered).await?;
        let (events, commands, handle) = IntentHandle::channel();
        let node_db = self.node_db.clone();
        let reconciler = self.reconciler.clone();
        tokio::spawn(run_intent(
            folder,
            IntentConfig {
                mode,
                params,
                filter,
                reconciler,
            },
            self.blob_store.clone(),
            live_events,
            events,
            commands,
            node_db,
        ));
        Ok(handle)
    }
}

struct IntentConfig {
    mode: SessionMode,
    params: SubscribeParams,
    filter: Option<FilterEngine>,
    reconciler: Option<Arc<dyn Fn(NamespaceId) + Send + Sync>>,
}

async fn run_intent(
    folder: crate::folder::SyncwebFolder,
    config: IntentConfig,
    blob_store: BlobStore,
    mut live_events: impl n0_future::Stream<Item = Result<LiveEvent>> + Send + Unpin + 'static,
    events: mpsc::UnboundedSender<SyncEvent>,
    mut commands: mpsc::UnboundedReceiver<SyncCommand>,
    node_db: Option<NodeDatabase>,
) {
    const SETTLE: Duration = Duration::from_secs(2);
    const FORCED_QUIESCE: Duration = Duration::from_secs(10);

    let started = Instant::now();
    if !send_initial_events(&events, started) {
        return;
    }

    let (mut state, mut quiesce, mut forced_quiesce, checkpoint, completed_keys) =
        setup_intent_checkpoint(&folder, node_db.as_ref());

    loop {
        tokio::select! {
            received_command = commands.recv() => {
                let Some(command) = received_command else {
                    let _result = folder.doc().leave().await;
                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                    return;
                };
                if !handle_command(&folder, command, &mut state.paused, &events).await {
                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                    return;
                }
            }
            () = &mut quiesce, if !config.mode.is_continuous() && state.sync_finished && state.sizes.is_empty() => {
                finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                return;
            }
            () = &mut forced_quiesce, if !config.mode.is_continuous() && state.sync_finished => {
                let pending = state.sizes.len();
                if pending > 0 {
                    tracing::warn!(%pending, "forced quiesce: content for some blobs never arrived");
                }
                finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                return;
            }
            next_event = live_events.next(), if !state.paused => {
                let processor = IntentEventProcessor {
                    folder: &folder,
                    config: &config,
                    blob_store: &blob_store,
                    events: &events,
                    checkpoint: checkpoint.as_ref(),
                    state: &mut state,
                    completed_keys: &completed_keys,
                    started,
                };
                match processor.process(next_event).await {
                    IntentEventOutcome::Finished => return,
                    IntentEventOutcome::Skipped => continue,
                    IntentEventOutcome::Applied => {}
                }
                if !config.mode.is_continuous() && state.sync_finished {
                    let now = tokio::time::Instant::now();
                    quiesce.as_mut().reset(now.checked_add(SETTLE).unwrap_or(now));
                    forced_quiesce.as_mut().reset(now.checked_add(FORCED_QUIESCE).unwrap_or(now));
                }
            }
        }
    }
}

enum IntentEventOutcome {
    Finished,
    Skipped,
    Applied,
}

struct IntentEventProcessor<'a> {
    folder: &'a crate::folder::SyncwebFolder,
    config: &'a IntentConfig,
    blob_store: &'a BlobStore,
    events: &'a mpsc::UnboundedSender<SyncEvent>,
    checkpoint: Option<&'a SyncCheckpoint>,
    state: &'a mut IntentState,
    completed_keys: &'a HashSet<Vec<u8>>,
    started: Instant,
}

impl IntentEventProcessor<'_> {
    async fn process(self, next_event: Option<Result<LiveEvent>>) -> IntentEventOutcome {
        let Some(event_result) = next_event else {
            let _result = self.events.send(SyncEvent::Finished);
            finalize_checkpoint(self.checkpoint, self.state, self.events);
            return IntentEventOutcome::Finished;
        };
        let live_event = match event_result {
            Ok(event) => event,
            Err(error) => {
                let _result = self.events.send(SyncEvent::Failed(error.to_string()));
                finalize_checkpoint(self.checkpoint, self.state, self.events);
                return IntentEventOutcome::Finished;
            }
        };
        let event_name = match &live_event {
            LiveEvent::InsertLocal { .. } => "InsertLocal",
            LiveEvent::InsertRemote { .. } => "InsertRemote",
            LiveEvent::ContentReady { .. } => "ContentReady",
            LiveEvent::PendingContentReady => "PendingContentReady",
            LiveEvent::NeighborUp(_) => "NeighborUp",
            LiveEvent::NeighborDown(_) => "NeighborDown",
            LiveEvent::SyncFinished(_) => "SyncFinished",
        };
        tracing::info!(
            namespace_id = %self.folder.namespace_id(),
            event = event_name,
            included = include_event(&self.config.params, &live_event),
            "engine live event"
        );
        if !include_event(&self.config.params, &live_event) {
            return IntentEventOutcome::Skipped;
        }
        if event_key(&live_event).is_some_and(|key| self.completed_keys.contains(&key)) {
            return IntentEventOutcome::Skipped;
        }
        if let LiveEvent::ContentReady { hash } = &live_event {
            tracing::info!(%hash, "engine saw ContentReady");
            match self.blob_store.has(*hash).await {
                Ok(true) => {}
                Ok(false) => {
                    let message = format!("content-ready blob {hash} is missing locally");
                    let _result = self.events.send(SyncEvent::Failed(message));
                    finalize_checkpoint(self.checkpoint, self.state, self.events);
                    return IntentEventOutcome::Finished;
                }
                Err(error) => {
                    let _result = self.events.send(SyncEvent::Failed(error.to_string()));
                    finalize_checkpoint(self.checkpoint, self.state, self.events);
                    return IntentEventOutcome::Finished;
                }
            }
            if let Some(cp) = self.checkpoint {
                let _ = cp.mark_completed(hash.as_bytes(), *hash, self.state.sizes.get(hash).copied().unwrap_or(0));
            }
            reconcile_after_sync(self.folder.namespace_id(), self.config);
        }
        if matches!(live_event, LiveEvent::InsertRemote { .. } | LiveEvent::SyncFinished(_)) {
            reconcile_after_sync(self.folder.namespace_id(), self.config);
        }
        self.state
            .apply(live_event, &self.config.params, self.config.filter.as_ref());
        if !send_progress(self.events, self.state, self.started.elapsed()) {
            finalize_checkpoint(self.checkpoint, self.state, self.events);
            return IntentEventOutcome::Finished;
        }
        IntentEventOutcome::Applied
    }
}

fn event_key(event: &LiveEvent) -> Option<Vec<u8>> {
    match event {
        LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => Some(entry.key().to_vec()),
        LiveEvent::ContentReady { .. }
        | LiveEvent::PendingContentReady
        | LiveEvent::NeighborUp(_)
        | LiveEvent::NeighborDown(_)
        | LiveEvent::SyncFinished(_) => None,
    }
}

fn setup_intent_checkpoint(folder: &crate::folder::SyncwebFolder, node_db: Option<&NodeDatabase>) -> IntentSetup {
    let namespace_id = folder.namespace_id();
    let checkpoint = node_db.and_then(|db| match SyncCheckpoint::resume(db, namespace_id) {
        Ok(Some(cp)) => {
            tracing::info!(%namespace_id, session = %cp.session_id(), "resuming sync checkpoint");
            Some(cp)
        }
        Ok(None) => SyncCheckpoint::create_session(db, namespace_id)
            .map_err(|e| tracing::warn!(%namespace_id, %e, "failed to create checkpoint"))
            .ok(),
        Err(err) => {
            tracing::warn!(%namespace_id, %err, "failed to resume checkpoint");
            SyncCheckpoint::create_session(db, namespace_id)
                .map_err(|e2| tracing::warn!(%namespace_id, %e2, "failed to create checkpoint"))
                .ok()
        }
    });
    let completed_keys: HashSet<Vec<u8>> = checkpoint
        .as_ref()
        .and_then(|cp| cp.completed_keys().ok())
        .unwrap_or_default();

    let state = IntentState {
        completed: u64::try_from(completed_keys.len()).unwrap_or(u64::MAX),
        ..IntentState::default()
    };
    let quiesce = Box::pin(tokio::time::sleep(Duration::MAX));
    let forced_quiesce = Box::pin(tokio::time::sleep(Duration::MAX));
    (state, quiesce, forced_quiesce, checkpoint, completed_keys)
}

fn finalize_checkpoint(
    checkpoint: Option<&SyncCheckpoint>,
    state: &IntentState,
    events: &mpsc::UnboundedSender<SyncEvent>,
) {
    if let Some(cp) = checkpoint {
        if state.completed > 0 {
            let _ = cp.complete();
        } else {
            let _ = cp.leave_unfinished();
        }
    }
    let _ = events.send(SyncEvent::Finished);
}

fn send_initial_events(events: &mpsc::UnboundedSender<SyncEvent>, started: Instant) -> bool {
    events.send(SyncEvent::Started).is_ok()
        && events
            .send(SyncEvent::Progress {
                completed: 0,
                total: Some(0),
            })
            .is_ok()
        && events
            .send(SyncEvent::Stats(TransferStats::from_progress(
                0,
                Some(0),
                started.elapsed(),
                0,
            )))
            .is_ok()
}

fn send_progress(events: &mpsc::UnboundedSender<SyncEvent>, state: &IntentState, elapsed: Duration) -> bool {
    events
        .send(SyncEvent::Progress {
            completed: state.completed,
            total: None,
        })
        .is_ok()
        && events
            .send(SyncEvent::Stats(TransferStats::from_progress(
                state.transferred,
                None,
                elapsed,
                state.peer_count,
            )))
            .is_ok()
}

#[derive(Default)]
struct IntentState {
    paused: bool,
    completed: u64,
    transferred: u64,
    peer_count: usize,
    sizes: HashMap<Hash, u64>,
    area_count: u64,
    area_size: u64,
    sync_finished: bool,
}

impl IntentState {
    fn apply(&mut self, event: LiveEvent, params: &SubscribeParams, filter: Option<&FilterEngine>) {
        match event {
            LiveEvent::NeighborUp(_) => self.peer_count = self.peer_count.saturating_add(1),
            LiveEvent::NeighborDown(_) => self.peer_count = self.peer_count.saturating_sub(1),
            LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
                self.apply_insert(&entry, params, filter);
            }
            LiveEvent::ContentReady { hash } => {
                self.transferred = self.transferred.saturating_add(self.sizes.remove(&hash).unwrap_or(0));
            }
            LiveEvent::PendingContentReady => {}
            LiveEvent::SyncFinished(sync_event) => {
                self.sync_finished = true;
                match sync_event.result {
                    Ok(details)
                        if params.area_filter.is_none() && params.area_of_interest.is_none() && filter.is_none() =>
                    {
                        self.completed = self
                            .completed
                            .saturating_add(u64::try_from(details.entries_received).unwrap_or(u64::MAX));
                    }
                    Ok(_details) => {}
                    Err(error) => {
                        // A failed sync *attempt* is a normal, non-fatal event
                        // in a peer-to-peer swarm: the remote peer may be
                        // offline, busy, or already syncing. Tearing down the
                        // continuous intent here would drop the live sync
                        // connection and restart it with backoff, which both
                        // stops local changes from being pushed and makes the
                        // intent replay every entry on each restart. Record it
                        // and keep syncing.
                        tracing::debug!(error, "sync attempt failed; keeping the live intent alive");
                    }
                }
            }
        }
    }

    fn apply_insert(&mut self, entry: &iroh_docs::Entry, params: &SubscribeParams, filter: Option<&FilterEngine>) {
        let path = entry_path(entry);
        let hash = entry.content_hash();
        let accepted = params.accepts(&path, &hash)
            && filter.is_none_or(|engine| {
                engine.evaluate(&FilterEntry::new(path.clone(), entry.content_len())) != FilterAction::Reject
            });
        if !accepted {
            return;
        }
        let size = entry.content_len();
        if params
            .area_of_interest
            .as_ref()
            .is_some_and(|area| !area.permits(self.area_count, self.area_size, size))
        {
            return;
        }
        if params.area_of_interest.is_some() {
            self.area_count = self.area_count.saturating_add(1);
            self.area_size = self.area_size.saturating_add(size);
        }
        self.completed = self.completed.saturating_add(1);
        self.sizes.insert(hash, size);
    }
}

async fn handle_command(
    folder: &crate::folder::SyncwebFolder,
    command: SyncCommand,
    paused: &mut bool,
    events: &mpsc::UnboundedSender<SyncEvent>,
) -> bool {
    match command {
        SyncCommand::Pause => {
            if !*paused {
                if let Err(error) = folder.doc().leave().await {
                    let _result = events.send(SyncEvent::Failed(error.to_string()));
                    return false;
                }
                *paused = true;
            }
            events.send(SyncEvent::Paused).is_ok()
        }
        SyncCommand::Resume => {
            if *paused {
                if let Err(error) = folder.doc().start_sync(Vec::new()).await {
                    let _result = events.send(SyncEvent::Failed(error.to_string()));
                    return false;
                }
                *paused = false;
            }
            events.send(SyncEvent::Resumed).is_ok()
        }
        SyncCommand::Cancel => {
            if let Err(error) = folder.doc().leave().await {
                let _result = events.send(SyncEvent::Failed(error.to_string()));
                return false;
            }
            let _result = events.send(SyncEvent::Cancelled);
            false
        }
    }
}

fn include_event(params: &SubscribeParams, event: &LiveEvent) -> bool {
    if params.ingest_only && matches!(event, LiveEvent::InsertLocal { .. }) {
        return false;
    }
    if params.ignore_session.is_some() && matches!(event, LiveEvent::InsertLocal { .. }) {
        return false;
    }
    match event {
        LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
            let path = entry_path(entry);
            params.accepts(&path, &entry.content_hash())
        }
        LiveEvent::ContentReady { hash } => params
            .area_of_interest
            .as_ref()
            .is_none_or(|area| area.area.matches_hash(hash)),
        LiveEvent::PendingContentReady
        | LiveEvent::NeighborUp(_)
        | LiveEvent::NeighborDown(_)
        | LiveEvent::SyncFinished(_) => true,
    }
}

fn entry_path(entry: &iroh_docs::Entry) -> std::path::PathBuf {
    Path::new(String::from_utf8_lossy(entry.key()).as_ref()).to_path_buf()
}

/// Notify a folder's reconciler that content just arrived locally.
///
/// The daemon attaches a reconciler to writable folders so remote entries are
/// materialized (with conflict resolution) as soon as their blobs are local.
/// The reconciler is fire-and-forget: materialization is idempotent and skips
/// blobs that are already on disk.
fn reconcile_after_sync(namespace: NamespaceId, config: &IntentConfig) {
    if let Some(reconciler) = &config.reconciler {
        tracing::info!(%namespace, "sync engine invoking folder reconciler");
        reconciler(namespace);
    }
}

#[cfg(test)]
mod tests {
    use std::time::SystemTime;

    use iroh::SecretKey;
    use iroh_docs::engine::{LiveEvent, Origin, SyncEvent};

    use super::{IntentState, SubscribeParams};

    /// Regression: a `SyncFinished` event carrying an `Err` describes a failed
    /// sync *attempt* with one peer, which is routine in a swarm. It must not
    /// fail the intent (the old behaviour terminated the continuous intent and
    /// restarted it with backoff, dropping the live sync connection).
    #[test]
    fn failed_sync_attempt_does_not_fail_the_intent() {
        let event = LiveEvent::SyncFinished(SyncEvent {
            peer: SecretKey::generate().public(),
            origin: Origin::Accept,
            started: SystemTime::now(),
            finished: SystemTime::now(),
            result: Err("peer disconnected".to_owned()),
        });
        let mut state = IntentState::default();
        state.apply(event, &SubscribeParams::default(), None);
        assert!(
            state.sync_finished,
            "the intent should still record that a sync round finished"
        );
    }
}