tachyon-i2p 0.0.2

//! [`Destination`]: a local I2P destination (an eepsite's identity), reachable at a
//! `.b32.i2p` address.

use crate::error::I2pError;
use crate::router::I2pRouter;
use crate::stream::I2pStream;
use std::ffi::c_void;
use std::os::raw::c_int;
use std::sync::Arc;

/// How many not-yet-[`accept`](Destination::accept)ed inbound streams may queue up before new
/// ones are refused outright (closed immediately, on libi2pd's own accept thread, without ever
/// reaching this side). Bounded so a destination nobody is calling `accept` on can't grow memory
/// without limit; libi2pd's accept callback must never block (see `shim.h`), so "refuse" -- not
/// "wait for room" -- is the only option when this is full.
const ACCEPT_QUEUE_CAPACITY: usize = 128;

/// A raw stream pointer, moved from libi2pd's own accept thread into the queue an
/// [`accept`](Destination::accept) call drains. `i2pd_sys::I2pdStream` is a small owned-pointer
/// wrapper (see `shim.h`); nothing about handing the pointer value itself across threads is
/// unsafe -- only dereferencing it needs care, which `I2pStream` handles.
struct RawStream(*mut i2pd_sys::I2pdStream);
// SAFETY: the pointer is treated as an opaque, movable handle here; see `shim.h`'s thread-safety
// note -- libi2pd's own Stream API is designed to be driven from an arbitrary thread.
unsafe impl Send for RawStream {}

struct AcceptCtx {
    tx: tokio::sync::mpsc::Sender<RawStream>,
}

/// Invoked on libi2pd's own internal thread for every inbound stream. Must never block (per
/// `shim.h`'s contract) -- `try_send` either queues the stream or, if the queue is full, closes
/// it immediately right here (both non-blocking libi2pd calls).
extern "C" fn on_accept(ctx: *mut c_void, stream: *mut i2pd_sys::I2pdStream) {
    // SAFETY: `ctx` is the `AcceptCtx` this callback was registered with in `Destination::from_raw`,
    // kept alive (see `DestinationHandle`) for exactly as long as the callback can still fire.
    let ctx = unsafe { &*ctx.cast::<AcceptCtx>() };
    if ctx.tx.try_send(RawStream(stream)).is_err() {
        // SAFETY: `stream` is a valid, freshly-created handle this callback uniquely owns until
        // it's either handed off (above) or released (here) -- both `i2pd_stream_close` and
        // `i2pd_destroy_stream` are safe, non-blocking calls per `shim.h`.
        unsafe {
            i2pd_sys::i2pd_stream_close(stream);
            i2pd_sys::i2pd_destroy_stream(stream);
        }
    }
}

pub(crate) struct DestinationHandle {
    pub(crate) ptr: *mut i2pd_sys::I2pdDestination,
    accept_ctx: *mut AcceptCtx,
}

// SAFETY: every libi2pd entry point this handle's pointer is passed to is documented (`shim.h`)
// as safe to call from an arbitrary thread.
unsafe impl Send for DestinationHandle {}
unsafe impl Sync for DestinationHandle {}

impl Drop for DestinationHandle {
    fn drop(&mut self) {
        // SAFETY: `self.ptr` is a valid destination handle owned by this struct. libi2pd's
        // `ClientDestination::Stop` (invoked synchronously by `i2pd_destroy_destination`) clears
        // its accept-callback registration *before* returning, so it's guaranteed `on_accept`
        // cannot fire again once this call returns -- freeing `accept_ctx` right after is safe.
        unsafe {
            i2pd_sys::i2pd_destroy_destination(self.ptr);
            drop(Box::from_raw(self.accept_ctx));
        }
    }
}

/// A local I2P destination -- an eepsite's identity, reachable at a stable `.b32.i2p` address
/// for as long as this value (or a clone of the [`I2pRouter`] it came from, if that matters to
/// your use case) is alive.
///
/// Create one via [`I2pRouter::create_transient_destination`],
/// [`I2pRouter::create_persistent_destination`], or
/// [`I2pRouter::destination_from_keys_file`].
#[derive(Debug)]
pub struct Destination {
    handle: Arc<DestinationHandle>,
    b32_address: Box<str>,
    ident_hash: [u8; 32],
    accept_rx: tokio::sync::mpsc::Receiver<RawStream>,
    // Keeps the whole router (and thus libi2pd's global crypto/context state) alive for at least
    // as long as this destination. Declared *last*: Rust drops struct fields in declaration
    // order, and `handle`'s own `Drop` (tearing down this destination via libi2pd, touching the
    // global tunnel pool/netDb) must run before the router's `Drop` (tearing down those same
    // globals) -- not after.
    router: I2pRouter,
}

impl std::fmt::Debug for DestinationHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DestinationHandle").finish_non_exhaustive()
    }
}

impl Destination {
    /// Takes ownership of a (possibly null, on failure) destination pointer fresh out of
    /// libi2pd, fetches its address, and registers the persistent multi-stream acceptor. Called
    /// only from a `spawn_blocking` worker in `router.rs`.
    pub(crate) fn from_raw(
        router: I2pRouter,
        ptr: *mut i2pd_sys::I2pdDestination,
    ) -> Result<Self, I2pError> {
        if ptr.is_null() {
            return Err(I2pError::DestinationCreationFailed);
        }

        // SAFETY: `ptr` was just null-checked and came straight from libi2pd; the returned
        // string pointer (if non-null) is freed immediately after copying it out.
        let b32_address = unsafe {
            let raw = i2pd_sys::i2pd_destination_b32_address(ptr);
            if raw.is_null() {
                i2pd_sys::i2pd_destroy_destination(ptr);
                return Err(I2pError::DestinationCreationFailed);
            }
            let s = std::ffi::CStr::from_ptr(raw.cast())
                .to_string_lossy()
                .into_owned();
            i2pd_sys::i2pd_free_string(raw);
            s
        };

        // SAFETY: `ptr` was just null-checked and came straight from libi2pd; `out` is a valid
        // 32-byte local buffer.
        let mut ident_hash = [0u8; 32];
        let ok = unsafe { i2pd_sys::i2pd_destination_ident_hash(ptr, ident_hash.as_mut_ptr()) };
        if ok == 0 {
            // SAFETY: `ptr` is still valid and not yet handed to any `DestinationHandle`.
            unsafe { i2pd_sys::i2pd_destroy_destination(ptr) };
            return Err(I2pError::DestinationCreationFailed);
        }

        let (tx, accept_rx) = tokio::sync::mpsc::channel(ACCEPT_QUEUE_CAPACITY);
        let accept_ctx = Box::into_raw(Box::new(AcceptCtx { tx }));

        // SAFETY: `ptr` is a live destination; `accept_ctx` stays valid until `DestinationHandle`
        // is dropped, which only happens after this destination is destroyed (see its `Drop`
        // impl) -- guaranteeing the callback never fires against a freed `AcceptCtx`.
        unsafe {
            i2pd_sys::i2pd_accept_stream(ptr, Some(on_accept), accept_ctx.cast());
        }

        Ok(Self {
            handle: Arc::new(DestinationHandle { ptr, accept_ctx }),
            b32_address: b32_address.into_boxed_str(),
            ident_hash,
            accept_rx,
            router,
        })
    }

    /// This destination's `<52 chars>.b32.i2p` address.
    #[must_use]
    pub fn b32_address(&self) -> &str {
        &self.b32_address
    }

    /// This destination's raw 32-byte `IdentHash` -- the form
    /// [`connect`](Self::connect) expects for a remote destination. `b32_address` is derived
    /// from this same hash.
    #[must_use]
    pub const fn ident_hash(&self) -> [u8; 32] {
        self.ident_hash
    }

    /// Waits for and returns the next inbound stream. Multiple concurrent/repeated calls
    /// (e.g. one per accepted connection, in a loop) are the intended usage -- unlike the
    /// underlying libi2pd accept callback, which fires for every inbound stream indefinitely,
    /// not just once.
    ///
    /// # Errors
    /// Returns [`I2pError::DestinationClosed`] if this destination is dropped while a call is
    /// pending.
    pub async fn accept(&mut self) -> Result<I2pStream, I2pError> {
        let RawStream(ptr) = self
            .accept_rx
            .recv()
            .await
            .ok_or(I2pError::DestinationClosed)?;
        // SAFETY: `ptr` is a freshly-accepted, uniquely-owned stream handle handed to us by
        // `on_accept`; `I2pStream::from_raw` takes ownership of it.
        Ok(unsafe { I2pStream::from_raw(self.router.clone(), ptr) })
    }

    /// Opens an outbound stream from this destination to the destination identified by
    /// `remote_ident_hash` (its raw 32-byte `IdentHash`, *not* its `.b32.i2p` string form).
    /// Blocks (internally, off the async runtime thread) for up to `timeout` waiting for the
    /// remote's lease set to become known and a real I2P tunnel to be built -- this commonly
    /// takes several seconds to a few minutes on a cold start.
    ///
    /// `timeout` is truncated to whole seconds (libi2pd's own retry loop is second-granular); in
    /// particular any `timeout` under one second -- including [`Duration::ZERO`](std::time::Duration::ZERO)
    /// -- truncates to zero, which libi2pd treats as "try once, don't wait or retry at all"
    /// rather than "wait almost a second".
    ///
    /// # Errors
    /// Returns [`I2pError::ConnectFailed`] on timeout or failure.
    pub async fn connect(
        &self,
        remote_ident_hash: [u8; 32],
        timeout: std::time::Duration,
    ) -> Result<I2pStream, I2pError> {
        let handle = self.handle.clone();
        let router = self.router.clone();
        let timeout_secs = c_int::try_from(timeout.as_secs()).unwrap_or(c_int::MAX);
        let RawStream(ptr) = tokio::task::spawn_blocking(move || {
            // SAFETY: `handle.ptr` is a live destination for the duration of this blocking call
            // (the `Arc<DestinationHandle>` clone above keeps it alive even if `self` is
            // dropped concurrently); `remote_ident_hash` is a valid 32-byte buffer.
            RawStream(unsafe {
                i2pd_sys::i2pd_create_stream(handle.ptr, remote_ident_hash.as_ptr(), timeout_secs)
            })
        })
        .await
        .map_err(|_| I2pError::WorkerPanicked)?;

        if ptr.is_null() {
            return Err(I2pError::ConnectFailed);
        }
        // SAFETY: `ptr` is a freshly-created, uniquely-owned stream handle.
        Ok(unsafe { I2pStream::from_raw(router, ptr) })
    }
}