syncweb-core 0.1.1-alpha.4

use std::{
    collections::{HashMap, HashSet},
    path::Path,
    pin::Pin,
    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>,
}

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,
        }
    }

    /// 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
    }

    #[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?;
        let live_events = self.docs_engine.watch(folder.doc()).await?;
        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");
                }
                Ok(_) => {}
                Err(error) => {
                    tracing::debug!(%folder_id, %error, "failed to discover peers via topic tracker");
                }
            }
        }
        self.docs_engine.start_sync(folder.doc(), Vec::new()).await?;
        let (events, commands, handle) = IntentHandle::channel();
        let node_db = self.node_db.clone();
        tokio::spawn(run_intent(
            folder,
            IntentConfig { mode, params, filter },
            self.blob_store.clone(),
            live_events,
            events,
            commands,
            node_db,
        ));
        Ok(handle)
    }
}

struct IntentConfig {
    mode: SessionMode,
    params: SubscribeParams,
    filter: Option<FilterEngine>,
}

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 Some(event_result) = next_event else {
                    let _result = events.send(SyncEvent::Finished);
                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                    return;
                };
                let live_event = match event_result {
                    Ok(event) => event,
                    Err(error) => {
                        let _result = events.send(SyncEvent::Failed(error.to_string()));
                        finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                        return;
                    }
                };
                if !include_event(&config.params, &live_event) {
                    continue;
                }
                let entry_key = match &live_event {
                    LiveEvent::InsertLocal { entry } | LiveEvent::InsertRemote { entry, .. } => {
                        Some(entry.key().to_vec())
                    }
                    LiveEvent::ContentReady { .. }
                    | LiveEvent::PendingContentReady
                    | LiveEvent::NeighborUp(_)
                    | LiveEvent::NeighborDown(_)
                    | LiveEvent::SyncFinished(_) => None,
                };
                if let Some(key) = &entry_key && completed_keys.contains(key) {
                    continue;
                }
                if let LiveEvent::ContentReady { hash } = &live_event {
                    match blob_store.has(*hash).await {
                        Ok(true) => {}
                        Ok(false) => {
                            let msg = format!("content-ready blob {hash} is missing locally");
                            let _result = events.send(SyncEvent::Failed(msg));
                            finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                            return;
                        }
                        Err(error) => {
                            let _result = events.send(SyncEvent::Failed(error.to_string()));
                            finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                            return;
                        }
                    }
                    if let Some(cmd) = throttle_bandwidth(hash, &state, &config, &mut commands).await {
                        let _ = handle_command(&folder, cmd, &mut state.paused, &events).await;
                        finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                        return;
                    }
                    if let Some(ref cp) = checkpoint {
                        let _ = cp.mark_completed(hash.as_bytes(), *hash, state.sizes.get(hash).copied().unwrap_or(0));
                    }
                }
                if let Err(error) = state.apply(live_event, &config.params, config.filter.as_ref()) {
                    let _result = events.send(SyncEvent::Failed(error));
                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                    return;
                }
                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));
                }
                if !send_progress(&events, &state, started.elapsed()) {
                    finalize_checkpoint(checkpoint.as_ref(), &state, &events);
                    return;
                }
            }
        }
    }
}

async fn throttle_bandwidth(
    hash: &iroh_blobs::Hash,
    state: &IntentState,
    config: &IntentConfig,
    commands: &mut mpsc::UnboundedReceiver<SyncCommand>,
) -> Option<SyncCommand> {
    if let Some(rate) = config.params.bandwidth.as_ref().and_then(|limits| limits.max_download)
        && rate > 0
    {
        let bytes = state.sizes.get(hash).copied().unwrap_or(0);
        let milliseconds = bytes.saturating_mul(1_000).div_ceil(rate);
        if milliseconds > 0 {
            tokio::select! {
                () = tokio::time::sleep(Duration::from_millis(milliseconds)) => {}
                received_command = commands.recv() => {
                    return received_command;
                }
            }
        }
    }
    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>,
    ) -> std::result::Result<(), String> {
        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) => return Err(error),
                }
            }
        }
        Ok(())
    }

    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()
}