iris-chat 0.1.44

Iris Chat command line client and shared encrypted chat core
Documentation
use super::*;
use base64::Engine;
use fips_core::config::{PeerConfig, RoutingMode, TransportInstances, UdpConfig};
use fips_core::FipsEndpoint;
use hashtree_core::{BlobRoute, MemoryStore, StoreBlobRoute};
use hashtree_fips_transport::{TcpBlobTransport, TcpBlobTransportConfig, TCP_BLOB_CAPABILITY};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, UdpSocket};
use std::time::Instant;

fn reserve_udp_addr() -> SocketAddr {
    let socket = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).expect("reserve UDP address");
    socket.local_addr().expect("reserved UDP address")
}

async fn endpoint(
    rendezvous: SocketAddrV4,
    udp_addr: SocketAddr,
    scope: &str,
    peers: Vec<PeerConfig>,
) -> Arc<FipsEndpoint> {
    let mut config = fips_core::Config::new();
    config.node.discovery.nostr.enabled = false;
    config.node.discovery.lan.enabled = false;
    config.node.discovery.local.rendezvous_addr = rendezvous;
    config.node.routing.mode = RoutingMode::ReplyLearned;
    config.peers = peers;
    config.transports.udp = TransportInstances::Single(UdpConfig {
        bind_addr: Some(udp_addr.to_string()),
        advertise_on_nostr: Some(false),
        public: Some(false),
        ..UdpConfig::default()
    });
    Arc::new(
        FipsEndpoint::builder()
            .config(config)
            .discovery_scope(scope)
            .local_rendezvous()
            .without_system_tun()
            .bind()
            .await
            .expect("bind test endpoint"),
    )
}

fn wait_until(mut condition: impl FnMut() -> bool) {
    let deadline = Instant::now() + Duration::from_secs(8);
    while !condition() {
        assert!(Instant::now() < deadline, "condition timed out");
        std::thread::sleep(Duration::from_millis(25));
    }
}

async fn seed_file(store: Arc<MemoryStore>, bytes: &[u8]) -> String {
    let tree = HashTree::new(HashTreeConfig::new(store));
    let (cid, _) = tree.put(bytes).await.expect("seed Hashtree file");
    nhash_encode_full(&NHashData {
        hash: cid.hash,
        decrypt_key: cid.key,
    })
    .expect("encode seeded file")
}

async fn download_bytes(nhash: &str) -> Vec<u8> {
    let data = download_hashtree_attachment_base64(nhash)
        .await
        .expect("download attachment");
    base64::engine::general_purpose::STANDARD
        .decode(data)
        .expect("decode attachment")
}

#[test]
fn single_device_without_relays_reuses_provider_and_preserves_fallback_and_outbound_link() {
    let provider_runtime = tokio::runtime::Runtime::new().expect("provider runtime");
    let rendezvous = match reserve_udp_addr() {
        SocketAddr::V4(addr) => addr,
        SocketAddr::V6(_) => unreachable!("IPv4 reservation"),
    };
    let sibling_rendezvous = match reserve_udp_addr() {
        SocketAddr::V4(addr) => addr,
        SocketAddr::V6(_) => unreachable!("IPv4 reservation"),
    };
    let sibling_addr = reserve_udp_addr();
    let sibling = provider_runtime.block_on(endpoint(
        sibling_rendezvous,
        sibling_addr,
        "iris-chat-sibling-test",
        Vec::new(),
    ));
    let provider_endpoint = provider_runtime.block_on(endpoint(
        rendezvous,
        reserve_udp_addr(),
        "iris-drive-provider-test",
        Vec::new(),
    ));
    let provider_local = Arc::new(MemoryStore::new());
    let provider_data = b"attachment supplied by the same-host provider".to_vec();
    let provider_nhash =
        provider_runtime.block_on(seed_file(provider_local.clone(), &provider_data));
    let provider = Arc::new(
        provider_runtime
            .block_on(TcpBlobTransport::bind_advertised_with_config(
                provider_endpoint.clone(),
                provider_local,
                TcpBlobTransportConfig::default(),
                100,
            ))
            .expect("bind provider"),
    );
    let fallback = Arc::new(MemoryStore::new());
    let before = b"standalone fallback after provider miss".to_vec();
    let after = b"standalone fallback after provider death".to_vec();
    let before_nhash = provider_runtime.block_on(seed_file(fallback.clone(), &before));
    let after_nhash = provider_runtime.block_on(seed_file(fallback.clone(), &after));
    let fallback: Arc<dyn BlobRoute> = Arc::new(StoreBlobRoute::new(fallback));

    let owner = nostr::Keys::generate();
    let device = nostr::Keys::generate();
    let temp_dir = tempfile::tempdir().expect("Chat data directory");
    let (update_tx, _update_rx) = flume::unbounded();
    let mut core = AppCore::new(
        update_tx,
        flume::unbounded().0,
        temp_dir.path().to_string_lossy().to_string(),
        Arc::new(RwLock::new(AppState::empty())),
    );
    core.logged_in = Some(LoggedInState {
        owner_pubkey: owner.public_key(),
        owner_keys: Some(owner),
        device_keys: device.clone(),
        client: nostr_sdk::Client::new(device),
        relay_urls: Vec::new(),
        authorization_state: LocalAuthorizationState::Authorized,
    });
    assert!(
        core.app_keys.is_empty(),
        "test profile has no roster sibling"
    );
    core.reconcile_same_host_hashtree_for_test(
        rendezvous,
        fallback,
        vec![PeerConfig::new(
            sibling.npub(),
            "udp",
            sibling_addr.to_string(),
        )],
    );
    let (consumer, has_device_sync, sibling_count, store) = core
        .same_host_runtime_for_test()
        .expect("shared FIPS runtime and attachment store");
    assert!(!has_device_sync, "device-sync service stays disabled");
    assert_eq!(sibling_count, 0, "no device-sync peer is synthesized");
    assert!(Arc::ptr_eq(
        &store.store,
        &active_attachment_blob_store().expect("registered attachment store")
    ));

    wait_until(|| {
        consumer
            .local_instance_advertisements()
            .is_ok_and(|adverts| {
                adverts.iter().any(|advert| {
                    advert.npub == provider_endpoint.npub()
                        && advert.capability(TCP_BLOB_CAPABILITY).is_some()
                })
            })
    });
    wait_until(|| {
        core.runtime.block_on(consumer.peers()).is_ok_and(|peers| {
            peers.iter().any(|peer| {
                peer.npub == sibling.npub()
                    && peer.connected
                    && peer.transport_type.as_deref() == Some("udp")
            })
        })
    });

    assert_eq!(
        core.runtime.block_on(download_bytes(&provider_nhash)),
        provider_data
    );
    assert_eq!(core.runtime.block_on(download_bytes(&before_nhash)), before);
    drop(provider);
    provider_runtime
        .block_on(provider_endpoint.shutdown())
        .unwrap();
    wait_until(|| {
        consumer
            .local_instance_advertisements()
            .is_ok_and(|adverts| {
                adverts.iter().all(|advert| {
                    advert.npub != provider_endpoint.npub()
                        || advert.capability(TCP_BLOB_CAPABILITY).is_none()
                })
            })
    });

    assert_eq!(core.runtime.block_on(download_bytes(&after_nhash)), after);
    assert!(core
        .runtime
        .block_on(consumer.peers())
        .unwrap()
        .iter()
        .any(|peer| {
            peer.npub == sibling.npub()
                && peer.connected
                && peer.transport_type.as_deref() == Some("udp")
        }));

    drop(store);
    core.stop_device_sync();
    core.runtime.block_on(async {
        tokio::time::sleep(Duration::from_millis(50)).await;
    });
    provider_runtime.block_on(sibling.shutdown()).unwrap();
}