syncweb-core 0.1.1-alpha.11

use std::{
    collections::{HashMap, HashSet},
    net::SocketAddr,
    path::{Path, PathBuf},
    sync::{Arc, Mutex, RwLock},
    time::{Duration, Instant},
};

use iroh_docs::NamespaceId;
use n0_future::StreamExt;
use tokio::{
    sync::{broadcast, mpsc, oneshot},
    task::JoinHandle,
};

use crate::{
    constants::NETWORK_MEMBERS_KEY,
    error::{Result, SyncwebError},
    filter::{FilterAction, FilterEngine, FilterEntry},
    folder::{FolderManager, PublicSubscription},
    fs::{FsWatcher, Importer},
    indexing::IndexingService,
    net::{NetworkLogger, NetworkManager, RelayManager, RelayManagerOptions},
    node::{identity::IdentityManager, iroh_node::IrohNode},
    schedule::ScheduleManager,
    storage::{config::SubscribeFilters, node_db::NodeDatabase, stats_db::StatsDatabase},
    sync::{SubscribeParams, SyncEngine, cancel_session, is_active},
    transfer_limits::{Direction, PendingTransfer, TransferBudget, TransferLimits, TransferQueue},
};

use crate::node::iroh_node::discovery_scope;

use super::{
    DaemonHandle, DaemonState, DaemonStatus, FolderEntry, IpcServer, ManagedPool, PidLock, WatchConfig,
    current_timestamp, daemon_socket_path,
    state::{BandwidthSnapshot, DaemonStatusReport, ScheduleStatus, load_filter_engine},
    supervisor::{BandwidthAccountant, DEFAULT_RETRY_WINDOW, IntentControls, IntentSupervisor, SupervisionOptions},
};

/// Configuration used to construct and run a daemon.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct DaemonConfig {
    pub data_dir: PathBuf,
    pub network: Option<String>,
    pub sync_interval: Duration,
    pub observation_ttl: Duration,
    /// Window after the first intent attempt during which failures are retried.
    pub retry_window: Duration,
    pub backoff_base: Duration,
    pub backoff_max: Duration,
    pub rayon_threads: usize,
    pub log_level: String,
    pub log_file: Option<PathBuf>,
    pub watch_debounce: Duration,
    pub relay_mode: crate::node::iroh_node::RelayMode,
    pub media_listen: Option<SocketAddr>,
    pub discovery: crate::node::iroh_node::DiscoveryConfig,
    /// Shared address lookup for tests that need multiple daemons to discover each other.
    pub address_lookup: Option<iroh::address_lookup::memory::MemoryLookup>,
    /// Syncthing (BEP) relay configuration. When `enabled`, the daemon starts
    /// a relay manager and installs its custom transport on the endpoint.
    pub bep: crate::storage::config::BepConfig,
}

impl Default for DaemonConfig {
    fn default() -> Self {
        Self {
            data_dir: PathBuf::from("."),
            network: None,
            sync_interval: Duration::from_mins(1),
            observation_ttl: Duration::from_hours(1),
            retry_window: DEFAULT_RETRY_WINDOW,
            backoff_base: Duration::from_secs(1),
            backoff_max: Duration::from_mins(1),
            rayon_threads: std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get),
            log_level: "info".to_owned(),
            log_file: None,
            watch_debounce: Duration::from_millis(500),
            relay_mode: crate::node::iroh_node::RelayMode::Default,
            media_listen: None,
            discovery: crate::node::iroh_node::DiscoveryConfig::default(),
            address_lookup: None,
            bep: crate::storage::config::BepConfig::default(),
        }
    }
}

impl DaemonConfig {
    #[must_use]
    pub fn new(data_dir: impl Into<PathBuf>) -> Self {
        Self {
            data_dir: data_dir.into(),
            ..Self::default()
        }
    }
}

/// Owns the Iroh node, synchronization intents, IPC server, and daemon
/// lifecycle state.
pub struct Daemon {
    config: DaemonConfig,
    node_db: NodeDatabase,
    stats_db: StatsDatabase,
    pid_lock: PidLock,
    ipc_server: IpcServer,
    intent_supervisor: IntentSupervisor,
    folder_manager: FolderManager,
    sync_engine: SyncEngine,
    schedule_manager: tokio::sync::RwLock<Option<ScheduleManager>>,
    transfer_budget: Arc<TransferBudget>,
    transfer_queue: Arc<TransferQueue>,
    /// Folders whose downloads were deferred to satisfy a download quota, so
    /// the default download policy can be restored when limits are removed.
    deferred_downloads: Mutex<HashSet<NamespaceId>>,
    node: Arc<IrohNode>,
    handle: DaemonHandle,
    sync_receiver: tokio::sync::Mutex<mpsc::UnboundedReceiver<Option<String>>>,
    intent_tasks: Mutex<HashMap<NamespaceId, Option<JoinHandle<()>>>>,
    intent_controls: IntentControls,
    watchers: Mutex<HashMap<String, FsWatcher>>,
    pending_watch_events: Mutex<HashMap<String, PendingWatch>>,
    filter_engine: tokio::sync::RwLock<Option<FilterEngine>>,
    archive_pool: Arc<ManagedPool>,
    network_logger: NetworkLogger,
    network_manager: tokio::sync::RwLock<NetworkManager>,
    maintenance_task: tokio::sync::Mutex<Option<JoinHandle<()>>>,
    indexing: Option<IndexingService>,
}

struct PendingWatch {
    paths: HashMap<PathBuf, bool>,
    ready_at: Instant,
}

/// The filter and ignore-pattern set applied to one folder's watch events.
#[derive(Clone, Copy)]
struct WatchEvaluation<'a> {
    filter: Option<&'a FilterEngine>,
    exclude: &'a [String],
}

/// The daemon's spawned listener tasks: Unix-socket IPC, WebSocket bridge, and
/// the optional media HTTP server.
type ServerTasks = (
    JoinHandle<Result<()>>,
    Option<JoinHandle<Result<()>>>,
    Option<JoinHandle<Result<()>>>,
);

type SignalTask<'a> = std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = Result<()>> + Send + 'a>>;

impl Daemon {
    /// Return a clone of the daemon's iroh endpoint, useful for awaiting
    /// relay connectivity in tests.
    #[must_use]
    pub fn endpoint(&self) -> iroh::Endpoint {
        self.node.endpoint().clone()
    }

    #[expect(
        clippy::too_many_arguments,
        reason = "daemon bootstrap passes each subsystem its config"
    )]
    async fn open_identity_and_node(
        data_dir: &Path,
        node_db: &NodeDatabase,
        pid_lock: &PidLock,
        relay_mode: crate::node::iroh_node::RelayMode,
        discovery: crate::node::iroh_node::DiscoveryConfig,
        member_keys: Arc<RwLock<HashSet<iroh::PublicKey>>>,
        address_lookup: Option<iroh::address_lookup::memory::MemoryLookup>,
        bep: &crate::storage::config::BepConfig,
    ) -> Result<(Arc<IrohNode>, FolderManager, SyncEngine)> {
        let identity = IdentityManager::new(data_dir.join("identity.key")).inspect_err(|_| {
            let _ = node_db.remove_lifecycle();
            let _ = pid_lock.release();
        })?;
        let err_cleanup = |_: &_| {
            let _ = node_db.remove_lifecycle();
            let _ = pid_lock.release();
        };
        let relay = if bep.enabled {
            let options = RelayManagerOptions::new(bep.relay_config(), data_dir.join("relay"));
            Some(RelayManager::start(options, identity.node_id()).inspect_err(err_cleanup)?)
        } else {
            None
        };
        let node = Arc::new(match address_lookup {
            Some(lookup) => IrohNode::new_with_address_lookup_and_relay(
                identity,
                data_dir.join("data"),
                relay_mode,
                lookup,
                discovery,
                member_keys,
                relay,
            )
            .await
            .inspect_err(err_cleanup)?,
            None => IrohNode::new_with_relay(
                identity,
                data_dir.join("data"),
                relay_mode,
                member_keys,
                discovery,
                relay,
            )
            .await
            .inspect_err(err_cleanup)?,
        });
        let folder_manager = FolderManager::new(&node);
        let sync_engine = SyncEngine::new(
            folder_manager.clone(),
            node.blob_store().clone(),
            node.docs_engine().clone(),
            Some(node.topic_tracker().clone()),
        );
        Ok((node, folder_manager, sync_engine))
    }

    /// Open node and stats databases and acquire the process lock.
    ///
    /// # Errors
    ///
    /// Returns an error if a database cannot be opened or the lock is held by
    /// another process.
    fn init_databases(config: &DaemonConfig) -> Result<(NodeDatabase, StatsDatabase, PidLock)> {
        let node_db = NodeDatabase::open(config.data_dir.join("node.db"))?;
        let stats_db = StatsDatabase::open(config.data_dir.join("stats.db"))?;
        let pid_lock = PidLock::new(&config.data_dir);
        if !pid_lock.try_acquire()? {
            return Err(SyncwebError::operation(
                "daemon already running",
                config.data_dir.display(),
            ));
        }
        Ok((node_db, stats_db, pid_lock))
    }

    /// Load app config, schedule manager, filter engine, and write initial lifecycle state.
    ///
    /// # Errors
    ///
    /// Returns an error if loading config, filters, or saving the lifecycle fails.
    fn load_app_state(
        node_db: &NodeDatabase,
        pid_lock: &PidLock,
        data_dir: &Path,
    ) -> Result<(
        Option<ScheduleManager>,
        Option<FilterEngine>,
        DaemonState,
        TransferLimits,
    )> {
        let app_config = node_db.load_app_config().inspect_err(|_error| {
            let _ = pid_lock.release();
        })?;
        let use_schedule = app_config.schedule != crate::schedule::ScheduleConfig::default();
        let schedule_manager = match ScheduleManager::from_config(&app_config.schedule) {
            Ok(s) if use_schedule => Some(s),
            Ok(_) => None,
            Err(error) => {
                let _ = pid_lock.release();
                return Err(error);
            }
        };
        let transfer_limits = TransferLimits::from_config(&app_config.transfer_limits).inspect_err(|_error| {
            let _ = pid_lock.release();
        })?;
        let filter_engine = load_filter_engine(node_db, data_dir).inspect_err(|_error| {
            let _ = pid_lock.release();
        })?;
        let initial_state = DaemonState::new(
            std::process::id(),
            String::new(),
            current_timestamp(),
            data_dir,
            DaemonStatus::Starting,
        );
        node_db.save_lifecycle(&initial_state).inspect_err(|_error| {
            let _ = pid_lock.release();
        })?;
        Ok((schedule_manager, filter_engine, initial_state, transfer_limits))
    }

    /// Create a daemon, acquire its process lock, and persist its running
    /// state.
    ///
    /// # Errors
    ///
    /// Returns an error if another daemon owns the data directory, the node
    /// cannot be opened, or configuration cannot be parsed.
    pub async fn new(config: DaemonConfig) -> Result<Self> {
        std::fs::create_dir_all(&config.data_dir)?;
        let (node_db, stats_db, pid_lock) = Self::init_databases(&config)?;
        node_db.recover_transfer_jobs()?;
        let (schedule_manager, filter_engine, initial_state, transfer_limits) =
            Self::load_app_state(&node_db, &pid_lock, &config.data_dir)?;

        let member_keys: Arc<RwLock<HashSet<iroh::PublicKey>>> = {
            let networks = node_db.list_networks()?;
            let keys: HashSet<iroh::PublicKey> = networks.iter().flat_map(|n| n.members.iter().copied()).collect();
            Arc::new(RwLock::new(keys))
        };
        let mut node_discovery = config.discovery.clone();
        node_discovery.scope = discovery_scope(config.network.as_deref());
        let (node, folder_manager, mut sync_engine) = match Self::open_identity_and_node(
            &config.data_dir,
            &node_db,
            &pid_lock,
            config.relay_mode.clone(),
            node_discovery,
            member_keys.clone(),
            config.address_lookup.clone(),
            &config.bep,
        )
        .await
        {
            Ok(result) => result,
            Err(error) => {
                let _ = node_db.remove_lifecycle();
                let _ = pid_lock.release();
                return Err(error);
            }
        };
        sync_engine = sync_engine
            .with_node_db(node_db.clone())
            .with_reconciler(build_reconciler(node.clone(), folder_manager.clone(), node_db.clone()));

        let archive_pool = match ManagedPool::new("syncweb-archive", config.rayon_threads) {
            Ok(value) => Arc::new(value),
            Err(error) => {
                let _ = node.stop().await;
                let _ = node_db.remove_lifecycle();
                let _ = pid_lock.release();
                return Err(SyncwebError::operation("failed to create daemon thread pool", error));
            }
        };

        let initial_handle = DaemonHandle::new(initial_state);
        {
            let mut state = initial_handle.state.write().await;
            state.node_id = node.endpoint().id().to_string();
            state.status = DaemonStatus::Running;
            let running_state = state.clone();
            drop(state);
            if let Err(error) = node_db.save_lifecycle(&running_state) {
                let _ = node.stop().await;
                let _ = node_db.remove_lifecycle();
                let _ = pid_lock.release();
                return Err(error);
            }
        }
        let (sync_sender, sync_receiver) = mpsc::unbounded_channel();
        let handle = DaemonHandle::with_channels_and_reload(
            initial_handle.state.clone(),
            initial_handle.folder_registry.clone(),
            initial_handle.shutdown_sender.clone(),
            sync_sender,
            initial_handle.reload_requested.clone(),
        );
        let indexing = match IndexingService::new(config.data_dir.join("indexing.sqlite")) {
            Ok(indexing) => Some(indexing),
            Err(error) => {
                tracing::warn!(%error, "failed to open indexing database; catalog channels disabled");
                None
            }
        };
        let intent_supervisor = IntentSupervisor::new(config.retry_window, config.backoff_base, config.backoff_max);
        let network_logger = NetworkLogger::new(stats_db.clone());
        let local_node_id = node.endpoint().id();
        let nm_result =
            Self::create_network_manager(&node_db, &local_node_id, &network_logger, Arc::clone(&member_keys));
        let network_manager = match nm_result {
            Ok(nm) => nm,
            Err(error) => {
                tracing::warn!("failed to create network manager with logger: {error}");
                NetworkManager::new(node_db.clone(), local_node_id, member_keys)?
            }
        };
        let mut ipc_server = Self::build_ipc_server(
            &config,
            &handle,
            &node,
            &archive_pool,
            &folder_manager,
            &node_db,
            indexing.clone(),
        );
        ipc_server = ipc_server.with_network_manager(Arc::new(tokio::sync::RwLock::new(network_manager.clone())));

        Ok(Self {
            config,
            node_db,
            stats_db,
            pid_lock,
            ipc_server,
            intent_supervisor,
            folder_manager,
            sync_engine,
            schedule_manager: tokio::sync::RwLock::new(schedule_manager),
            transfer_budget: Arc::new(TransferBudget::new(transfer_limits)),
            transfer_queue: Arc::new(TransferQueue::new()),
            deferred_downloads: Mutex::new(HashSet::new()),
            node,
            handle,
            sync_receiver: tokio::sync::Mutex::new(sync_receiver),
            intent_tasks: Mutex::new(HashMap::new()),
            intent_controls: Arc::new(Mutex::new(HashMap::new())),
            watchers: Mutex::new(HashMap::new()),
            pending_watch_events: Mutex::new(HashMap::new()),
            filter_engine: tokio::sync::RwLock::new(filter_engine),
            archive_pool,
            network_logger,
            network_manager: tokio::sync::RwLock::new(network_manager),
            maintenance_task: tokio::sync::Mutex::new(None),
            indexing,
        })
    }

    fn create_network_manager(
        node_db: &NodeDatabase,
        local_node_id: &iroh::PublicKey,
        network_logger: &NetworkLogger,
        member_keys: Arc<RwLock<HashSet<iroh::PublicKey>>>,
    ) -> Result<NetworkManager> {
        NetworkManager::with_logger(node_db.clone(), *local_node_id, network_logger.clone(), member_keys)
    }

    fn build_ipc_server(
        config: &DaemonConfig,
        handle: &DaemonHandle,
        node: &Arc<IrohNode>,
        archive_pool: &Arc<ManagedPool>,
        folder_manager: &FolderManager,
        node_db: &NodeDatabase,
        indexing: Option<IndexingService>,
    ) -> IpcServer {
        IpcServer::with_archive_context(
            daemon_socket_path(&config.data_dir),
            config.data_dir.clone(),
            handle.clone(),
            node.clone(),
            archive_pool.clone(),
            indexing,
        )
        .with_folder_manager(folder_manager.clone())
        .with_node_db(node_db.clone())
    }

    /// Run the daemon until IPC or operating-system shutdown is requested.
    ///
    /// # Errors
    ///
    /// Returns an error if an IPC listener, sync cycle, or cleanup operation
    /// fails.
    pub async fn run(&self) -> Result<()> {
        let run_result = self.run_inner().await;
        let cleanup_result = self.shutdown_resources().await;
        match run_result {
            Err(error) => {
                cleanup_result?;
                Err(error)
            }
            Ok(()) => cleanup_result,
        }
    }

    /// Return a snapshot of the daemon lifecycle state.
    pub async fn state(&self) -> DaemonState {
        self.handle.state.read().await.clone()
    }

    /// Return the daemon's fixed archive pool.
    #[must_use]
    pub fn archive_pool(&self) -> &ManagedPool {
        self.archive_pool.as_ref()
    }

    async fn run_inner(&self) -> Result<()> {
        tracing::debug!(
            rayon_threads = self.archive_pool.thread_count(),
            "daemon runtime initialized"
        );
        self.run_initial_setup().await?;
        let (mut server_task, bridge_task, media_task) = self.spawn_server_tasks()?;
        let (mut shutdown, mut signal_task, mut interval, mut watch_interval, shutdown_sender) = self.runtime_timers();
        if let Err(error) = self.run_cycle().await {
            tracing::error!(%error, "initial daemon cycle failed");
        }
        self.spawn_listeners().await;
        let result = self
            .run_event_loop(
                &mut server_task,
                &mut signal_task,
                &mut shutdown,
                &shutdown_sender,
                &mut interval,
                &mut watch_interval,
            )
            .await;
        send_shutdown(&shutdown_sender);
        self.wait_for_server_tasks(server_task, bridge_task, media_task).await?;
        result
    }

    async fn run_initial_setup(&self) -> Result<()> {
        self.load_folders().await?;
        self.reannounce_folders().await;
        self.load_subscriptions().await?;
        self.start_watching().await?;
        self.spawn_membership_listeners().await;
        self.subscribe_channels().await;
        Ok(())
    }

    fn runtime_timers(
        &self,
    ) -> (
        broadcast::Receiver<()>,
        SignalTask<'_>,
        tokio::time::Interval,
        tokio::time::Interval,
        broadcast::Sender<()>,
    ) {
        let shutdown = self.handle.shutdown_sender.subscribe();
        let shutdown_sender = self.handle.shutdown_sender.clone();
        let signal_task: SignalTask<'_> = Box::pin(self.handle_signals(shutdown_sender.clone()));
        let interval_duration = self.config.sync_interval.max(Duration::from_millis(1));
        let interval = tokio::time::interval(interval_duration);
        let watch_interval = tokio::time::interval(Duration::from_millis(100));
        (shutdown, signal_task, interval, watch_interval, shutdown_sender)
    }

    async fn spawn_listeners(&self) {
        self.spawn_maintenance_task().await;
    }

    fn spawn_server_tasks(&self) -> Result<ServerTasks> {
        let server = self.ipc_server.clone();
        let server_task = tokio::spawn(async move { server.serve().await });

        let bridge =
            crate::media::BridgeServer::new(crate::media::bridge::DEFAULT_BRIDGE_LISTEN, self.ipc_server.clone())?;
        let bridge_shutdown = self.handle.shutdown_sender.clone();
        let bridge_task = tokio::spawn(async move { bridge.run(bridge_shutdown).await });

        let media_task = self.config.media_listen.map(|addr| {
            let media_srv = crate::media::MediaServer::new(addr, self.node.blob_store().clone());
            let media_shutdown = self.handle.shutdown_sender.clone();
            tokio::spawn(async move { media_srv.run(media_shutdown).await })
        });

        Ok((server_task, Some(bridge_task), media_task))
    }

    async fn wait_for_server_tasks(
        &self,
        server_task: JoinHandle<Result<()>>,
        bridge_task: Option<JoinHandle<Result<()>>>,
        media_task: Option<JoinHandle<Result<()>>>,
    ) -> Result<()> {
        if !server_task.is_finished() {
            match server_task.await {
                Ok(server_result) => server_result?,
                Err(error) => return Err(SyncwebError::operation("daemon IPC task failed", error)),
            }
        }
        if let Some(task) = bridge_task
            && !task.is_finished()
            && let Err(error) = task.await
        {
            tracing::warn!(%error, "websocket bridge task failed");
        }
        if let Some(task) = media_task
            && !task.is_finished()
            && let Err(error) = task.await
        {
            tracing::warn!(%error, "media server task failed");
        }
        Ok(())
    }

    async fn run_event_loop(
        &self,
        server_task: &mut JoinHandle<Result<()>>,
        signal_task: &mut SignalTask<'_>,
        shutdown: &mut broadcast::Receiver<()>,
        shutdown_sender: &broadcast::Sender<()>,
        interval: &mut tokio::time::Interval,
        watch_interval: &mut tokio::time::Interval,
    ) -> Result<()> {
        loop {
            tokio::select! {
                signal_result = &mut *signal_task => {
                    signal_result?;
                    send_shutdown(shutdown_sender);
                    break Ok(());
                }
                shutdown_result = shutdown.recv() => {
                    if matches!(shutdown_result, Ok(()) | Err(broadcast::error::RecvError::Closed)) {
                        break Ok(());
                    }
                }
                server_result = &mut *server_task => {
                    match server_result {
                        Ok(result) => break result,
                        Err(error) => break Err(SyncwebError::operation("daemon IPC task failed", error)),
                    }
                }
                trigger = self.receive_sync_trigger() => {
                    match trigger {
                        Some(namespace) => self.run_trigger(namespace).await?,
                        None => break Ok(()),
                    }
                }
                _ = watch_interval.tick() => self.handle_watch_events().await?,
                _ = interval.tick() => self.run_cycle().await?,
            }
        }
    }

    async fn run_cycle(&self) -> Result<()> {
        self.reload_if_requested().await?;
        if let Err(error) = self.load_folders().await {
            tracing::warn!(%error, "folder load cycle failed — continuing");
        }
        self.start_watching().await?;
        self.refresh_transfer_limits();
        let live_folders = self.enabled_subscribe_filters();
        let statuses = self.handle.folder_registry.read().await.statuses();
        for folder in statuses {
            let namespace = folder
                .namespace
                .parse::<NamespaceId>()
                .map_err(|error| SyncwebError::operation("invalid managed folder namespace", error))?;
            let folder_name = (!folder.path.as_os_str().is_empty()).then(|| folder.path.to_string_lossy().into_owned());
            let Some(filters) = live_folders.get(&namespace) else {
                if is_active(namespace) && !cancel_session(namespace) {
                    tracing::warn!(%namespace, "live syncing is disabled; intent did not accept cancellation");
                }
                continue;
            };
            self.apply_folder_schedule(namespace, filters, folder_name.as_deref())
                .await?;
        }
        self.enforce_transfer_limits().await?;
        self.save_status_report().await?;
        Ok(())
    }

    /// Start or stop one folder's live session according to its schedule.
    async fn apply_folder_schedule(
        &self,
        namespace: NamespaceId,
        filters: &SubscribeFilters,
        folder_name: Option<&str>,
    ) -> Result<()> {
        let active = {
            let schedule_manager = self.schedule_manager.read().await;
            schedule_manager
                .as_ref()
                .is_none_or(|manager| manager.is_active(folder_name))
        };
        if active {
            self.start_supervision(namespace, filters).await?;
            self.set_intent_active(namespace, true)
        } else if cancel_session(namespace) {
            tracing::debug!(%namespace, "outside scheduled active window; pausing intent");
            Ok(())
        } else {
            Ok(())
        }
    }

    /// Apply the configured per-hour/day/month transfer limits.
    ///
    /// When a download quota is configured, iroh-docs eager downloads are
    /// disabled for managed folders and remote blobs are queued; the queue is
    /// drained only while the rolling-window budget allows. When an upload
    /// quota is exhausted, live sessions are paused so the node stops seeding
    /// until the window frees budget.
    /// Re-read `[transfer_limits]` from the node database into the live budget.
    fn refresh_transfer_limits(&self) {
        if let Ok(config) = self.node_db.load_app_config()
            && let Ok(limits) = TransferLimits::from_config(&config.transfer_limits)
        {
            self.transfer_budget.set_limits(limits);
        }
    }

    async fn enforce_transfer_limits(&self) -> Result<()> {
        let download_limited = !self.transfer_budget.limits().download.is_unlimited();
        let upload_limited = !self.transfer_budget.limits().upload.is_unlimited();
        if !download_limited && !upload_limited {
            return Ok(());
        }

        if upload_limited && self.transfer_budget.available(Direction::Upload) == 0 {
            self.pause_sessions_for_upload_limit().await;
        }

        if !download_limited {
            return self.restore_deferred_downloads().await;
        }

        let statuses = self.handle.folder_registry.read().await.statuses();
        let mut deferred_now = Vec::new();
        for status in &statuses {
            if let Ok(namespace) = status.namespace.parse::<NamespaceId>() {
                self.defer_folder_downloads(namespace, &mut deferred_now).await;
            }
        }
        self.record_deferred_downloads(deferred_now)?;
        self.drain_transfer_queue().await;
        Ok(())
    }

    /// Stop seeding by cancelling every live session while upload budget is spent.
    async fn pause_sessions_for_upload_limit(&self) {
        let statuses = self.handle.folder_registry.read().await.statuses();
        for status in &statuses {
            if let Ok(namespace) = status.namespace.parse::<NamespaceId>()
                && cancel_session(namespace)
            {
                tracing::info!(%namespace, "upload budget exhausted; pausing live session");
            }
        }
    }

    /// Restore the default download policy for folders we previously deferred.
    async fn restore_deferred_downloads(&self) -> Result<()> {
        let to_clear: Vec<NamespaceId> = {
            let mut deferred = self
                .deferred_downloads
                .lock()
                .map_err(|error| SyncwebError::operation("deferred download mutex is poisoned", error))?;
            deferred.drain().collect()
        };
        for namespace in to_clear {
            if let Ok(folder) = self.folder_manager.get(namespace).await {
                let _ = folder.clear_metadata_only().await;
            }
        }
        Ok(())
    }

    /// Disable eager downloads for a folder and queue its remote blobs.
    async fn defer_folder_downloads(&self, namespace: NamespaceId, deferred: &mut Vec<NamespaceId>) {
        let Ok(folder) = self.folder_manager.get(namespace).await else {
            return;
        };
        if let Err(error) = folder.set_metadata_only().await {
            tracing::debug!(%namespace, %error, "could not defer content downloads for transfer limits");
            return;
        }
        deferred.push(namespace);
        let Ok(entries) = folder.content_entries().await else {
            return;
        };
        for entry in entries {
            let hash = entry.content_hash();
            if folder.has_local(hash).await.unwrap_or(false) {
                continue;
            }
            let path = PathBuf::from(String::from_utf8_lossy(entry.key()).into_owned());
            self.transfer_queue
                .enqueue(PendingTransfer::new(namespace, hash, entry.content_len(), path));
        }
    }

    fn record_deferred_downloads(&self, deferred_now: Vec<NamespaceId>) -> Result<()> {
        let mut deferred = self
            .deferred_downloads
            .lock()
            .map_err(|error| SyncwebError::operation("deferred download mutex is poisoned", error))?;
        deferred.extend(deferred_now);
        drop(deferred);
        Ok(())
    }

    /// Fetch queued blobs while the download budget allows.
    async fn drain_transfer_queue(&self) {
        while let Some(item) = self.transfer_queue.pop_fitting(&self.transfer_budget) {
            let peers = self
                .node
                .topic_tracker()
                .find_peers(item.namespace)
                .await
                .unwrap_or_default();
            let mut fetched = false;
            for peer in peers {
                if self
                    .node
                    .blob_store()
                    .force_fetch_from_peer(self.node.endpoint(), &peer, item.hash)
                    .await
                    .is_ok()
                {
                    fetched = true;
                    break;
                }
            }
            if fetched {
                self.transfer_budget.record(Direction::Download, item.size);
                let network_id = self
                    .network_manager
                    .read()
                    .await
                    .network_for_folder(&item.namespace)
                    .ok()
                    .flatten();
                let folder = item.namespace.to_string();
                if let Err(error) = self.network_logger.record_bandwidth_download(
                    item.size,
                    1,
                    Some(&folder),
                    None,
                    network_id.as_deref(),
                ) {
                    tracing::warn!(%error, "failed to record queued download");
                }
                tracing::debug!(
                    namespace = %item.namespace,
                    hash = %item.hash,
                    size = item.size,
                    queued = self.transfer_queue.len(),
                    "queued transfer fetched within budget"
                );
            } else {
                // No peer served it yet; keep it queued for a later cycle.
                self.transfer_queue.enqueue(item);
                break;
            }
        }
    }

    async fn run_trigger(&self, namespace: Option<String>) -> Result<()> {
        match namespace {
            Some(value) => self.trigger_namespace(value).await,
            None => self.run_cycle().await,
        }
    }

    async fn trigger_namespace(&self, value: String) -> Result<()> {
        let parsed_namespace = value
            .parse::<NamespaceId>()
            .map_err(|error| SyncwebError::operation("invalid sync namespace", error))?;
        let tracked = self
            .handle
            .folder_registry
            .read()
            .await
            .statuses()
            .iter()
            .any(|status| status.namespace == value);
        if !tracked {
            tracing::warn!(%parsed_namespace, "not a syncweb folder; nothing to synchronize");
            return Ok(());
        }
        let Some(filters) = self.enabled_subscribe_filters().get(&parsed_namespace).cloned() else {
            tracing::warn!(
                %parsed_namespace,
                "live syncing is disabled for this folder; enable it with `join --subscribe` \
                 or `config set <folder>.subscribe on`"
            );
            return Ok(());
        };
        if !self.schedule_active_for(parsed_namespace).await {
            tracing::debug!(%parsed_namespace, "outside scheduled active window; not starting sync");
            if cancel_session(parsed_namespace) {
                self.save_status_report().await?;
            }
            return Ok(());
        }
        self.start_supervision(parsed_namespace, &filters).await?;
        self.set_intent_active(parsed_namespace, true)?;
        self.save_status_report().await?;
        Ok(())
    }

    /// Return whether the schedule lets a folder synchronize right now.
    async fn schedule_active_for(&self, namespace: NamespaceId) -> bool {
        let folder_name = self
            .handle
            .folder_registry
            .read()
            .await
            .statuses()
            .into_iter()
            .find(|status| status.namespace == namespace.to_string())
            .and_then(|status| {
                (!status.path.as_os_str().is_empty()).then(|| status.path.to_string_lossy().into_owned())
            });
        let schedule_manager = self.schedule_manager.read().await;
        schedule_manager
            .as_ref()
            .is_none_or(|manager| manager.is_active(folder_name.as_deref()))
    }

    /// The `subscribe-changes` folders with live syncing enabled, mapped to their filters.
    fn enabled_subscribe_filters(&self) -> std::collections::BTreeMap<NamespaceId, SubscribeFilters> {
        let Ok(config) = self.node_db.load_app_config() else {
            return std::collections::BTreeMap::new();
        };
        config
            .subscribe
            .folders
            .into_iter()
            .filter(|(_, entry)| entry.enabled)
            .filter_map(|(namespace, entry)| namespace.parse::<NamespaceId>().ok().map(|id| (id, entry.filters)))
            .collect()
    }

    async fn start_supervision(&self, namespace: NamespaceId, filters: &SubscribeFilters) -> Result<()> {
        self.refresh_transfer_limits();
        {
            let mut tasks = self
                .intent_tasks
                .lock()
                .map_err(|error| SyncwebError::operation("daemon intent task mutex is poisoned", error))?;
            tasks.retain(|_, task| task.as_ref().is_none_or(|handle| !handle.is_finished()));
            if tasks.contains_key(&namespace) || is_active(namespace) {
                return Ok(());
            }
            tasks.insert(namespace, None);
        }
        // With a download quota configured, defer content before iroh-docs can
        // eagerly fetch it; the transfer queue refills it within budget.
        if !self.transfer_budget.limits().download.is_unlimited()
            && let Ok(folder) = self.folder_manager.get(namespace).await
            && let Err(error) = folder.set_metadata_only().await
        {
            tracing::debug!(%namespace, %error, "could not defer downloads before supervising");
        }
        let network_id = {
            let guard = self.network_manager.read().await;
            guard.network_for_folder(&namespace)?
        };
        let session_id = network_id.as_ref().and_then(|net_id| {
            self.network_logger
                .record_sync_start(net_id, &namespace.to_string())
                .ok()
        });
        let sync = self.sync_engine.clone();
        let supervisor = self.intent_supervisor;
        let shutdown = self.handle.shutdown_sender.subscribe();
        let controls = self.intent_controls.clone();
        let filter = self.filter_engine.read().await.clone();
        let network_logger = self.network_logger.clone();
        let params = match SubscribeParams::from_filters(filters) {
            Ok(params) => params,
            Err(error) => {
                tracing::warn!(%namespace, %error, "invalid subscription filters; supervising without them");
                SubscribeParams::default()
            }
        };
        let (ready_sender, ready_receiver) = oneshot::channel();
        let accountant = BandwidthAccountant::new(network_logger.clone(), network_id.clone(), namespace.to_string());
        let task = tokio::spawn(async move {
            let options =
                SupervisionOptions::with_ready(controls, filter, ready_sender).with_bandwidth(accountant.clone());
            let result = supervisor
                .supervise_with_controls_and_ready(&sync, namespace, params, shutdown, options)
                .await;
            if network_id.is_some()
                && let Some(sid) = session_id
            {
                match &result {
                    Ok(supervised) => {
                        let has_error = supervised.last_error.is_some();
                        let bytes = accountant.total();
                        let files = supervised.files_transferred;
                        let errors = u64::from(has_error);
                        let status = if has_error { "failed" } else { "completed" };
                        let _ = network_logger.record_sync_finish(sid, files, bytes, errors, status);
                    }
                    Err(_) => {
                        let _ = network_logger.record_sync_finish(sid, 0, 0, 1, "failed");
                    }
                }
            }
            match &result {
                Ok(supervised) => {
                    if let Some(error) = &supervised.last_error {
                        tracing::warn!(%namespace, retry_count = supervised.retry_count, %error, "supervised intent stopped");
                    }
                }
                Err(error) => tracing::error!(%namespace, %error, "supervised intent failed"),
            }
        });
        self.intent_tasks
            .lock()
            .map_err(|error| SyncwebError::operation("daemon intent task mutex is poisoned", error))?
            .insert(namespace, Some(task));
        ready_receiver
            .await
            .map_err(|error| SyncwebError::operation("daemon intent startup notification failed", error))?;
        Ok(())
    }

    fn set_intent_active(&self, namespace: NamespaceId, active: bool) -> Result<()> {
        let control = self
            .intent_controls
            .lock()
            .map_err(|error| SyncwebError::operation("daemon intent control mutex is poisoned", error))?
            .get(&namespace)
            .cloned();
        let Some(intent_control) = control else {
            return Ok(());
        };
        let result = if active {
            intent_control.resume()
        } else {
            intent_control.pause()
        };
        result.map_err(|error| SyncwebError::operation("failed to update scheduled intent", error))
    }

    /// Return whether the global schedule is currently active.
    pub async fn is_in_active_window(&self) -> bool {
        self.schedule_manager
            .read()
            .await
            .as_ref()
            .is_none_or(|manager| manager.is_active(None))
    }

    async fn start_watching(&self) -> Result<()> {
        let statuses = self.handle.folder_registry.read().await.statuses();
        let wanted: HashMap<_, _> = statuses
            .iter()
            .filter(|status| !status.path.as_os_str().is_empty())
            .map(|status| (status.namespace.clone(), status.path.clone()))
            .collect();
        let candidates = self.watcher_candidates(&wanted)?;
        let mut to_add = Vec::new();
        for (namespace, path) in candidates {
            if self.watcher_allowed(&namespace, &path).await {
                to_add.push((namespace, path));
            }
        }
        self.install_watchers(to_add)
    }

    /// Drop watchers whose folder left the registry and return the newly-wanted
    /// namespaces. The watcher mutex is a `std::sync::Mutex`, so it is dropped
    /// before the async capability lookup in [`Self::watcher_allowed`].
    fn watcher_candidates(&self, wanted: &HashMap<String, PathBuf>) -> Result<Vec<(String, PathBuf)>> {
        let mut watchers = self
            .watchers
            .lock()
            .map_err(|error| SyncwebError::operation("daemon watcher mutex is poisoned", error))?;
        watchers.retain(|namespace, _| wanted.contains_key(namespace));
        let candidates = wanted
            .iter()
            .filter(|(namespace, _)| !watchers.contains_key(*namespace))
            .map(|(namespace, path)| (namespace.clone(), path.clone()))
            .collect();
        drop(watchers);
        Ok(candidates)
    }

    /// Whether a filesystem watcher should be created for `namespace`.
    ///
    /// A replica joined from a read-only ticket cannot accept local writes:
    /// watching it would try to import every materialized file and fail in a
    /// retry loop, so it is skipped entirely.
    async fn watcher_allowed(&self, namespace: &str, path: &Path) -> bool {
        let Ok(namespace_id) = namespace.parse::<NamespaceId>() else {
            return false;
        };
        match self.folder_manager.get(namespace_id).await {
            Ok(folder) if !folder.is_writable() => {
                tracing::debug!(%namespace, "read-only folder; not watching for local changes");
                false
            }
            Ok(_) if path.exists() => true,
            Ok(_) => {
                tracing::warn!(%namespace, path = %path.display(), "managed folder path does not exist; watcher deferred");
                false
            }
            Err(error) => {
                tracing::debug!(%namespace, %error, "folder capability unavailable; watcher deferred");
                false
            }
        }
    }

    fn install_watchers(&self, to_add: Vec<(String, PathBuf)>) -> Result<()> {
        let mut watchers = self
            .watchers
            .lock()
            .map_err(|error| SyncwebError::operation("daemon watcher mutex is poisoned", error))?;
        for (namespace, path) in to_add {
            if let std::collections::hash_map::Entry::Vacant(slot) = watchers.entry(namespace) {
                slot.insert(FsWatcher::new(&path)?);
            }
        }
        Ok(())
    }

    async fn handle_watch_events(&self) -> Result<()> {
        // Reconcile watchers with the registry on every tick so a `watch`
        // registration is picked up promptly rather than at the next sync
        // interval.
        self.start_watching().await?;
        let mut observed = Vec::new();
        {
            let mut watchers = self
                .watchers
                .lock()
                .map_err(|error| SyncwebError::operation("daemon watcher mutex is poisoned", error))?;
            for (namespace, watcher) in watchers.iter_mut() {
                loop {
                    match watcher.try_recv() {
                        Ok(Some(event)) => observed.push((namespace.clone(), event)),
                        Ok(None) => break,
                        Err(error) => {
                            tracing::warn!(%namespace, %error, "filesystem watcher event channel failed");
                            break;
                        }
                    }
                }
            }
        }

        if !observed.is_empty() {
            let mut pending = self
                .pending_watch_events
                .lock()
                .map_err(|error| SyncwebError::operation("daemon watch queue mutex is poisoned", error))?;
            let ready_at = Instant::now()
                .checked_add(self.config.watch_debounce)
                .unwrap_or_else(Instant::now);
            for (namespace, event) in observed {
                let removed = matches!(event.event.kind, notify::EventKind::Remove(_));
                let entry = pending.entry(namespace).or_insert_with(|| PendingWatch {
                    paths: HashMap::new(),
                    ready_at,
                });
                entry.ready_at = ready_at;
                for path in event.paths {
                    entry.paths.insert(path, removed);
                }
            }
            drop(pending);
        }

        let ready = {
            let mut pending = self
                .pending_watch_events
                .lock()
                .map_err(|error| SyncwebError::operation("daemon watch queue mutex is poisoned", error))?;
            let now = Instant::now();
            pending
                .iter_mut()
                .filter_map(|(namespace, batch)| {
                    (batch.ready_at <= now).then(|| (namespace.clone(), std::mem::take(&mut batch.paths)))
                })
                .filter(|(_, paths)| !paths.is_empty())
                .collect::<Vec<_>>()
        };
        if ready.is_empty() {
            return Ok(());
        }

        let (roots, watch_configs): (HashMap<_, _>, HashMap<_, _>) = {
            let registry = self.handle.folder_registry.read().await;
            let statuses = registry.statuses();
            let roots = statuses
                .iter()
                .map(|status| (status.namespace.clone(), status.path.clone()))
                .collect();
            let watch_configs = statuses
                .iter()
                .map(|status| (status.namespace.clone(), registry.watch_for(&status.namespace)))
                .collect();
            drop(registry);
            (roots, watch_configs)
        };
        let live_folders = self.enabled_subscribe_filters();
        let global_engine = self.filter_engine.read().await.clone();
        for (namespace, paths) in ready {
            let Some(root) = roots.get(&namespace) else {
                continue;
            };
            let namespace_id = namespace
                .parse::<NamespaceId>()
                .map_err(|error| SyncwebError::operation("invalid watched folder namespace", error))?;
            let watch = watch_configs.get(&namespace).cloned().unwrap_or_default();
            let folder_engine = watch.compiled_filter()?;
            let evaluation = WatchEvaluation {
                filter: folder_engine.as_ref().or(global_engine.as_ref()),
                exclude: &watch.exclude,
            };
            for (path, removed) in paths {
                self.process_watch_event(
                    namespace_id,
                    root,
                    &path,
                    removed,
                    live_folders.get(&namespace_id),
                    evaluation,
                )
                .await?;
            }
        }
        self.save_status_report().await?;
        Ok(())
    }

    async fn process_watch_event(
        &self,
        namespace: NamespaceId,
        root: &Path,
        path: &Path,
        removed: bool,
        live_filters: Option<&SubscribeFilters>,
        evaluation: WatchEvaluation<'_>,
    ) -> Result<()> {
        let relative = path.strip_prefix(root).unwrap_or(path);
        if relative.as_os_str().is_empty() {
            return Ok(());
        }
        let size = std::fs::metadata(path).map_or(0, |metadata| metadata.len());
        let accepted = evaluation.filter.is_none_or(|active| {
            active.evaluate_for_folder(&namespace.to_string(), &FilterEntry::new(relative.to_path_buf(), size))
                != FilterAction::Reject
        });
        if !accepted {
            return Ok(());
        }

        let folder = self.folder_manager.get(namespace).await?;
        // A read-only replica has no write key, so importing or deleting here
        // can only fail. Ignore the event instead of looping on the error.
        if !folder.is_writable() {
            tracing::debug!(%namespace, path = %path.display(), "ignoring filesystem change for read-only folder");
            return Ok(());
        }
        let result = if removed || !path.exists() {
            folder
                .delete_entry(relative.as_os_str().as_encoded_bytes())
                .await
                .map(|()| 1_u64)
        } else if path.is_file() {
            let importer = Importer::new(
                self.node.blob_store().clone(),
                self.node.docs_engine().clone(),
                folder.doc().clone(),
                folder.author(),
            )
            .with_root(root)
            .with_ignore_patterns(evaluation.exclude.to_vec());
            importer
                .import_path(path)
                .await
                .map(|entries| u64::try_from(entries.len()).unwrap_or(u64::MAX))
        } else {
            Ok(0)
        };
        match result {
            Ok(entries) if entries > 0 => {
                self.handle
                    .folder_registry
                    .write()
                    .await
                    .record_import(namespace, entries, current_timestamp());
                if let Some(filters) = live_filters {
                    self.start_supervision(namespace, filters).await?;
                }
            }
            Ok(_) => {}
            Err(error) => {
                self.handle
                    .folder_registry
                    .write()
                    .await
                    .record_error(namespace, error.to_string());
                tracing::warn!(%namespace, path = %path.display(), %error, "filesystem change will be retried");
                if !removed && is_recoverable_watch_error(&error) {
                    let mut pending = self.pending_watch_events.lock().map_err(|poisoned| {
                        SyncwebError::operation("daemon watch queue mutex is poisoned", poisoned)
                    })?;
                    let ready_at = Instant::now()
                        .checked_add(self.config.watch_debounce)
                        .unwrap_or_else(Instant::now);
                    let entry = pending.entry(namespace.to_string()).or_insert_with(|| PendingWatch {
                        paths: HashMap::new(),
                        ready_at,
                    });
                    entry.ready_at = ready_at;
                    entry.paths.insert(path.to_path_buf(), false);
                    drop(pending);
                }
            }
        }
        Ok(())
    }

    async fn save_status_report(&self) -> Result<()> {
        let state = self.handle.state.read().await.clone();
        let statuses = self.handle.folder_registry.read().await.statuses();
        let schedule = self.schedule_manager.read().await.clone();
        let schedule_report = schedule.as_ref().map(|manager| {
            let minute = crate::schedule::current_minute();
            let next = manager.next_active_window_start_at(None, minute);
            let next_window_start = next.map(|next_minute| {
                let offset = if next_minute >= minute {
                    next_minute.saturating_sub(minute)
                } else {
                    1_440_u16.saturating_sub(minute).saturating_add(next_minute)
                };
                current_timestamp().saturating_add(u64::from(offset).saturating_mul(60))
            });
            ScheduleStatus {
                in_active_window: manager.is_active(None),
                next_window_start,
            }
        });
        let current_stats = self.stats_db.current_stats().unwrap_or_default();
        let report = DaemonStatusReport {
            pid: state.pid,
            node_id: state.node_id,
            started_at: state.started_at,
            uptime_seconds: current_timestamp().saturating_sub(state.started_at),
            folders: statuses.into_iter().collect(),
            bandwidth: BandwidthSnapshot {
                upload_total: current_stats.total_upload,
                download_total: current_stats.total_download,
                upload_rate: 0,
                download_rate: 0,
            },
            schedule: schedule_report,
            rayon_threads: self.archive_pool.thread_count(),
        };
        self.node_db.save_status(&report)
    }

    async fn load_folders(&self) -> Result<()> {
        let folders = self.folder_manager.list().await?;
        let folder_mounts = self
            .node_db
            .load_folder_mounts()?
            .into_iter()
            .collect::<HashMap<_, _>>();
        let watch_configs: HashMap<String, WatchConfig> = self
            .node_db
            .load_folder_watch_configs()?
            .into_iter()
            .filter_map(|(namespace, encoded)| {
                serde_json::from_str::<WatchConfig>(&encoded)
                    .ok()
                    .map(|watch| (namespace, watch))
            })
            .collect();
        let mut registry = self.handle.folder_registry.write().await;
        for folder in folders {
            let namespace_key = folder.namespace_id().to_string();
            if registry.is_removed(&namespace_key) {
                continue;
            }
            if !registry
                .statuses()
                .iter()
                .any(|status| status.namespace == namespace_key)
            {
                let path = folder_mounts.get(&namespace_key).cloned().unwrap_or_default();
                let watch = watch_configs.get(&namespace_key).cloned().unwrap_or_default();
                registry.add(
                    FolderEntry::new(folder.namespace_id(), path)
                        .with_mode(folder.mode().to_string())
                        .with_watch(watch),
                )?;
            }
        }
        drop(registry);
        Ok(())
    }

    async fn reannounce_folders(&self) {
        let Ok(folders) = self.folder_manager.list().await else {
            return;
        };
        for folder in &folders {
            self.folder_manager.announce_namespace(folder.namespace_id()).await;
        }
    }

    async fn load_subscriptions(&self) -> Result<()> {
        let subscriptions = self.node_db.load_subscriptions_with_sizes()?;
        if subscriptions.is_empty() {
            return Ok(());
        }
        let hashes: Vec<_> = subscriptions.iter().map(|(hash, _)| *hash).collect();
        self.folder_manager.seed_subscriptions(hashes).await;

        let mut registry = self.handle.folder_registry.write().await;
        for (hash, size) in subscriptions {
            let subscription = PublicSubscription::new(hash, None, size);
            registry.add_subscription(subscription);
        }
        drop(registry);
        Ok(())
    }

    async fn reload_if_requested(&self) -> Result<()> {
        if !self
            .handle
            .reload_requested
            .swap(false, std::sync::atomic::Ordering::AcqRel)
        {
            return Ok(());
        }
        let app_config = self.node_db.load_app_config()?;
        let parsed_schedule = ScheduleManager::from_config(&app_config.schedule)?;
        let schedule_manager =
            (app_config.schedule != crate::schedule::ScheduleConfig::default()).then_some(parsed_schedule);
        let filter = load_filter_engine(&self.node_db, &self.config.data_dir)?;
        *self.schedule_manager.write().await = schedule_manager;
        *self.filter_engine.write().await = filter;
        let statuses = self.handle.folder_registry.read().await.statuses();
        for status in statuses {
            if let Ok(namespace) = status.namespace.parse::<NamespaceId>()
                && is_active(namespace)
                && !cancel_session(namespace)
            {
                tracing::warn!(%namespace, "reloaded intent did not accept cancellation");
            }
        }
        tracing::info!("daemon configuration reloaded");
        Ok(())
    }

    async fn receive_sync_trigger(&self) -> Option<Option<String>> {
        self.sync_receiver.lock().await.recv().await
    }

    async fn shutdown_resources(&self) -> Result<()> {
        self.handle.set_status(DaemonStatus::Stopping).await;
        let stopping_state = self.handle.state.read().await.clone();
        self.node_db.save_lifecycle(&stopping_state)?;
        self.save_status_report().await?;

        self.cancel_active_sessions().await;
        self.join_intent_tasks().await?;
        self.abort_listeners().await;
        self.finalize_node_stop().await
    }

    async fn cancel_active_sessions(&self) {
        let namespaces: Vec<_> = self
            .handle
            .folder_registry
            .read()
            .await
            .statuses()
            .into_iter()
            .filter_map(|status| status.namespace.parse::<NamespaceId>().ok())
            .collect();
        for namespace in namespaces {
            if is_active(namespace) && !cancel_session(namespace) {
                tracing::warn!(%namespace, "intent did not accept shutdown cancellation");
            }
        }
    }

    async fn join_intent_tasks(&self) -> Result<()> {
        let tasks = {
            let mut task_map = self
                .intent_tasks
                .lock()
                .map_err(|error| SyncwebError::operation("daemon intent task mutex is poisoned", error))?;
            std::mem::take(&mut *task_map)
        };
        for (namespace, task) in tasks {
            if let Some(handle) = task {
                match tokio::time::timeout(Duration::from_secs(10), handle).await {
                    Ok(Ok(())) => {}
                    Ok(Err(error)) => {
                        tracing::warn!(%namespace, %error, "daemon intent task failed");
                    }
                    Err(_) => {
                        tracing::warn!(%namespace, "daemon intent task did not shut down within 10s timeout");
                    }
                }
            }
        }
        Ok(())
    }

    async fn abort_listeners(&self) {
        let maintenance_handle = self.maintenance_task.lock().await.take();
        if let Some(inner) = maintenance_handle {
            inner.abort();
        }
    }

    async fn finalize_node_stop(&self) -> Result<()> {
        let node_result = tokio::time::timeout(Duration::from_secs(30), self.node.stop()).await;
        self.handle.set_status(DaemonStatus::Stopped).await;
        let remove_result = self.node_db.remove_lifecycle();
        let remove_status_result = self.node_db.remove_status();
        let release_result = self.pid_lock.release();
        match node_result {
            Ok(Ok(())) => {}
            Ok(Err(error)) => tracing::warn!(%error, "node stop reported an error"),
            Err(_) => tracing::warn!("node stop timed out after 30s"),
        }
        remove_result?;
        remove_status_result?;
        release_result?;
        Ok(())
    }

    /// Spawn a background task that periodically vacuums databases with
    /// excessive freelist pages.
    async fn spawn_maintenance_task(&self) {
        let node_db = self.node_db.clone();
        let stats_db = self.stats_db.clone();
        let mut shutdown = self.handle.shutdown_sender.subscribe();
        let handle: tokio::task::JoinHandle<()> = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_hours(24));
            loop {
                tokio::select! {
                    _ = shutdown.recv() => {
                        tracing::info!("maintenance task shutting down");
                        break;
                    }
                    _ = interval.tick() => {
                        let dbs: [&dyn crate::storage::Vacuumable; 2] = [&node_db, &stats_db];
                        for db in dbs {
                            match db.freelist_count() {
                                Ok(count) if count > 100 => {
                                    tracing::info!(freelist = %count, "running vacuum");
                                    if let Err(error) = db.vacuum() {
                                        tracing::warn!(%error, "maintenance vacuum failed");
                                    }
                                }
                                Ok(_) => {}
                                Err(error) => tracing::warn!(%error, "maintenance freelist query failed"),
                            }
                        }
                    }
                }
            }
        });
        *self.maintenance_task.lock().await = Some(handle);
    }

    async fn subscribe_channels(&self) {
        let Some(ref indexing) = self.indexing else {
            tracing::debug!("indexing unavailable; skipping catalog channel subscription");
            return;
        };
        let Some(app_config) = self.load_channel_config() else {
            return;
        };
        let author = match self.node.docs_engine().author().await {
            Ok(a) => a,
            Err(error) => {
                tracing::warn!(%error, "failed to get author for channel subscription");
                return;
            }
        };
        let catalog_service = indexing.catalog_service(self.node.docs_engine(), self.node.blob_store(), author);
        for (channel_name, channel_config) in &app_config.channels {
            self.subscribe_one_channel(indexing, &catalog_service, channel_name, channel_config)
                .await;
        }
    }

    fn load_channel_config(&self) -> Option<crate::storage::config::Config> {
        let config_path = self.config.data_dir.join("config.toml");
        match crate::storage::config::Config::load(&config_path) {
            Ok(c) if !c.channels.is_empty() => Some(c),
            Ok(_) => None,
            Err(error) => {
                tracing::warn!(%error, "failed to load config for channel subscription");
                None
            }
        }
    }

    async fn subscribe_one_channel(
        &self,
        indexing: &IndexingService,
        catalog_service: &crate::indexing::CatalogService,
        channel_name: &str,
        channel_config: &crate::storage::config::ChannelConfig,
    ) {
        let Ok(ticket) = channel_config.ticket.parse::<iroh_docs::DocTicket>() else {
            tracing::warn!(%channel_name, "invalid channel doc ticket; skipping");
            return;
        };
        let catalog = match catalog_service.subscribe(ticket).await {
            Ok(c) => c,
            Err(error) => {
                tracing::warn!(%channel_name, %error, "failed to subscribe to catalog channel");
                return;
            }
        };
        Self::record_channel_namespace(indexing, channel_name, &catalog);
    }

    fn record_channel_namespace(indexing: &IndexingService, channel_name: &str, catalog: &crate::indexing::Catalog) {
        let namespace_id = catalog.namespace_id();
        if let Err(error) = indexing.database().set_channel_namespace(channel_name, namespace_id) {
            tracing::warn!(%channel_name, %error, "failed to persist channel namespace");
        }
        tracing::info!(%channel_name, %namespace_id, "subscribed to catalog channel");
    }

    async fn spawn_membership_listeners(&self) {
        let networks: Vec<_> = {
            let guard = self.network_manager.read().await;
            guard.list().into_iter().cloned().collect::<Vec<_>>()
        };
        let docs_engine = self.node.docs_engine().clone();
        let blob_store = self.node.blob_store().clone();
        let shutdown = self.handle.shutdown_sender.subscribe();
        let local_pk = self.node.endpoint().id();
        let node_db = self.node_db.clone();
        for network in networks {
            let Some(ref doc_ticket_str) = network.doc_ticket else {
                continue;
            };
            let Ok(doc_ticket) = doc_ticket_str.parse::<iroh_docs::DocTicket>() else {
                tracing::warn!(network_id = %network.id, "invalid doc_ticket in network");
                continue;
            };
            let de = docs_engine.clone();
            let bs = blob_store.clone();
            let mut member_shutdown = shutdown.resubscribe();
            let network_id = network.id;
            let local_key_str = local_pk.to_string();
            let db = node_db.clone();
            tokio::spawn(async move {
                let doc = match de.import_ticket(doc_ticket).await {
                    Ok(doc) => doc,
                    Err(error) => {
                        tracing::warn!(%network_id, %error, "failed to import membership doc from ticket");
                        return;
                    }
                };
                if let Err(error) = de.start_sync(&doc, Vec::new()).await {
                    tracing::warn!(%network_id, %error, "failed to start membership doc sync");
                }
                let mut stream = match de.watch(&doc).await {
                    Ok(stream) => stream,
                    Err(error) => {
                        tracing::warn!(%network_id, %error, "failed to watch membership doc");
                        return;
                    }
                };
                loop {
                    tokio::select! {
                        _ = member_shutdown.recv() => {
                            tracing::debug!(%network_id, "membership listener shutting down");
                            break;
                        }
                        event = stream.next() => {
                            let Some(event_result) = event else {
                                break;
                            };
                            match event_result {
                                Ok(iroh_docs::engine::LiveEvent::InsertLocal { entry } | iroh_docs::engine::LiveEvent::InsertRemote { entry, .. }) => {
                                    if entry.key() == NETWORK_MEMBERS_KEY {
                                        let hash = entry.content_hash();
                                        let Ok(content) = bs.get(hash).await else { continue; };
                                        match serde_json::from_slice::<crate::net::membership_doc::SignedMemberList>(&content) {
                                            Ok(member_list) => {
                                                if let Err(error) = member_list.verify() {
                                                    tracing::warn!(%network_id, %error, "invalid membership list signature");
                                                    continue;
                                                }
                                                let is_member = member_list.members.iter().any(|m| m.key == local_key_str);
                                                if is_member {
                                                    tracing::debug!(%network_id, members = member_list.members.len(), "membership updated");
                                                } else {
                                                    tracing::warn!(%network_id, "local node was removed from network");
                                                    let _ = db.delete_network(network_id);
                                                }
                                            }
                                            Err(error) => {
                                                tracing::warn!(%network_id, %error, "failed to parse membership list");
                                            }
                                        }
                                    }
                                }
                                Ok(_) => {}
                                Err(error) => {
                                    tracing::warn!(%network_id, %error, "membership doc event error");
                                }
                            }
                        }
                    }
                }
            });
        }
    }

    async fn handle_signals(&self, shutdown: broadcast::Sender<()>) -> Result<()> {
        #[cfg(unix)]
        {
            let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
            tokio::select! {
                result = tokio::signal::ctrl_c() => result?,
                _ = terminate.recv() => {}
            }
        }
        #[cfg(not(unix))]
        {
            tokio::signal::ctrl_c().await?;
        }
        send_shutdown(&shutdown);
        Ok(())
    }
}

impl Drop for Daemon {
    fn drop(&mut self) {
        if let Err(error) = self.node_db.remove_lifecycle() {
            tracing::warn!(%error, "failed to remove daemon lifecycle");
        }
        if let Err(error) = self.node_db.remove_status() {
            tracing::warn!(%error, "failed to remove daemon status");
        }
    }
}

fn send_shutdown(sender: &broadcast::Sender<()>) {
    match sender.send(()) {
        Ok(_) | Err(broadcast::error::SendError(())) => {}
    }
}

/// Always remove a namespace from a shared busy set, ignoring a poisoned lock.
fn fully_remove(set: &std::sync::Mutex<HashSet<NamespaceId>>, namespace: &NamespaceId) {
    let mut guard = set.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
    guard.remove(namespace);
}

/// Build the daemon's folder reconciler.
///
/// The reconciler is invoked by the sync engine after remote content becomes
/// local for a writable folder. It coalesces the burst of `ContentReady`
/// events from a sync pass into one trailing-edge materialization per folder:
/// mark the folder busy, wait a short settle window, then materialize
/// everything that is local. Later blob completions re-trigger the reconciler
/// and are picked up on the next trailing edge, so nothing is lost to timing.
fn build_reconciler(
    node: Arc<IrohNode>,
    manager: FolderManager,
    db: NodeDatabase,
) -> Arc<dyn Fn(NamespaceId) + Send + Sync> {
    const RECONCILE_SETTLE: Duration = Duration::from_millis(750);
    let busy = Arc::new(std::sync::Mutex::new(HashSet::<NamespaceId>::new()));
    Arc::new(move |namespace| {
        let mut busy_guard = match busy.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        if !busy_guard.insert(namespace) {
            return;
        }
        drop(busy_guard);
        tracing::debug!(%namespace, "folder reconciler triggered");
        let node_handle = node.clone();
        let manager_handle = manager.clone();
        let db_handle = db.clone();
        let busy_set = busy.clone();
        tokio::spawn(async move {
            tokio::time::sleep(RECONCILE_SETTLE).await;
            let mounts = match db_handle.load_folder_mounts() {
                Ok(mounts) => mounts,
                Err(error) => {
                    tracing::warn!(%namespace, %error, "reconciler could not read folder mounts");
                    fully_remove(&busy_set, &namespace);
                    return;
                }
            };
            let Some((_, mount_root)) = mounts
                .into_iter()
                .find(|(mounted, _)| mounted.parse::<NamespaceId>().is_ok_and(|id| id == namespace))
            else {
                fully_remove(&busy_set, &namespace);
                return;
            };
            let folder = match manager_handle.get(namespace).await {
                Ok(folder) => folder,
                Err(error) => {
                    tracing::warn!(%namespace, %error, "reconciler could not open folder");
                    fully_remove(&busy_set, &namespace);
                    return;
                }
            };
            // Read-only replicas materialize on explicit download only; a
            // subscribe replica must not auto-write the mount.
            if !folder.is_writable() {
                fully_remove(&busy_set, &namespace);
                return;
            }
            match crate::sync::materialize_folder_content(&node_handle, &manager_handle, namespace, &mount_root).await {
                Ok(outcome) => {
                    if outcome.summary.materialized > 0 || outcome.summary.conflicts > 0 {
                        tracing::info!(
                            %namespace,
                            materialized = outcome.summary.materialized,
                            conflicts = outcome.summary.conflicts,
                            pending = outcome.summary.pending,
                            "reconciled folder after live sync"
                        );
                    }
                    // Pending blobs may finish shortly after this pass; nudge
                    // once more so the next trailing edge picks them up.
                    if outcome.summary.pending > 0 {
                        tokio::time::sleep(RECONCILE_SETTLE).await;
                        let _ = crate::sync::materialize_folder_content(
                            &node_handle,
                            &manager_handle,
                            namespace,
                            &mount_root,
                        )
                        .await;
                    }
                }
                Err(error) => {
                    tracing::warn!(%namespace, %error, "reconciler failed to materialize folder");
                }
            }
            fully_remove(&busy_set, &namespace);
        });
    })
}

fn is_recoverable_watch_error(error: &SyncwebError) -> bool {
    let message = error.to_string();
    message.contains("file changed during import") || message.contains("input path does not exist")
}