varde/varde-daemon/src/daemon.rs

//! Request dispatch and daemon state.

use std::path::PathBuf;
use std::str::FromStr;

use anyhow::Context;
use iroh_blobs::{BlobFormat, Hash, HashAndFormat};
use tokio::sync::broadcast;
use varde_proto::{
    Envelope, ErrorCode, Event, GlobalStatus, HashStatus, PinInfo, Request, ResponseData,
};

use crate::config::Config;
use crate::meta::{Meta, StoredFormat};
use crate::store::{BlobStore, OpError};
use crate::transfer::Transfer;

/// Shared daemon state. Handlers grow with the milestones; anything not
/// yet wired returns [`ErrorCode::Unimplemented`] rather than pretending.
pub struct Daemon {
    config: Config,
    store: BlobStore,
    meta: Meta,
    transfer: Transfer,
    events: broadcast::Sender<Event>,
}

impl Daemon {
    /// Open the blob store and metadata under the configured store dir,
    /// and bring up the iroh endpoint.
    pub async fn open(config: Config) -> anyhow::Result<Daemon> {
        std::fs::create_dir_all(&config.store_dir)
            .with_context(|| format!("creating store directory {}", config.store_dir.display()))?;
        let store = BlobStore::open(&config.store_dir).await?;
        let meta = Meta::load(&config.store_dir)?;
        // Capacity bounds memory if a subscriber stalls; laggards get a
        // Lagged error, not unbounded buffering.
        let (events, _) = broadcast::channel(1024);
        let transfer =
            Transfer::start(&config, &store, store.pool_handle(), events.clone()).await?;
        Ok(Daemon {
            config,
            store,
            meta,
            transfer,
            events,
        })
    }

    /// Flush the store and close the endpoint.
    pub async fn shutdown(&self) {
        self.transfer.shutdown().await;
        self.store.shutdown().await;
    }

    pub fn config(&self) -> &Config {
        &self.config
    }

    pub fn store(&self) -> &BlobStore {
        &self.store
    }

    pub fn meta(&self) -> &Meta {
        &self.meta
    }

    /// New receiver for the daemon event stream.
    pub fn subscribe_events(&self) -> broadcast::Receiver<Event> {
        self.events.subscribe()
    }

    /// Handle one API request. Infallible at this level: all failures are
    /// mapped to structured error envelopes.
    pub async fn handle(&self, request: Request) -> Envelope {
        match self.dispatch(request).await {
            Ok(data) => Envelope::ok(data),
            Err(e) => Envelope::err(error_code(&e), e.to_string()),
        }
    }

    async fn dispatch(&self, request: Request) -> Result<ResponseData, OpError> {
        match request {
            Request::Add { path, recursive } => {
                let (hash, bytes, format) =
                    self.store.add_path(&PathBuf::from(path), recursive).await?;
                self.meta.record_format(&hash.to_hex(), format)?;
                Ok(ResponseData::Added {
                    hash: hash.to_hex().to_string(),
                    bytes,
                })
            }
            Request::Pin { hash, policy } => {
                let parsed = parse_hash(&hash)?;
                self.meta.pin(&parsed.to_hex(), policy)?;
                // Fetching absent pinned content arrives with the
                // transfer milestone.
                Ok(ResponseData::Pinned {
                    hash: parsed.to_hex().to_string(),
                })
            }
            Request::Unpin { hash } => {
                let parsed = parse_hash(&hash)?;
                if !self.meta.unpin(&parsed.to_hex())? {
                    return Err(OpError::NotFound(format!("no pin for {}", parsed.to_hex())));
                }
                Ok(ResponseData::Done {})
            }
            Request::Materialize { hash, dest, mode } => {
                let parsed = parse_hash(&hash)?;
                let format = self.meta.format_of(&parsed.to_hex());
                let (bytes, reflinked) = self
                    .store
                    .materialize(parsed, format, &PathBuf::from(dest), mode)
                    .await?;
                Ok(ResponseData::Materialized { bytes, reflinked })
            }
            Request::Status { hash: None } => Ok(ResponseData::Status(GlobalStatus {
                version: env!("CARGO_PKG_VERSION").to_string(),
                protocol: varde_proto::PROTOCOL_VERSION,
                node_id: Some(self.transfer.node_id()),
                store_path: self.config.store_dir.display().to_string(),
                pins: self.meta.pins().len() as u64,
                connected_peers: 0,
            })),
            Request::Status { hash: Some(hash) } => {
                let parsed = parse_hash(&hash)?;
                let (have_bytes, total_bytes, complete) = self.store.presence(parsed).await?;
                Ok(ResponseData::HashStatus(HashStatus {
                    hash: parsed.to_hex().to_string(),
                    have_bytes,
                    total_bytes,
                    complete,
                    pinned: self.meta.is_pinned(&parsed.to_hex()),
                    active_peers: Vec::new(),
                }))
            }
            Request::List {} => {
                let mut pins = Vec::new();
                for (hash, record) in self.meta.pins() {
                    let parsed = parse_hash(&hash)?;
                    let (have_bytes, total_bytes, complete) = self.store.presence(parsed).await?;
                    pins.push(PinInfo {
                        hash,
                        policy: record.policy,
                        complete,
                        have_bytes,
                        total_bytes,
                    });
                }
                Ok(ResponseData::Pins { pins })
            }
            Request::Gc {} => {
                let roots = self.pin_roots()?;
                let blobs_removed = self.store.gc(roots).await?;
                Ok(ResponseData::GcDone { blobs_removed })
            }
            Request::TicketExport { hash } => {
                let parsed = parse_hash(&hash)?;
                let (_have, _total, complete) = self.store.presence(parsed).await?;
                if !complete {
                    return Err(OpError::NotFound(format!(
                        "{} is not fully present; cannot export a ticket",
                        parsed.to_hex()
                    )));
                }
                let content = HashAndFormat {
                    hash: parsed,
                    format: blob_format(self.meta.format_of(&parsed.to_hex())),
                };
                let ticket = self.transfer.export_ticket(content).await?;
                Ok(ResponseData::Ticket { ticket })
            }
            Request::TicketImport { ticket, pin_policy } => {
                let (hash, format, provider) =
                    crate::transfer::parse_ticket(&ticket).map_err(OpError::InvalidArgument)?;
                let stored = match format {
                    BlobFormat::Raw => StoredFormat::Raw,
                    BlobFormat::HashSeq => StoredFormat::HashSeq,
                };
                self.meta.record_format(&hash.to_hex(), stored)?;
                // Pin first: the pin is the GC root that keeps the fetch's
                // data alive, and it is the standing intent even if the
                // provider is unreachable right now.
                self.meta.pin(&hash.to_hex(), pin_policy)?;
                self.transfer
                    .spawn_fetch(HashAndFormat { hash, format }, vec![provider]);
                Ok(ResponseData::Pinned {
                    hash: hash.to_hex().to_string(),
                })
            }
            Request::Subscribe {} => Ok(ResponseData::Done {}),
            other => Err(OpError::Unimplemented(format!(
                "not implemented yet: {other:?}"
            ))),
        }
    }

    /// The GC roots: every pinned hash with its recorded format.
    fn pin_roots(&self) -> Result<Vec<HashAndFormat>, OpError> {
        let mut roots = Vec::new();
        for (hash, _record) in self.meta.pins() {
            let parsed = parse_hash(&hash)?;
            roots.push(HashAndFormat {
                hash: parsed,
                format: blob_format(self.meta.format_of(&hash)),
            });
        }
        Ok(roots)
    }
}

fn blob_format(format: StoredFormat) -> BlobFormat {
    match format {
        StoredFormat::Raw => BlobFormat::Raw,
        StoredFormat::HashSeq => BlobFormat::HashSeq,
    }
}

fn parse_hash(s: &str) -> Result<Hash, OpError> {
    Hash::from_str(s).map_err(|e| OpError::InvalidArgument(format!("invalid hash {s:?}: {e}")))
}

fn error_code(e: &OpError) -> ErrorCode {
    match e {
        OpError::NotFound(_) => ErrorCode::NotFound,
        OpError::InvalidArgument(_) => ErrorCode::InvalidArgument,
        OpError::Io(_) => ErrorCode::Io,
        OpError::Internal(_) => ErrorCode::Internal,
        OpError::Unimplemented(_) => ErrorCode::Unimplemented,
    }
}