sendmer 0.5.1

//! 发送端功能：将本地文件/目录导入 Blob 存储并通过 iroh 协议对外提供。
//!
//! 主要导出 `start_share`，它会导入数据、启动路由器并返回用于后续管理的 `SendResult`。

use crate::core::endpoint::base_endpoint_builder;
use crate::core::events::AppHandle;
use crate::core::options::{AddrInfoOptions, SendOptions, apply_options};
use crate::core::progress::{SenderProgressReporter, SenderTransferStatus, TransferId};
use crate::core::results::SendResult;
use crate::core::storage::{load_fs_store, unique_temp_dir};
use anyhow::Context;
use iroh::{Endpoint, address_lookup::PkarrPublisher};
use iroh_blobs::{
    BlobFormat, BlobsProtocol,
    api::{
        Store, TempTag,
        blobs::{AddPathOptions, ImportMode},
    },
    format::collection::Collection,
    provider::events::{ConnectMode, EventMask, EventSender, RequestMode},
    store::fs::FsStore,
    ticket::BlobTicket,
};
use n0_future::StreamExt;
use n0_future::{BufferedStreamExt, task::AbortOnDropHandle};
use std::{
    path::{Component, Path, PathBuf},
    time::Duration,
};
use tokio::{
    select,
    sync::{Semaphore, mpsc, watch},
};
use tracing::{info, trace, warn};
use walkdir::WalkDir;

const PROVIDER_PROGRESS_TASK_LIMIT: usize = 32;
const ENDPOINT_ONLINE_WAIT_TIMEOUT: Duration = Duration::from_secs(10);

/// Prepare endpoint with the given options
async fn prepare_endpoint(options: &SendOptions) -> anyhow::Result<Endpoint> {
    let mut builder = base_endpoint_builder(options, vec![iroh_blobs::protocol::ALPN.to_vec()])?;

    if options.ticket_type == AddrInfoOptions::Id {
        builder = builder.address_lookup(PkarrPublisher::n0_dns());
    }

    builder.bind().await.map_err(Into::into)
}

/// Prepare temporary directory for blob storage
fn prepare_temp_directory() -> anyhow::Result<PathBuf> {
    unique_temp_dir(".sendmer-send-")
}

/// Validate the path to be shared
fn validate_share_path(path: &Path) -> anyhow::Result<()> {
    let canonical_cwd = std::env::current_dir()?.canonicalize()?;
    let canonical_path = path.canonicalize().with_context(|| {
        format!(
            "share path {} does not exist or cannot be accessed",
            path.display()
        )
    })?;

    if canonical_path == canonical_cwd {
        anyhow::bail!("can not share from the current directory");
    }

    anyhow::ensure!(
        canonical_path.is_file() || canonical_path.is_dir(),
        "share path {} is not a file or directory",
        path.display()
    );

    Ok(())
}

/// Set up data sharing and return immediately after the router starts.
///
/// The caller owns the returned setup before it waits for endpoint readiness, so a
/// later cancellation can use the normal graceful shutdown path.
async fn setup_data_sharing(
    endpoint: Endpoint,
    blobs_data_dir: PathBuf,
    share_request: ShareRequest,
) -> anyhow::Result<SharingSetup> {
    let cleanup_dir = blobs_data_dir.clone();
    let (progress_tx, progress_rx) = mpsc::channel(32);
    let (transfer_status_tx, transfer_status_rx) = watch::channel(SenderTransferStatus::Idle);

    let setup_future = async move {
        let store = load_fs_store(&blobs_data_dir).await?;

        let blobs = BlobsProtocol::new(&store, Some(create_event_sender(progress_tx)));

        let imported = import(share_request.path, blobs.store()).await?;
        let size = imported.size;
        let progress_handle = spawn_provider_progress_task(
            progress_rx,
            share_request.app_handle,
            size,
            share_request.entry_type,
            transfer_status_tx,
        );

        let router = iroh::protocol::Router::builder(endpoint)
            .accept(iroh_blobs::protocol::ALPN, blobs.clone())
            .spawn();

        anyhow::Ok(SharingSetup {
            router,
            imported,
            blobs_data_dir,
            store,
            progress_handle,
            transfer_status_rx,
        })
    };

    match setup_future.await {
        Ok(setup) => Ok(setup),
        Err(error) => Err(finalize_failed_sender_setup(
            error,
            remove_temp_sender_dir(&cleanup_dir).await,
        )),
    }
}

/// Remove a sender-owned temporary store after setup fails, tolerating a store
/// that was never created.
async fn remove_temp_sender_dir(path: &Path) -> anyhow::Result<()> {
    match tokio::fs::remove_dir_all(path).await {
        Ok(()) => Ok(()),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error.into()),
    }
}

/// Preserve the startup error while recording a best-effort cleanup failure.
fn finalize_failed_sender_setup(
    primary_error: anyhow::Error,
    cleanup_result: anyhow::Result<()>,
) -> anyhow::Error {
    if let Err(error) = cleanup_result {
        warn!(error = %error, "failed to clean sender temporary data after setup error");
    }
    primary_error
}

struct ShareRequest {
    path: PathBuf,
    entry_type: crate::core::types::EntryType,
    app_handle: AppHandle,
}

struct SharePlan {
    entry_type: crate::core::types::EntryType,
    wait_for_online: bool,
    blobs_data_dir: PathBuf,
    ticket_type: AddrInfoOptions,
}

struct ImportedSource {
    name: String,
    path: PathBuf,
}

struct ImportedBlob {
    name: String,
    temp_tag: TempTag,
    size: u64,
}

fn create_event_sender(
    progress_tx: mpsc::Sender<iroh_blobs::provider::events::ProviderMessage>,
) -> EventSender {
    EventSender::new(
        progress_tx,
        EventMask {
            connected: ConnectMode::Notify,
            get: RequestMode::NotifyLog,
            ..EventMask::DEFAULT
        },
    )
}

fn spawn_provider_progress_task(
    progress_rx: mpsc::Receiver<iroh_blobs::provider::events::ProviderMessage>,
    app_handle: AppHandle,
    total_file_size: u64,
    entry_type: crate::core::types::EntryType,
    transfer_status_tx: watch::Sender<SenderTransferStatus>,
) -> AbortOnDropHandle<anyhow::Result<()>> {
    AbortOnDropHandle::new(tokio::spawn(show_provide_progress_with_provider_tracker(
        progress_rx,
        app_handle,
        total_file_size,
        entry_type,
        transfer_status_tx,
    )))
}

async fn wait_until_endpoint_is_online(
    endpoint: &iroh::Endpoint,
    wait_for_online: bool,
) -> anyhow::Result<()> {
    if wait_for_online {
        match tokio::time::timeout(ENDPOINT_ONLINE_WAIT_TIMEOUT, endpoint.online()).await {
            Ok(()) => {}
            Err(error) => {
                warn!(
                    error = %error,
                    timeout_secs = ENDPOINT_ONLINE_WAIT_TIMEOUT.as_secs(),
                    "endpoint online probe timed out; continuing with available addresses"
                );
            }
        }
    }
    Ok(())
}

struct SharingSetup {
    router: iroh::protocol::Router,
    imported: ImportedCollection,
    blobs_data_dir: PathBuf,
    store: FsStore,
    progress_handle: AbortOnDropHandle<anyhow::Result<()>>,
    transfer_status_rx: watch::Receiver<SenderTransferStatus>,
}

struct ImportedCollection {
    temp_tag: TempTag,
    size: u64,
    _collection: Collection,
}

impl SharePlan {
    fn new(path: &Path, options: &SendOptions) -> anyhow::Result<Self> {
        Ok(Self {
            entry_type: detect_entry_type(path),
            wait_for_online: !matches!(
                options.relay_mode,
                crate::core::options::RelayModeOption::Disabled
            ),
            blobs_data_dir: prepare_temp_directory()?,
            ticket_type: options.ticket_type,
        })
    }

    fn build_request(&self, path: PathBuf, app_handle: AppHandle) -> ShareRequest {
        ShareRequest {
            path,
            entry_type: self.entry_type,
            app_handle,
        }
    }
}

impl SharingSetup {
    fn into_send_result(
        self,
        entry_type: crate::core::types::EntryType,
        ticket_type: AddrInfoOptions,
    ) -> SendResult {
        let Self {
            router,
            imported,
            blobs_data_dir,
            store,
            progress_handle,
            transfer_status_rx,
        } = self;
        let ImportedCollection { temp_tag, size, .. } = imported;
        let hash = temp_tag.hash();

        let mut addr = router.endpoint().addr();
        apply_options(&mut addr, ticket_type);

        let ticket = BlobTicket::new(addr, hash, BlobFormat::HashSeq);

        SendResult {
            ticket,
            hash,
            size,
            entry_type,
            router,
            temp_tag,
            blobs_data_dir,
            _progress_handle: progress_handle,
            _store: store,
            transfer_status_rx,
        }
    }
}

/// Shut down a fully initialized share after its final startup step fails or is cancelled.
///
/// A `SharingSetup` already owns a live router and file-backed store, so it must be
/// converted into `SendResult` to reuse the ordered shutdown and directory cleanup logic.
async fn shutdown_started_sender_setup(
    setup: SharingSetup,
    entry_type: crate::core::types::EntryType,
    ticket_type: AddrInfoOptions,
    primary_error: anyhow::Error,
) -> anyhow::Error {
    let cleanup_result = setup
        .into_send_result(entry_type, ticket_type)
        .shutdown()
        .await;
    finalize_failed_sender_setup(primary_error, cleanup_result)
}

/// 开始共享（发送）指定的 `path`（文件或目录）。
///
/// - `path`：要分享的文件或目录路径。
/// - `options`：发送配置（转发模式、ticket 类型等）。
/// - `app_handle`：可选的事件发射器句柄，用于 UI/CLI 上报进度。
///
/// 返回 `SendResult`，其中包含票据、hash、大小以及需要保持存活的 router/store 句柄。
pub async fn send(
    path: PathBuf,
    options: SendOptions,
    app_handle: AppHandle,
) -> anyhow::Result<SendResult> {
    info!(
        path = %path.display(),
        relay_mode = ?options.relay_mode,
        ticket_type = ?options.ticket_type,
        "starting send"
    );
    validate_share_path(&path)?;

    let plan = SharePlan::new(&path, &options)?;
    let endpoint = prepare_endpoint(&options).await?;
    let share_request = plan.build_request(path, app_handle);
    let ctrl_c = tokio::signal::ctrl_c();
    tokio::pin!(ctrl_c);

    // If setup and Ctrl+C become ready together, retain the completed setup so it can
    // be shut down gracefully below instead of dropping a live router.
    let setup_result = select! {
        biased;
        x = setup_data_sharing(
            endpoint,
            plan.blobs_data_dir.clone(),
            share_request
        ) => x,
        _ = &mut ctrl_c => {
            Err(anyhow::anyhow!("Operation cancelled"))
        }
    };
    let setup = match setup_result {
        Ok(setup) => setup,
        Err(error) => {
            return Err(finalize_failed_sender_setup(
                error,
                remove_temp_sender_dir(&plan.blobs_data_dir).await,
            ));
        }
    };

    let online_wait_outcome = {
        let online_wait =
            wait_until_endpoint_is_online(setup.router.endpoint(), plan.wait_for_online);
        tokio::pin!(online_wait);

        select! {
            biased;
            _ = &mut ctrl_c => None,
            result = &mut online_wait => Some(result),
        }
    };

    match online_wait_outcome {
        Some(Ok(())) => {}
        Some(Err(error)) => {
            return Err(shutdown_started_sender_setup(
                setup,
                plan.entry_type,
                plan.ticket_type,
                error,
            )
            .await);
        }
        None => {
            return Err(shutdown_started_sender_setup(
                setup,
                plan.entry_type,
                plan.ticket_type,
                anyhow::anyhow!("Operation cancelled"),
            )
            .await);
        }
    }

    let result = setup.into_send_result(plan.entry_type, plan.ticket_type);

    info!(
        hash = %result.hash,
        size = result.size,
        entry_type = %result.entry_type,
        "send setup complete"
    );
    Ok(result)
}

fn detect_entry_type(path: &Path) -> crate::core::types::EntryType {
    if path.is_file() {
        crate::core::types::EntryType::File
    } else {
        crate::core::types::EntryType::Directory
    }
}

/// 将 `path`（文件或目录）导入到给定的 `Store`，并返回导入后的集合信息。
async fn import(path: PathBuf, db: &Store) -> anyhow::Result<ImportedCollection> {
    let sources = collect_import_sources(path)?;
    let parallelism = import_parallelism(sources.len());
    let imported = import_sources(db, sources, parallelism).await?;
    build_collection_from_imports(db, imported).await
}

fn import_parallelism(source_count: usize) -> usize {
    let available = std::thread::available_parallelism()
        .map(usize::from)
        .unwrap_or(1);
    available.min(source_count.max(1))
}

fn collect_import_sources(path: PathBuf) -> anyhow::Result<Vec<ImportedSource>> {
    let path = path.canonicalize()?;
    anyhow::ensure!(path.exists(), "path {} does not exist", path.display());
    let root = path.parent().context("context get parent")?;

    WalkDir::new(path.clone())
        .into_iter()
        .map(|entry| {
            let entry = entry?;
            if !entry.file_type().is_file() {
                return Ok(None);
            }

            let path = entry.into_path();
            let relative = path.strip_prefix(root)?;
            let name = canonicalized_path_to_string(relative, true)?;
            anyhow::Ok(Some(ImportedSource { name, path }))
        })
        .filter_map(Result::transpose)
        .collect::<anyhow::Result<Vec<_>>>()
}

async fn import_sources(
    db: &Store,
    sources: Vec<ImportedSource>,
    parallelism: usize,
) -> anyhow::Result<Vec<ImportedBlob>> {
    n0_future::stream::iter(sources)
        .map(|source| {
            let db = db.clone();
            async move { import_source(&db, source).await }
        })
        .buffered_unordered(parallelism)
        .collect::<Vec<_>>()
        .await
        .into_iter()
        .collect::<anyhow::Result<Vec<_>>>()
}

async fn import_source(db: &Store, source: ImportedSource) -> anyhow::Result<ImportedBlob> {
    let import = db.add_path_with_opts(AddPathOptions {
        path: source.path,
        mode: ImportMode::TryReference,
        format: BlobFormat::Raw,
    });
    let mut stream = import.stream().await;
    let mut item_size = 0;
    let temp_tag = loop {
        let item = stream
            .next()
            .await
            .context("import stream ended without a tag")?;
        trace!("importing {} {item:?}", source.name);
        match item {
            iroh_blobs::api::blobs::AddProgressItem::Size(size) => {
                item_size = size;
            }
            iroh_blobs::api::blobs::AddProgressItem::CopyProgress(_) => {}
            iroh_blobs::api::blobs::AddProgressItem::CopyDone => {}
            iroh_blobs::api::blobs::AddProgressItem::OutboardProgress(_) => {}
            iroh_blobs::api::blobs::AddProgressItem::Error(cause) => {
                anyhow::bail!("error importing {}: {}", source.name, cause);
            }
            iroh_blobs::api::blobs::AddProgressItem::Done(tt) => {
                break tt;
            }
        }
    };

    Ok(ImportedBlob {
        name: source.name,
        temp_tag,
        size: item_size,
    })
}

async fn build_collection_from_imports(
    db: &Store,
    mut imported: Vec<ImportedBlob>,
) -> anyhow::Result<ImportedCollection> {
    imported.sort_by(|a, b| a.name.cmp(&b.name));
    let size = imported.iter().map(|item| item.size).sum::<u64>();
    let (collection, tags) = imported
        .into_iter()
        .map(|item| ((item.name, item.temp_tag.hash()), item.temp_tag))
        .unzip::<_, _, Collection, Vec<_>>();
    let temp_tag = collection.clone().store(db).await?;
    drop(tags);
    Ok(ImportedCollection {
        temp_tag,
        size,
        _collection: collection,
    })
}

/// 将已经标准化的路径转换为库内部使用的字符串表示，路径分隔使用 `/`。
///
/// - `must_be_relative`：如果为 true，则遇到根目录将返回错误（要求相对路径）。
pub fn canonicalized_path_to_string(
    path: impl AsRef<Path>,
    must_be_relative: bool,
) -> anyhow::Result<String> {
    let mut path_str = String::new();
    let parts = path
        .as_ref()
        .components()
        .filter_map(|c| match c {
            Component::Normal(x) => {
                let c = match x.to_str() {
                    Some(c) => c,
                    None => return Some(Err(anyhow::anyhow!("invalid character in path"))),
                };

                if !c.contains('/') && !c.contains('\\') {
                    Some(Ok(c))
                } else {
                    Some(Err(anyhow::anyhow!("invalid path component {:?}", c)))
                }
            }
            Component::RootDir => {
                if must_be_relative {
                    Some(Err(anyhow::anyhow!("invalid path component {:?}", c)))
                } else {
                    path_str.push('/');
                    None
                }
            }
            _ => Some(Err(anyhow::anyhow!("invalid path component {:?}", c))),
        })
        .collect::<anyhow::Result<Vec<_>>>()?;
    let parts = parts.join("/");
    path_str.push_str(&parts);
    Ok(path_str)
}

/// 从提供者事件流中读取进度信息并使用ProviderProgressTracker进行跟踪。
///
/// 该函数使用ProviderProgressTracker来管理多个并发传输的进度，并根据完成状态发射相应的事件。
async fn show_provide_progress_with_provider_tracker(
    mut recv: mpsc::Receiver<iroh_blobs::provider::events::ProviderMessage>,
    app_handle: AppHandle,
    total_file_size: u64,
    entry_type: crate::core::types::EntryType,
    transfer_status_tx: watch::Sender<SenderTransferStatus>,
) -> anyhow::Result<()> {
    let reporter = SenderProgressReporter::new(app_handle, entry_type, transfer_status_tx);
    let request_task_limit = std::sync::Arc::new(Semaphore::new(PROVIDER_PROGRESS_TASK_LIMIT));

    while let Some(item) = recv.recv().await {
        match item {
            iroh_blobs::provider::events::ProviderMessage::ClientConnectedNotify(_msg) => {}
            iroh_blobs::provider::events::ProviderMessage::ConnectionClosed(_msg) => {}
            iroh_blobs::provider::events::ProviderMessage::GetRequestReceivedNotify(msg) => {
                let transfer_id = TransferId::new(msg.connection_id, msg.request_id);
                reporter
                    .on_request_received(transfer_id, total_file_size)
                    .await;

                let reporter_clone = reporter.clone();
                let mut rx = msg.rx;
                let Ok(permit) = request_task_limit.clone().acquire_owned().await else {
                    break;
                };
                tokio::spawn(async move {
                    let _permit = permit;
                    while let Ok(Some(update)) = rx.recv().await {
                        reporter_clone.on_request_update(transfer_id, update).await;
                    }
                });
            }
            _ => {
                // Handle other message types that we don't need to track
            }
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{
        ShareRequest, canonicalized_path_to_string, collect_import_sources, detect_entry_type,
        import_parallelism, prepare_endpoint, setup_data_sharing, shutdown_started_sender_setup,
        validate_share_path,
    };
    use crate::core::options::{AddrInfoOptions, RelayModeOption, SendOptions, apply_options};
    use crate::core::types::EntryType;
    use iroh::{EndpointAddr, RelayUrl, SecretKey, TransportAddr};
    use std::path::Path;
    use std::str::FromStr;

    fn sample_addr() -> iroh::EndpointAddr {
        let node_id = SecretKey::generate().public();
        let relay = RelayUrl::from_str("https://relay.example").expect("valid relay url");
        let ip = "127.0.0.1:7777".parse().expect("valid socket addr");
        EndpointAddr::new(node_id)
            .with_relay_url(relay)
            .with_ip_addr(ip)
    }

    #[test]
    fn apply_options_matches_ticket_type_semantics() {
        let base = sample_addr();

        let mut id_only = base.clone();
        apply_options(&mut id_only, AddrInfoOptions::Id);
        assert!(id_only.addrs.is_empty());

        let mut relay_only = base.clone();
        apply_options(&mut relay_only, AddrInfoOptions::Relay);
        assert!(
            relay_only
                .addrs
                .iter()
                .all(|addr| matches!(addr, TransportAddr::Relay(_)))
        );
        assert!(!relay_only.addrs.is_empty());

        let mut ip_only = base.clone();
        apply_options(&mut ip_only, AddrInfoOptions::Addresses);
        assert!(
            ip_only
                .addrs
                .iter()
                .all(|addr| matches!(addr, TransportAddr::Ip(_)))
        );
        assert!(!ip_only.addrs.is_empty());

        let mut full = base.clone();
        apply_options(&mut full, AddrInfoOptions::RelayAndAddresses);
        assert_eq!(full.addrs.len(), base.addrs.len());
    }

    #[test]
    fn disabled_relay_skips_online_wait() {
        let wait_for_online = !matches!(
            crate::core::options::RelayModeOption::Disabled,
            crate::core::options::RelayModeOption::Disabled
        );
        assert!(!wait_for_online);
    }

    #[test]
    fn canonicalized_relative_path_uses_forward_slashes() {
        let path = Path::new("folder").join("nested").join("file.txt");
        let value = canonicalized_path_to_string(&path, true).expect("path should convert");
        assert_eq!(value, "folder/nested/file.txt");
    }

    #[test]
    fn canonicalized_absolute_path_keeps_leading_slash_when_allowed() {
        let value = canonicalized_path_to_string(Path::new("/folder/file.txt"), false)
            .expect("absolute path should convert");
        assert_eq!(value, "/folder/file.txt");
    }

    #[test]
    fn canonicalized_absolute_path_is_rejected_when_relative_required() {
        let err = canonicalized_path_to_string(Path::new("/folder/file.txt"), true)
            .expect_err("absolute path should be rejected");
        assert!(err.to_string().contains("invalid path component"));
    }

    #[test]
    fn detect_entry_type_distinguishes_file_and_directory() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let file_path = temp_dir.path().join("demo.txt");
        std::fs::write(&file_path, b"demo").expect("write file");

        assert_eq!(detect_entry_type(&file_path), EntryType::File);
        assert_eq!(detect_entry_type(temp_dir.path()), EntryType::Directory);
    }

    #[test]
    fn collect_import_sources_returns_relative_sorted_names_after_sorting() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let root = temp_dir.path().join("data");
        let nested = root.join("nested");
        std::fs::create_dir_all(&nested).expect("create dirs");
        std::fs::write(root.join("alpha.txt"), b"a").expect("write alpha");
        std::fs::write(nested.join("beta.txt"), b"b").expect("write beta");

        let mut names = collect_import_sources(root)
            .expect("sources")
            .into_iter()
            .map(|source| source.name)
            .collect::<Vec<_>>();
        names.sort();

        assert_eq!(names, vec!["data/alpha.txt", "data/nested/beta.txt"]);
    }

    #[test]
    fn import_parallelism_is_bounded_by_available_work() {
        assert_eq!(import_parallelism(0), 1);
        assert_eq!(import_parallelism(1), 1);
        assert!(import_parallelism(usize::MAX) >= 1);
    }

    #[test]
    fn validate_share_path_rejects_current_directory_aliases() {
        let dot_err = validate_share_path(Path::new("."))
            .expect_err("`.` should be treated as current directory");
        assert!(dot_err.to_string().contains("current directory"));

        let dot_slash_err = validate_share_path(Path::new("./"))
            .expect_err("`./` should be treated as current directory");
        assert!(dot_slash_err.to_string().contains("current directory"));
    }

    #[test]
    fn validate_share_path_rejects_current_directory_absolute_path() {
        let cwd = std::env::current_dir().expect("current dir");
        let err =
            validate_share_path(&cwd).expect_err("absolute current directory should be rejected");
        assert!(err.to_string().contains("current directory"));
    }

    #[test]
    fn validate_share_path_rejects_missing_path() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let missing = temp_dir.path().join("missing-share");

        let err = validate_share_path(&missing).expect_err("missing path should be rejected");

        assert!(
            err.to_string()
                .contains("does not exist or cannot be accessed")
        );
    }

    #[test]
    fn validate_share_path_accepts_nested_path() {
        let temp_dir = tempfile::tempdir().expect("temp dir");
        let nested = temp_dir.path().join("nested").join("share");
        std::fs::create_dir_all(&nested).expect("create nested dir");
        validate_share_path(&nested).expect("nested path should be accepted");
    }

    #[tokio::test]
    async fn sender_setup_failure_removes_blob_store() {
        let source_dir = tempfile::tempdir().expect("source directory");
        let missing_source = source_dir.path().join("missing-source.bin");
        let storage_root = tempfile::tempdir().expect("storage root");
        let blobs_data_dir = storage_root.path().join("sender-store");
        let options = SendOptions {
            relay_mode: RelayModeOption::Disabled,
            ticket_type: AddrInfoOptions::RelayAndAddresses,
            ..SendOptions::default()
        };
        let endpoint = prepare_endpoint(&options)
            .await
            .expect("create sender endpoint");
        let share_request = ShareRequest {
            path: missing_source,
            entry_type: EntryType::File,
            app_handle: None,
        };

        let error = match setup_data_sharing(endpoint, blobs_data_dir.clone(), share_request).await
        {
            Ok(_) => panic!("missing source should fail sender setup"),
            Err(error) => error,
        };

        assert!(!error.to_string().is_empty());
        assert!(
            !blobs_data_dir.exists(),
            "sender setup failure should remove its blob store"
        );
    }

    #[tokio::test]
    async fn cancelling_after_router_starts_removes_blob_store() {
        let source_dir = tempfile::tempdir().expect("source directory");
        let source_file = source_dir.path().join("source.bin");
        tokio::fs::write(&source_file, b"cancel after setup")
            .await
            .expect("write source file");
        let storage_root = tempfile::tempdir().expect("storage root");
        let blobs_data_dir = storage_root.path().join("sender-store");
        let options = SendOptions {
            relay_mode: RelayModeOption::Disabled,
            ticket_type: AddrInfoOptions::RelayAndAddresses,
            ..SendOptions::default()
        };
        let endpoint = prepare_endpoint(&options)
            .await
            .expect("create sender endpoint");
        let setup = setup_data_sharing(
            endpoint,
            blobs_data_dir.clone(),
            ShareRequest {
                path: source_file,
                entry_type: EntryType::File,
                app_handle: None,
            },
        )
        .await
        .expect("start sender router");

        let error = shutdown_started_sender_setup(
            setup,
            EntryType::File,
            AddrInfoOptions::RelayAndAddresses,
            anyhow::anyhow!("Operation cancelled"),
        )
        .await;

        assert!(error.to_string().contains("Operation cancelled"));
        assert!(
            !blobs_data_dir.exists(),
            "cancelling a started sender should remove its blob store"
        );
    }
}