varde/varde-daemon/src/transfer.rs

//! iroh endpoint, blob provider and downloader.
//!
//! This module is the transport seam: trust gating and rate limiting
//! live in [`crate::shaped`]'s provider event handler wired up here, and
//! a scavenger congestion controller would later slot in at the same
//! place without touching the daemon logic above.

use std::path::Path;
use std::sync::Arc;

use anyhow::{Context, Result};
use iroh::address_lookup::memory::MemoryLookup;
use iroh::protocol::Router;
use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey};
use iroh_blobs::api::downloader::{DownloadOptions, Downloader, Shuffled, SplitStrategy};
use iroh_blobs::protocol::GetRequest;
use iroh_blobs::{BlobFormat, BlobsProtocol, Hash, HashAndFormat};
use iroh_mdns_address_lookup::{DiscoveryEvent, MdnsAddressLookup};
use n0_future::StreamExt;
use tokio::sync::{broadcast, mpsc};
use tracing::{debug, info, warn};
use varde_proto::{Direction, Event};

use crate::config::Config;
use crate::meta::Meta;
use crate::metered::MeteredState;
use crate::shaped::{Gate, ReceivedPush, ServeFilter, TokenBucket};
use crate::store::BlobStore;

/// The network side of the daemon: one iroh endpoint serving the blob
/// store (trust-gated, rate-limited), plus a downloader for fetching
/// pinned content from providers.
#[derive(Debug)]
pub struct Transfer {
    endpoint: Endpoint,
    router: Router,
    store: iroh_blobs::api::Store,
    downloader: Downloader,
    /// Address book for provider addresses learned out-of-band (ticket
    /// imports): the downloader dials by endpoint id and resolves
    /// addresses through the endpoint's address lookup services.
    known_addrs: MemoryLookup,
    events: broadcast::Sender<Event>,
    metered: MeteredState,
    discovery_rx: std::sync::Mutex<Option<mpsc::Receiver<EndpointId>>>,
    pushes_rx: std::sync::Mutex<Option<mpsc::Receiver<ReceivedPush>>>,
}

impl Transfer {
    /// Bind the endpoint and start serving the store.
    ///
    /// With `wan_upload` off (the default) the relay is disabled: the
    /// endpoint is only reachable via direct (LAN/localhost) paths and
    /// never uploads through third-party infrastructure.
    pub async fn start(
        config: &Config,
        store: &BlobStore,
        meta: Arc<Meta>,
        open_filter: ServeFilter,
        events: broadcast::Sender<Event>,
        metered: MeteredState,
    ) -> Result<Transfer> {
        let secret = load_or_create_secret(&config.store_dir.join("secret.key"))?;
        // With wan_upload the n0 defaults apply (their relays and DNS
        // lookup, matching the pre-1.0 default relay mode); otherwise the
        // endpoint gets no external services at all.
        let builder = if config.wan_upload {
            Endpoint::builder(iroh::endpoint::presets::N0)
        } else {
            Endpoint::builder(iroh::endpoint::presets::Minimal).relay_mode(RelayMode::Disabled)
        };
        let endpoint = builder
            .secret_key(secret.clone())
            .bind()
            .await
            .context("binding iroh endpoint")?;
        info!(endpoint_id = %endpoint.id(), discovery = config.discovery, "endpoint up");

        let known_addrs = MemoryLookup::with_provenance("varde-ticket");
        endpoint
            .address_lookup()
            .context("endpoint address lookup services")?
            .add(known_addrs.clone());

        // LAN discovery: advertise under iroh's mDNS-style local
        // discovery service (identifiable `_iroh` mdns records, not
        // anonymous noise). Discovered endpoint ids stream to the daemon,
        // which decides what to do based on trust.
        let (discovery_tx, discovery_rx) = mpsc::channel(64);
        if config.discovery {
            let mdns = MdnsAddressLookup::builder()
                .build(endpoint.id())
                .context("starting mdns local discovery")?;
            let mut sightings = mdns.subscribe().await;
            tokio::spawn(async move {
                while let Some(event) = sightings.next().await {
                    let DiscoveryEvent::Discovered { endpoint_info, .. } = event else {
                        continue;
                    };
                    if discovery_tx.send(endpoint_info.endpoint_id).await.is_err() {
                        break;
                    }
                }
            });
            endpoint
                .address_lookup()
                .context("endpoint address lookup services")?
                .add(mdns);
        }

        // Best-effort low-priority marking of our UDP traffic.
        crate::dscp::mark_endpoint_sockets(&endpoint.bound_sockets());

        let (pushes_tx, pushes_rx) = mpsc::channel(16);
        let provider_events = crate::shaped::provider_events(Gate {
            meta,
            open_filter,
            upload: TokenBucket::new(config.max_upload_bytes_per_sec),
            serving_enabled: config.max_upload_bytes_per_sec > 0,
            metered: metered.clone(),
            events: events.clone(),
            pushes: pushes_tx,
        });
        let provider = BlobsProtocol::new(store.api(), Some(provider_events));

        let downloader = store.api().downloader(&endpoint);
        let router = Router::builder(endpoint.clone())
            .accept(iroh_blobs::ALPN, provider)
            .spawn();

        Ok(Transfer {
            endpoint,
            router,
            store: store.api().clone(),
            downloader,
            known_addrs,
            events,
            metered,
            discovery_rx: std::sync::Mutex::new(Some(discovery_rx)),
            pushes_rx: std::sync::Mutex::new(Some(pushes_rx)),
        })
    }

    /// Our stable endpoint id.
    pub fn node_id(&self) -> String {
        self.endpoint.id().to_string()
    }

    /// The endpoint's identity key (also signs our announcements).
    pub fn secret_key(&self) -> SecretKey {
        self.endpoint.secret_key().clone()
    }

    /// Take the stream of locally discovered endpoint ids. Yields each
    /// sighting (mdns re-announces periodically); the daemon dedupes.
    /// Callable once; returns None afterwards or with discovery off.
    pub fn take_discovery(&self) -> Option<mpsc::Receiver<EndpointId>> {
        self.discovery_rx.lock().expect("discovery lock").take()
    }

    /// Take the stream of completed inbound pushes. Callable once.
    pub fn take_pushes(&self) -> Option<mpsc::Receiver<ReceivedPush>> {
        self.pushes_rx.lock().expect("pushes lock").take()
    }

    /// Push fully-present `content` to `peer`, unprompted. The peer
    /// accepts only if it trusts us. Returns payload bytes written.
    pub async fn push_to(&self, peer: EndpointId, content: HashAndFormat) -> Result<u64> {
        anyhow::ensure!(
            !self.metered.is_metered(),
            "metered connection: refusing to push"
        );
        let conn = self
            .endpoint
            .connect(peer, iroh_blobs::ALPN)
            .await
            .with_context(|| format!("connecting to {peer}"))?;
        let request = match content.format {
            BlobFormat::Raw => {
                iroh_blobs::protocol::PushRequest::from(GetRequest::blob(content.hash))
            }
            BlobFormat::HashSeq => {
                iroh_blobs::protocol::PushRequest::from(GetRequest::all(content.hash))
            }
        };
        // Upstream returns zeroed Stats for pushes; the byte count rides
        // the progress stream instead.
        use iroh_blobs::api::remote::PushProgressItem;
        let mut stream = self
            .store
            .remote()
            .execute_push(conn.clone(), request)
            .stream();
        let mut bytes = 0u64;
        let mut done = false;
        while let Some(item) = stream.next().await {
            match item {
                PushProgressItem::Progress(sent) => {
                    bytes = sent;
                    let _ = self.events.send(Event::TransferProgress {
                        hash: content.hash.to_hex().to_string(),
                        direction: Direction::Upload,
                        bytes_done: sent,
                        bytes_total: None,
                    });
                }
                PushProgressItem::Done(_) => {
                    done = true;
                    break;
                }
                PushProgressItem::Error(e) => {
                    return Err(anyhow::anyhow!(e)).context("pushing content");
                }
            }
        }
        anyhow::ensure!(done, "push stream ended without a result");

        // Writing is fire-and-forget at the QUIC level: closing the
        // connection now could discard data the peer has not read yet,
        // and says nothing about acceptance. Delivery is confirmed by
        // observing the peer's bitfields until they report complete —
        // the peer imports the root first, then children in order, so
        // root + last child covers the whole request.
        self.wait_remote_complete(&conn, content.hash).await?;
        if content.format == BlobFormat::HashSeq {
            let seq_bytes = self
                .store
                .blobs()
                .get_bytes(content.hash)
                .await
                .context("reading pushed hashseq")?;
            let seq = iroh_blobs::hashseq::HashSeq::try_from(seq_bytes)
                .map_err(|e| anyhow::anyhow!("invalid hashseq: {e}"))?;
            if let Some(last) = seq.iter().last() {
                self.wait_remote_complete(&conn, last).await?;
            }
        }
        info!(hash = %content.hash.to_hex(), peer = %peer, bytes, "push complete, verified by peer");
        Ok(bytes)
    }

    /// Watch `hash` on the remote end of `conn` until its bitfield
    /// reports the blob complete.
    async fn wait_remote_complete(
        &self,
        conn: &iroh::endpoint::Connection,
        hash: Hash,
    ) -> Result<()> {
        let observe = self.store.remote().observe(
            conn.clone(),
            iroh_blobs::protocol::ObserveRequest::new(hash),
        );
        let mut observe = std::pin::pin!(observe);
        while let Some(bitfield) = observe.next().await {
            let bitfield = bitfield.context("observing pushed content on the peer")?;
            if bitfield.is_complete() {
                return Ok(());
            }
        }
        anyhow::bail!("peer stopped reporting before the pushed content completed");
    }

    /// Produce a ticket for out-of-band sharing of `content`.
    pub async fn export_ticket(&self, content: HashAndFormat) -> Result<String> {
        let addr = self.endpoint.addr();
        anyhow::ensure!(
            !addr.addrs.is_empty(),
            "endpoint has no dialable addresses yet"
        );
        let ticket = iroh_blobs::ticket::BlobTicket::new(addr, content.hash, content.format);
        Ok(ticket.to_string())
    }

    /// Queue a background fetch of `content` from `providers`. Emits
    /// progress events and [`Event::PinComplete`] when fully verified.
    pub fn spawn_fetch(&self, content: HashAndFormat, providers: Vec<EndpointAddr>) {
        if self.metered.is_metered() {
            // The pin stays recorded; auto-sync retries once unmetered.
            info!(hash = %content.hash.to_hex(), "metered connection: fetch deferred");
            return;
        }
        // The downloader dials by EndpointId; teach the endpoint any
        // direct addresses we know (ticket-embedded ones). Id-only
        // provider entries resolve via discovery instead.
        let mut ids = Vec::with_capacity(providers.len());
        for provider in providers {
            ids.push(provider.id);
            if !provider.addrs.is_empty() {
                self.known_addrs.add_endpoint_info(provider);
            }
        }

        // With several providers one fetch stripes across all of them;
        // a lone provider gets the whole request.
        let strategy = if ids.len() > 1 {
            SplitStrategy::Split
        } else {
            SplitStrategy::None
        };
        let downloader = self.downloader.clone();
        let events = self.events.clone();
        tokio::spawn(async move {
            let progress = downloader.download_with_opts(DownloadOptions::new(
                content,
                Shuffled::new(ids),
                strategy,
            ));
            let mut stream = match progress.stream().await {
                Ok(stream) => stream,
                Err(e) => {
                    warn!(hash = %content.hash.to_hex(), error = %e, "fetch failed to start");
                    return;
                }
            };
            use iroh_blobs::api::downloader::DownloadProgressItem;
            let mut failed = false;
            while let Some(item) = stream.next().await {
                match item {
                    DownloadProgressItem::Progress(bytes_done) => {
                        let _ = events.send(Event::TransferProgress {
                            hash: content.hash.to_hex().to_string(),
                            direction: Direction::Download,
                            bytes_done,
                            bytes_total: None,
                        });
                    }
                    DownloadProgressItem::ProviderFailed { id, .. } => {
                        debug!(hash = %content.hash.to_hex(), provider = %id, "provider failed");
                    }
                    DownloadProgressItem::Error(e) => {
                        warn!(hash = %content.hash.to_hex(), error = %e, "fetch failed");
                        failed = true;
                    }
                    DownloadProgressItem::DownloadError => {
                        warn!(hash = %content.hash.to_hex(), "fetch failed");
                        failed = true;
                    }
                    _ => {}
                }
            }
            if !failed {
                info!(hash = %content.hash.to_hex(), "fetch complete");
                let _ = events.send(Event::PinComplete {
                    hash: content.hash.to_hex().to_string(),
                });
            }
        });
    }

    /// Gracefully close the endpoint.
    pub async fn shutdown(&self) {
        if let Err(e) = self.router.shutdown().await {
            warn!(error = %e, "router shutdown");
        }
        self.endpoint.close().await;
    }
}

/// Parse a ticket string into (root, format, provider address).
pub fn parse_ticket(ticket: &str) -> Result<(Hash, BlobFormat, EndpointAddr), String> {
    let ticket: iroh_blobs::ticket::BlobTicket =
        ticket.parse().map_err(|e| format!("invalid ticket: {e}"))?;
    let (addr, hash, format) = ticket.into_parts();
    Ok((hash, format, addr))
}

/// Load the endpoint secret key from `path`, creating it (mode 0600) on
/// first start.
fn load_or_create_secret(path: &Path) -> Result<SecretKey> {
    match std::fs::read_to_string(path) {
        Ok(hex) => parse_secret(hex.trim())
            .with_context(|| format!("parsing secret key {}", path.display())),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            let secret = SecretKey::generate();
            let hex = hex_encode(&secret.to_bytes());
            use std::io::Write;
            use std::os::unix::fs::OpenOptionsExt;
            let mut file = std::fs::OpenOptions::new()
                .write(true)
                .create_new(true)
                .mode(0o600)
                .open(path)
                .with_context(|| format!("creating secret key {}", path.display()))?;
            file.write_all(hex.as_bytes())?;
            Ok(secret)
        }
        Err(e) => Err(e).with_context(|| format!("reading secret key {}", path.display())),
    }
}

fn parse_secret(hex: &str) -> Result<SecretKey> {
    anyhow::ensure!(hex.len() == 64, "expected 64 hex chars, got {}", hex.len());
    let mut bytes = [0u8; 32];
    for (i, byte) in bytes.iter_mut().enumerate() {
        *byte = u8::from_str_radix(&hex[2 * i..2 * i + 2], 16)
            .map_err(|e| anyhow::anyhow!("bad hex at {}: {e}", 2 * i))?;
    }
    Ok(SecretKey::from_bytes(&bytes))
}

fn hex_encode(bytes: &[u8]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

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

    #[test]
    fn secret_key_round_trips_through_file() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("secret.key");
        let first = load_or_create_secret(&path).unwrap();
        let second = load_or_create_secret(&path).unwrap();
        assert_eq!(first.to_bytes(), second.to_bytes());

        use std::os::unix::fs::PermissionsExt;
        let mode = std::fs::metadata(&path).unwrap().permissions().mode();
        assert_eq!(mode & 0o777, 0o600, "secret key must be 0600");
    }
}