openrtc 0.2.0

OpenRTC: a Rust-first P2P runtime for device discovery, signaling, and iroh/QUIC networking.
Documentation
//! Internal native BLE discovery scaffolding.
//!
//! Full iroh custom-transport data plane integration awaits an
//! `iroh-ble-transport` release compatible with iroh `1.0.0-rc.1`.
//! This module wires BLE peer advertisement/scan via the public OpenRTC fork
//! of `iroh-ble-transport` and tracks
//! discovered peers for opportunistic dialing in private test builds.

#![cfg(all(not(target_arch = "wasm32"), openrtc_unpublished_ble))]

use std::sync::Arc;

use anyhow::Context;
use iroh_ble_transport::{
    AdvertisingConfig, AttributePermissions, BleDevice, Central, CentralConfig, CentralEvent,
    CharacteristicProperties, GattCharacteristic, GattService, Peripheral, ScanFilter,
};
use n0_future::StreamExt;
use sha2::Digest;
use tokio::sync::RwLock;

use crate::local_discovery::LocalDiscoveryRegistry;

/// Matches the discriminator used by the upstream iroh BLE transport.
pub(crate) const BLE_TRANSPORT_ID: u64 = 0x42_4c_45;

pub(crate) const OPENRTC_BLE_SERVICE_UUID: uuid::Uuid =
    uuid::uuid!("6f70656e-7274-632d-626c-652d6e730001");
const OPENRTC_BLE_NODE_ID_CHAR_UUID: uuid::Uuid =
    uuid::uuid!("6f70656e-7274-632d-626c-652d6e730002");

const OPENRTC_BLE_KEY_PREFIX: [u8; 4] = [0x6f, 0x72, 0x74, 0x63];
const OPENRTC_BLE_NAME_PREFIX: &str = "openrtc-ble-";
const NODE_ID_PREFIX_BYTES: usize = 12;

#[derive(Clone)]
pub struct BleDiscoveryService {
    registry: LocalDiscoveryRegistry,
    local_node_id: Arc<RwLock<Option<String>>>,
}

impl BleDiscoveryService {
    pub fn new(registry: LocalDiscoveryRegistry) -> Self {
        Self {
            registry,
            local_node_id: Arc::new(RwLock::new(None)),
        }
    }

    pub async fn start(&self, local_node_id: String) -> anyhow::Result<()> {
        let peripheral: Peripheral = Peripheral::new()
            .await
            .context("failed to open BLE peripheral")?;
        let central: Central = Central::with_config(CentralConfig::default())
            .await
            .context("failed to open BLE central")?;
        self.start_with_adapters(local_node_id, central, peripheral)
            .await
    }

    async fn start_with_adapters<C, P>(
        &self,
        local_node_id: String,
        central: Central<C>,
        peripheral: Peripheral<P>,
    ) -> anyhow::Result<()>
    where
        C: iroh_ble_transport::blew::central::backend::CentralBackend,
        P: iroh_ble_transport::blew::peripheral::backend::PeripheralBackend,
    {
        *self.local_node_id.write().await = Some(local_node_id.clone());

        let key_uuid = node_key_prefix_uuid(&local_node_id);
        peripheral
            .add_service(&GattService {
                uuid: OPENRTC_BLE_SERVICE_UUID,
                primary: true,
                characteristics: vec![ble_node_id_characteristic(&local_node_id)],
            })
            .await
            .context("failed to register OpenRTC BLE service")?;
        peripheral
            .add_service(&GattService {
                uuid: key_uuid,
                primary: false,
                characteristics: vec![],
            })
            .await
            .context("failed to register OpenRTC BLE key-prefix service")?;
        peripheral
            .start_advertising(&AdvertisingConfig {
                local_name: advertised_local_name(&local_node_id),
                service_uuids: vec![OPENRTC_BLE_SERVICE_UUID, key_uuid],
            })
            .await
            .context("failed to start BLE advertising")?;

        let registry = self.registry.clone();
        let local_node_id_for_scan = local_node_id.clone();
        tokio::spawn(async move {
            if let Err(error) = run_ble_scan_loop(registry, local_node_id_for_scan, central).await {
                eprintln!("[pluto-rtc][ble] scan loop exited error={error}");
            }
        });

        Ok(())
    }
}

pub(crate) fn node_key_prefix_uuid(node_id: &str) -> uuid::Uuid {
    let mut bytes = [0u8; 16];
    bytes[..OPENRTC_BLE_KEY_PREFIX.len()].copy_from_slice(&OPENRTC_BLE_KEY_PREFIX);
    let digest = sha2::Sha256::digest(node_id.as_bytes());
    bytes[OPENRTC_BLE_KEY_PREFIX.len()..].copy_from_slice(&digest[..NODE_ID_PREFIX_BYTES]);
    uuid::Uuid::from_bytes(bytes)
}

fn ble_node_id_characteristic(node_id: &str) -> GattCharacteristic {
    GattCharacteristic {
        uuid: OPENRTC_BLE_NODE_ID_CHAR_UUID,
        properties: CharacteristicProperties::READ,
        permissions: AttributePermissions::READ,
        value: node_id.as_bytes().to_vec(),
        descriptors: vec![],
    }
}

pub(crate) fn advertised_local_name(node_id: &str) -> String {
    let short = node_id.chars().take(16).collect::<String>();
    format!("{OPENRTC_BLE_NAME_PREFIX}{short}")
}

fn node_id_hint_from_device(device: &BleDevice) -> Option<String> {
    if !device.services.contains(&OPENRTC_BLE_SERVICE_UUID) {
        return None;
    }
    device
        .name
        .as_deref()
        .and_then(|name| name.strip_prefix(OPENRTC_BLE_NAME_PREFIX))
        .filter(|suffix| !suffix.is_empty())
        .map(ToOwned::to_owned)
}

async fn read_full_node_id<C>(central: &Central<C>, device: &BleDevice) -> Option<String>
where
    C: iroh_ble_transport::blew::central::backend::CentralBackend,
{
    central.connect(&device.id).await.ok()?;
    let bytes = central
        .read_characteristic(&device.id, OPENRTC_BLE_NODE_ID_CHAR_UUID)
        .await
        .ok()?;
    let _ = central.disconnect(&device.id).await;
    String::from_utf8(bytes)
        .ok()
        .filter(|value| !value.is_empty())
}

async fn record_discovered_device<C>(
    registry: &LocalDiscoveryRegistry,
    local_node_id: &str,
    central: Option<&Central<C>>,
    device: &BleDevice,
) where
    C: iroh_ble_transport::blew::central::backend::CentralBackend,
{
    let Some(mut node_id) = node_id_hint_from_device(device) else {
        return;
    };
    if let Some(central) = central {
        if let Some(full_node_id) = read_full_node_id(central, device).await {
            node_id = full_node_id;
        }
    }
    if local_node_id == node_id || local_node_id.starts_with(&node_id) {
        return;
    }
    registry.upsert_peer(node_id).await;
}

async fn run_ble_scan_loop<C>(
    registry: LocalDiscoveryRegistry,
    local_node_id: String,
    central: Central<C>,
) -> anyhow::Result<()>
where
    C: iroh_ble_transport::blew::central::backend::CentralBackend,
{
    let mut events = central.events();
    central
        .start_scan(ScanFilter {
            services: vec![OPENRTC_BLE_SERVICE_UUID],
            ..Default::default()
        })
        .await
        .context("failed to start BLE scan")?;

    while let Some(event) = events.next().await {
        let CentralEvent::DeviceDiscovered(device) = event else {
            continue;
        };
        record_discovered_device(&registry, &local_node_id, Some(&central), &device).await;
    }

    Ok(())
}

pub(crate) fn is_ble_custom_addr(addr: &iroh::TransportAddr) -> bool {
    matches!(addr, iroh::TransportAddr::Custom(custom) if custom.id() == BLE_TRANSPORT_ID)
}

pub async fn maybe_start_ble_discovery(
    enabled: bool,
    registry: &LocalDiscoveryRegistry,
    local_node_id: &str,
) -> anyhow::Result<()> {
    if !enabled {
        return Ok(());
    }
    let service = BleDiscoveryService::new(registry.clone());
    service.start(local_node_id.to_string()).await
}

#[cfg(test)]
mod tests {
    use super::*;
    use iroh_ble_transport::blew::testing::MockLink;

    #[test]
    fn node_key_prefix_uuid_is_stable_and_namespaced() {
        let first = node_key_prefix_uuid("node-a");
        let second = node_key_prefix_uuid("node-a");
        let other = node_key_prefix_uuid("node-b");

        assert_eq!(first, second);
        assert_ne!(first, other);
        assert_eq!(&first.as_bytes()[..4], &OPENRTC_BLE_KEY_PREFIX);
    }

    #[test]
    fn discovered_device_requires_openrtc_service_and_name_prefix() {
        let device = BleDevice {
            id: iroh_ble_transport::DeviceId::from("mock"),
            name: Some("openrtc-ble-peer-node".to_string()),
            rssi: Some(-50),
            services: vec![OPENRTC_BLE_SERVICE_UUID],
        };

        assert_eq!(
            node_id_hint_from_device(&device).as_deref(),
            Some("peer-node")
        );

        let wrong_service = BleDevice {
            services: vec![uuid::Uuid::nil()],
            ..device.clone()
        };
        assert!(node_id_hint_from_device(&wrong_service).is_none());

        let wrong_name = BleDevice {
            name: Some("other-peer".to_string()),
            ..device
        };
        assert!(node_id_hint_from_device(&wrong_name).is_none());
    }

    #[tokio::test]
    async fn mock_ble_advertise_scan_reads_full_peer_id_without_hardware() {
        let (central_ep, peripheral_ep) = MockLink::pair();
        let central = Central::from_backend(central_ep.central);
        let peripheral = Peripheral::from_backend(peripheral_ep.peripheral);
        let registry = LocalDiscoveryRegistry::new();

        let remote_node_id = "remote-node-full-iroh-endpoint-id";
        let key_uuid = node_key_prefix_uuid(remote_node_id);
        peripheral
            .add_service(&GattService {
                uuid: OPENRTC_BLE_SERVICE_UUID,
                primary: true,
                characteristics: vec![ble_node_id_characteristic(remote_node_id)],
            })
            .await
            .unwrap();
        peripheral
            .add_service(&GattService {
                uuid: key_uuid,
                primary: false,
                characteristics: vec![],
            })
            .await
            .unwrap();
        peripheral
            .start_advertising(&AdvertisingConfig {
                local_name: advertised_local_name(remote_node_id),
                service_uuids: vec![OPENRTC_BLE_SERVICE_UUID, key_uuid],
            })
            .await
            .unwrap();

        tokio::spawn(run_ble_scan_loop(
            registry.clone(),
            "local-node".to_string(),
            central,
        ));

        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
        loop {
            let peers = registry.list_peers().await;
            if peers.iter().any(|peer| peer.node_id == remote_node_id) {
                break;
            }
            if tokio::time::Instant::now() >= deadline {
                panic!("mock BLE scan did not record the full node id from GATT");
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }

        let remote = BleDevice {
            id: iroh_ble_transport::DeviceId::from("mock-remote"),
            name: Some("openrtc-ble-remote-node".to_string()),
            rssi: Some(-42),
            services: vec![OPENRTC_BLE_SERVICE_UUID],
        };
        record_discovered_device::<iroh_ble_transport::blew::testing::MockCentral>(
            &registry,
            "local-node",
            None,
            &remote,
        )
        .await;

        let peers = registry.list_peers().await;
        assert!(peers.iter().any(|peer| peer.node_id == "remote-node"));
        assert!(peers.iter().any(|peer| peer.node_id == remote_node_id));
        assert!(peers.iter().all(|peer| peer.local_reachable));
    }
}