openrtc 1.0.4

OpenRTC: a Rust-first P2P runtime for device discovery, signaling, and iroh/QUIC networking.
Documentation
#![cfg(not(target_arch = "wasm32"))]

use anyhow::{Context, Result};
use base64::Engine as _;
use openrtc::client::{AuthMode, Client};
use openrtc::native_auth::{FirebaseAuthConfig, NativeAuthState, PublicSpaceTokenRequest};
use serde_json::{json, Value};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

const DEFAULT_PROJECT_ID: &str = "pluto-rtc-prod";
const DEFAULT_API_KEY: &str = "pk_test_integration_test";
const DEFAULT_SECRET_KEY: &str = "sk_test_integration_plutonium";
const DEFAULT_APP_TAG: &str = "app_integration_test";

fn enabled() -> bool {
    std::env::var("OPENRTC_NATIVE_AUTH_EMULATOR")
        .map(|value| value == "1" || value.eq_ignore_ascii_case("true"))
        .unwrap_or(false)
}

fn project_id() -> String {
    std::env::var("OPENRTC_EMULATOR_PROJECT_ID").unwrap_or_else(|_| DEFAULT_PROJECT_ID.to_string())
}

fn api_key() -> String {
    std::env::var("OPENRTC_EMULATOR_API_KEY").unwrap_or_else(|_| DEFAULT_API_KEY.to_string())
}

fn secret_key() -> String {
    std::env::var("OPENRTC_EMULATOR_SECRET_KEY").unwrap_or_else(|_| DEFAULT_SECRET_KEY.to_string())
}

fn host_from_env(keys: &[&str], fallback: &str) -> String {
    keys.iter()
        .find_map(|key| {
            std::env::var(key)
                .ok()
                .map(|value| value.trim().trim_end_matches('/').to_string())
                .filter(|value| !value.is_empty())
        })
        .unwrap_or_else(|| fallback.to_string())
}

fn firebase_config(api_key: &str) -> FirebaseAuthConfig {
    let auth_host = host_from_env(&["FIREBASE_AUTH_EMULATOR_HOST"], "127.0.0.1:9100");
    FirebaseAuthConfig::new(api_key)
        .identity_toolkit_base_url(format!("http://{auth_host}/identitytoolkit.googleapis.com"))
        .secure_token_base_url(format!("http://{auth_host}/securetoken.googleapis.com"))
        .refresh_margin(Duration::from_secs(30))
}

fn functions_origin(project_id: &str) -> String {
    if let Ok(origin) = std::env::var("OPENRTC_FUNCTIONS_URL") {
        let trimmed = origin.trim().trim_end_matches('/');
        if !trimmed.is_empty() {
            return trimmed.to_string();
        }
    }
    let host = host_from_env(
        &["OPENRTC_FUNCTIONS_EMULATOR_HOST", "FUNCTIONS_EMULATOR_HOST"],
        "127.0.0.1:5002",
    );
    format!("http://{host}/{project_id}/us-central1")
}

fn firestore_base_url() -> String {
    let host = host_from_env(
        &["FIRESTORE_EMULATOR_HOST", "OPENRTC_FIRESTORE_HOST"],
        "127.0.0.1:8082",
    );
    format!("http://{host}/v1/projects")
}

fn now_millis() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("system time after epoch")
        .as_millis() as i64
}

fn decode_claims(token: &str) -> Result<Value> {
    let payload = token
        .split('.')
        .nth(1)
        .context("Firebase ID token missing payload segment")?;
    let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
        .decode(payload)
        .context("Firebase ID token payload was not base64url")?;
    serde_json::from_slice(&bytes).context("Firebase ID token payload was not JSON")
}

async fn mint_custom_token(project_id: &str, user_id: &str) -> Result<String> {
    let response = reqwest::Client::new()
        .post(format!("{}/v1Tokens", functions_origin(project_id)))
        .bearer_auth(secret_key())
        .json(&json!({ "userId": user_id, "ttl": 3600 }))
        .send()
        .await
        .context("v1Tokens request failed")?;
    let status = response.status();
    let body: Value = response.json().await.unwrap_or(Value::Null);
    if !status.is_success() {
        anyhow::bail!("v1Tokens failed status={status} body={body}");
    }
    body.get("token")
        .and_then(Value::as_str)
        .map(ToOwned::to_owned)
        .context("v1Tokens response missing token")
}

async fn provision_space(project_id: &str, api_key: &str, space_key: &str) -> Result<String> {
    let app_tag = openrtc::space_app_tag_from_keys(api_key, space_key);
    let namespace_id = app_tag
        .strip_prefix("space::")
        .context("space app tag missing space:: prefix")?
        .to_string();
    let now = now_millis();
    let url = format!(
        "{}/{}/databases/(default)/documents/provisioned_spaces/{}",
        firestore_base_url(),
        project_id,
        namespace_id
    );
    let response = reqwest::Client::new()
        .patch(url)
        .bearer_auth("owner")
        .json(&json!({
            "fields": {
                "namespaceId": { "stringValue": namespace_id },
                "apiKey": { "stringValue": api_key },
                "appTag": { "stringValue": DEFAULT_APP_TAG },
                "label": { "stringValue": space_key },
                "status": { "stringValue": "active" },
                "requireScopedAuth": { "booleanValue": true },
                "deviceCount": { "integerValue": "0" },
                "roomCount": { "integerValue": "0" },
                "createdAt": { "integerValue": now.to_string() },
                "updatedAt": { "integerValue": now.to_string() }
            }
        }))
        .send()
        .await
        .context("provisioned space write failed")?;
    let status = response.status();
    if !status.is_success() {
        let body = response.text().await.unwrap_or_default();
        anyhow::bail!("provisioned space write failed status={status} body={body}");
    }
    Ok(namespace_id)
}

async fn endpoint_ticket(client: &Client) -> Result<String> {
    client.init_iroh(None, vec![]).await?;
    client.endpoint_ticket().await
}

async fn assert_native_presence_roundtrip(
    publisher: &Client,
    observer: &Client,
    user_id: &str,
    device_id: &str,
    device_name: &str,
) -> Result<()> {
    let ticket = endpoint_ticket(publisher).await?;
    let metadata = format!(r#"{{"deviceId":"{device_id}","source":"native-auth-emulator"}}"#);
    publisher
        .update_durable_device_record_with_ttl(
            user_id,
            device_name,
            &ticket,
            7 * 24 * 60 * 60 * 1000,
            Some(&metadata),
        )
        .await
        .context("durable Firestore roster write failed")?;
    publisher
        .update_live_presence_record(user_id, device_name, &ticket, Some(&metadata))
        .await
        .context("RTDB live presence write failed")?;

    let mut last_devices = Vec::new();
    let observed = {
        let mut found = None;
        for _ in 0..40 {
            let devices = observer
                .search_devices(user_id)
                .await
                .context("native discovery read failed")?;
            if let Some(device) = devices
                .iter()
                .find(|device| device.device_id == device_id)
                .cloned()
            {
                found = Some(device);
                break;
            }
            last_devices = devices;
            tokio::time::sleep(Duration::from_millis(250)).await;
        }
        found.with_context(|| {
            format!("native discovery did not return {device_id}: {last_devices:?}")
        })?
    };
    assert_eq!(observed.device_name, device_name);
    assert!(
        observed.online,
        "RTDB overlay did not promote device online: {observed:?}"
    );
    assert_eq!(observed.ticket.as_deref(), Some(ticket.as_str()));
    assert_eq!(observed.metadata.as_deref(), Some(metadata.as_str()));
    Ok(())
}

#[tokio::test]
async fn native_auth_emulator_exercises_user_and_space_rtdb_presence() -> Result<()> {
    if !enabled() {
        eprintln!("skipping native auth emulator test: set OPENRTC_NATIVE_AUTH_EMULATOR=1");
        return Ok(());
    }

    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();

    let project_id = project_id();
    let api_key = api_key();
    let config = firebase_config(&api_key);

    let managed_user_id = format!("emulator_native_auth_{}", now_millis());
    let custom_token = mint_custom_token(&project_id, &managed_user_id).await?;
    let user_auth = NativeAuthState::new();
    let user_session = user_auth
        .sign_in_with_custom_token(config.clone(), &custom_token)
        .await
        .context("native custom-token exchange failed")?;
    let user_claims = decode_claims(&user_session.id_token)?;
    let firebase_uid = user_session
        .user_id
        .as_deref()
        .or_else(|| user_claims.get("user_id").and_then(Value::as_str))
        .context("custom-token session missing Firebase uid")?;
    assert!(
        firebase_uid.starts_with("pluto_app_integration_test_"),
        "managed token should exchange to the scoped Firebase uid used by rules, got {firebase_uid}"
    );
    assert_ne!(
        firebase_uid, managed_user_id,
        "managed app user ids must not be used directly as Firebase Auth uids"
    );
    assert_eq!(
        user_claims.get("appTag").and_then(Value::as_str),
        Some(DEFAULT_APP_TAG)
    );

    let user_observer_auth = user_auth.clone();
    let user_client =
        Client::builder_with_native_auth(project_id.clone(), api_key.clone(), user_auth)
            .auth_mode(AuthMode::External)
            .build();
    let user_observer =
        Client::builder_with_native_auth(project_id.clone(), api_key.clone(), user_observer_auth)
            .auth_mode(AuthMode::External)
            .build();
    assert_eq!(user_client.app_tag(), DEFAULT_APP_TAG);
    assert_native_presence_roundtrip(
        &user_client,
        &user_observer,
        firebase_uid,
        "native-user-device",
        "Native user-device emulator peer",
    )
    .await?;

    let space_key = format!("native-space-{}", now_millis());
    let namespace_id = provision_space(&project_id, &api_key, &space_key).await?;
    let space_auth = NativeAuthState::new();
    let space_session = space_auth
        .sign_in_with_public_space_token(
            config,
            functions_origin(&project_id),
            PublicSpaceTokenRequest::new(api_key.clone(), space_key.clone())
                .member_id("native-space-member")
                .ttl(3600),
        )
        .await
        .context("native public-space token exchange failed")?;
    let space_claims = decode_claims(&space_session.id_token)?;
    assert_eq!(
        space_claims.get("namespaceId").and_then(Value::as_str),
        Some(namespace_id.as_str())
    );
    assert_eq!(
        space_claims.get("appTag").and_then(Value::as_str),
        Some(format!("space::{namespace_id}").as_str())
    );

    let space_observer_auth = space_auth.clone();
    let space_client = Client::builder_with_native_space_auth(
        project_id.clone(),
        api_key.clone(),
        space_key.clone(),
        space_auth,
    )
    .build();
    let space_observer =
        Client::builder_with_native_space_auth(project_id, api_key, space_key, space_observer_auth)
            .build();
    assert_eq!(space_client.app_tag(), format!("space::{namespace_id}"));
    assert_native_presence_roundtrip(
        &space_client,
        &space_observer,
        "native-space-member",
        "native-space-device",
        "Native space emulator peer",
    )
    .await?;

    Ok(())
}