xlb 0.1.0

//! iroh-blobs adapter — wires iroh-blobs on top of an `xlb-net::Endpoint`.
//!
//! xlb-2: Static-discovery only (permanent seeds). LAN/Swarm discovery layers
//! land in xlb-3 alongside the full five-tier chain.
//!
//! Architecture:
//! - [`BlobTransport`] owns a [`MemStore`] and an iroh [`Router`] that accepts
//!   incoming iroh-blobs connections. This is the seeding side.
//! - [`IrohFetcher`] (private) implements [`BlobSource`] by dialling a peer and
//!   fetching a blob via the iroh-blobs protocol. This is the fetch side.
//! - Call [`BlobTransport::attach_fetcher`] to wire a fetcher into an
//!   [`AssetClass`] — the class fetch chain will then use iroh-blobs for the
//!   configured tier.

use std::sync::Arc;

use async_trait::async_trait;
use bytes::Bytes;
use iroh_blobs::{BlobsProtocol, store::mem::MemStore, Hash as IrohHash};
use xlb_net::{Endpoint, NodeAddr};

use crate::{
    source::{BlobSource, FetchTier},
    AssetClass, BlakeHash,
};

// ─── Hash conversion ─────────────────────────────────────────────────────────

fn to_iroh(h: &BlakeHash) -> IrohHash {
    IrohHash::from_bytes(*h.as_bytes())
}

// ─── BlobTransport ───────────────────────────────────────────────────────────

/// iroh-blobs seeder + fetcher for one xlb process.
///
/// Wraps a [`MemStore`] and an iroh [`Router`] so this node can both serve
/// blobs (incoming iroh-blobs ALPN connections handled by the Router) and
/// fetch blobs from remote peers (via [`IrohFetcher`]).
///
/// Build one per [`AssetClass`]. The `Endpoint` should not be running
/// another accept loop — the Router's background task owns the accept side.
pub struct BlobTransport {
    store: MemStore,
    router: iroh::protocol::Router,
    endpoint: Endpoint,
}

impl BlobTransport {
    /// Build a transport from a bound `xlb-net::Endpoint`.
    ///
    /// Spawns an iroh Router accept loop (on the endpoint's iroh socket) that
    /// handles incoming iroh-blobs requests. The Router calls `set_alpns` on
    /// the shared endpoint, so callers do not need to include
    /// `iroh_blobs::ALPN` in the endpoint builder's ALPN list.
    pub async fn new(endpoint: Endpoint) -> anyhow::Result<Self> {
        let store = MemStore::new();
        let blobs_proto = BlobsProtocol::new(&store, None);
        let router = iroh::protocol::Router::builder(endpoint.inner().clone())
            .accept(iroh_blobs::ALPN, blobs_proto)
            .spawn();
        Ok(Self { store, router, endpoint })
    }

    /// Add a blob to the local store so it can be served to remote peers.
    ///
    /// Returns the [`BlakeHash`] of the data (identical to `BlakeHash::hash(&data)`).
    pub async fn add_blob(&self, data: impl Into<Bytes>) -> anyhow::Result<BlakeHash> {
        let data: Bytes = data.into();
        let xlb_hash = BlakeHash::hash(&data);
        let mut tt = self.store.add_bytes(data).temp_tag().await
            .map_err(|e| anyhow::anyhow!("iroh-blobs add_bytes: {e}"))?;
        // Leak the temp tag so the blob is never GC'd for the life of this transport.
        tt.leak();
        Ok(xlb_hash)
    }

    /// The `NodeAddr` of this transport's endpoint.
    ///
    /// Give this to remote peers so they can resolve our address when fetching.
    pub fn node_addr(&self) -> NodeAddr {
        self.endpoint.endpoint_addr()
    }

    /// Wire an iroh-blobs fetcher into `class` at `tier`.
    ///
    /// The fetcher will dial `seeder` on every cache miss and pull the blob
    /// via the iroh-blobs protocol. Blobs fetched this way are verified by
    /// iroh-blobs' Bao streaming (per-chunk BLAKE3); xlb's own `FetchChain`
    /// then re-verifies at the root level before caching.
    pub fn attach_fetcher(&self, class: &AssetClass, seeder: NodeAddr, tier: FetchTier) {
        class.add_source(Arc::new(IrohFetcher {
            tier,
            endpoint: self.endpoint.clone(),
            peer: seeder,
            local_store: self.store.clone(),
        }));
    }

    /// Shut down the Router accept loop and wait for it to drain.
    pub async fn shutdown(self) {
        if let Err(e) = self.router.shutdown().await {
            tracing::warn!("BlobTransport::shutdown: {e}");
        }
    }
}

// ─── IrohFetcher ─────────────────────────────────────────────────────────────

/// `BlobSource` that fetches from a single iroh peer via the iroh-blobs protocol.
///
/// Each `fetch_raw` call dials the peer, fetches the blob into a local
/// [`MemStore`], and returns the bytes. Subsequent calls for the same hash
/// short-circuit to the in-memory store, so repeated fetches within one
/// session are cheap.
struct IrohFetcher {
    tier: FetchTier,
    endpoint: Endpoint,
    peer: NodeAddr,
    local_store: MemStore,
}

#[async_trait]
impl BlobSource for IrohFetcher {
    fn tier(&self) -> FetchTier {
        self.tier
    }

    async fn fetch_raw(&self, hash: &BlakeHash) -> Option<Bytes> {
        let iroh_hash = to_iroh(hash);

        // Fast path: already in the local store from a prior fetch.
        if let Ok(bytes) = self.local_store.get_bytes(iroh_hash).await {
            return Some(bytes);
        }

        // Dial the remote peer on the iroh-blobs ALPN.
        let conn = self
            .endpoint
            .inner()
            .connect(self.peer.clone(), iroh_blobs::ALPN)
            .await
            .map_err(|e| tracing::warn!(%hash, "iroh-blobs connect: {e}"))
            .ok()?;

        // Fetch the blob from the remote into our local MemStore.
        // iroh-blobs does per-chunk Bao-tree verification internally.
        self.local_store
            .remote()
            .fetch(conn, iroh_hash)
            .await
            .map_err(|e| tracing::warn!(%hash, "iroh-blobs fetch: {e}"))
            .ok()?;

        // Read the now-local blob back.
        self.local_store
            .get_bytes(iroh_hash)
            .await
            .map_err(|e| tracing::warn!(%hash, "iroh-blobs get_bytes: {e}"))
            .ok()
    }
}