openrtc 1.0.7

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

pub(crate) type TokenProvider = Arc<Mutex<Box<dyn Fn() -> Option<String> + Send + Sync>>>;

const RENEW_EARLY_MS: u64 = 10_000;
static NEXT_LEASE_ID: AtomicU64 = AtomicU64::new(1);

#[derive(Clone)]
pub(crate) struct LogicalUsageLeaseClient {
    http: reqwest::Client,
    project_id: String,
    token_provider: TokenProvider,
    instance_id: String,
    valid_until_by_key: Arc<Mutex<HashMap<String, u64>>>,
    disabled_for_unit_tests: bool,
}

#[derive(Serialize)]
struct CallableRequest<'a> {
    data: LeaseRequest<'a>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct LeaseRequest<'a> {
    operation_id: &'a str,
    lease_id: &'a str,
    route: &'a str,
}

#[derive(Deserialize)]
struct CallableResponse {
    #[serde(alias = "data")]
    result: LeaseResponse,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct LeaseResponse {
    valid_until_ms: u64,
}

impl LogicalUsageLeaseClient {
    pub(crate) fn new(project_id: impl Into<String>, token_provider: TokenProvider) -> Self {
        let sequence = NEXT_LEASE_ID.fetch_add(1, Ordering::Relaxed);
        let now = crate::firebase::now_millis_u64();
        Self {
            http: reqwest::Client::new(),
            project_id: project_id.into(),
            token_provider,
            instance_id: format!("{now:x}-{sequence:x}"),
            valid_until_by_key: Arc::new(Mutex::new(HashMap::new())),
            disabled_for_unit_tests: cfg!(test),
        }
    }

    pub(crate) async fn ensure(
        &self,
        operation_id: &str,
        resource_key: &str,
        route: &str,
    ) -> Result<u64> {
        if self.disabled_for_unit_tests {
            return Ok(crate::firebase::now_millis_u64());
        }
        let cache_key = format!("{operation_id}\0{resource_key}\0{route}");
        let now = crate::firebase::now_millis_u64();
        if self
            .valid_until_by_key
            .lock()
            .ok()
            .and_then(|cache| cache.get(&cache_key).copied())
            .is_some_and(|valid_until| valid_until > now.saturating_add(RENEW_EARLY_MS))
        {
            return Ok(now);
        }

        let token = (self.token_provider.lock().unwrap())()
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty())
            .context("Logical usage lease requires an auth token")?;
        let lease_id = format!(
            "{}-{}",
            self.instance_id,
            sanitize_resource_key(resource_key),
        );
        let response = self
            .http
            .post(self.callable_url())
            .bearer_auth(token)
            .json(&CallableRequest {
                data: LeaseRequest {
                    operation_id,
                    lease_id: &lease_id,
                    route,
                },
            })
            .send()
            .await
            .context("Logical usage lease request failed")?;
        let status = response.status();
        if !status.is_success() {
            let body = response.text().await.unwrap_or_default();
            anyhow::bail!(
                "Logical usage lease denied with status {}: {}",
                status,
                bounded_error(&body),
            );
        }
        let result = response
            .json::<CallableResponse>()
            .await
            .context("Logical usage lease returned an invalid response")?;
        self.valid_until_by_key
            .lock()
            .unwrap()
            .insert(cache_key, result.result.valid_until_ms);
        Ok(result.result.valid_until_ms)
    }

    fn callable_url(&self) -> String {
        #[cfg(target_arch = "wasm32")]
        {
            if let Some(host) = wasm_global_string("__OPENRTC_FUNCTIONS_EMULATOR_HOST__") {
                return format!(
                    "http://{host}/{}/us-central1/renewNativeUsageLease",
                    self.project_id
                );
            }
        }
        if let Ok(host) = std::env::var("OPENRTC_FUNCTIONS_EMULATOR_HOST") {
            let host = host.trim();
            if !host.is_empty() {
                return format!(
                    "http://{host}/{}/us-central1/renewNativeUsageLease",
                    self.project_id
                );
            }
        }
        format!(
            "https://us-central1-{}.cloudfunctions.net/renewNativeUsageLease",
            self.project_id
        )
    }
}

fn sanitize_resource_key(value: &str) -> String {
    let sanitized = value
        .chars()
        .map(|character| {
            if character.is_ascii_alphanumeric() || matches!(character, '-' | '_') {
                character
            } else {
                '-'
            }
        })
        .take(72)
        .collect::<String>();
    if sanitized.is_empty() {
        "default".to_string()
    } else {
        sanitized
    }
}

fn bounded_error(value: &str) -> String {
    value.chars().take(256).collect()
}

#[cfg(target_arch = "wasm32")]
fn wasm_global_string(key: &str) -> Option<String> {
    let global = js_sys::global();
    js_sys::Reflect::get(&global, &wasm_bindgen::JsValue::from_str(key))
        .ok()
        .and_then(|value| value.as_string())
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn resource_keys_are_bounded_and_callable_safe() {
        assert_eq!(sanitize_resource_key("user/a:b"), "user-a-b");
        assert_eq!(sanitize_resource_key(""), "default");
        assert_eq!(sanitize_resource_key(&"x".repeat(100)).len(), 72);
    }
}