use super::{NativeCapabilityRegistration, NativeRosterSource, OpenRtcTauriState};
use openrtc::client::{TransferPeerDemandRequest, TransferPeerDemandStatus};
#[derive(Debug, Clone)]
pub(super) struct NativeTransferPeerDemand {
request: TransferPeerDemandRequest,
runtime_id: String,
generation: Option<u64>,
}
impl NativeCapabilityRegistration {
fn transfer_namespace(&self, token: &str) -> Result<&str, String> {
if token != self.transfer_namespace
|| self.avenue_kind != "devices"
|| matches!(self.native_source, Some(NativeRosterSource::Retired))
{
return Err(
"execution_suspended: transfer peer demand requires the current active Devices capability"
.into(),
);
}
Ok(&self.transfer_namespace)
}
}
impl OpenRtcTauriState {
pub(super) async fn request_native_transfer_peer(
&self,
key: &str,
capability_token: &str,
request: TransferPeerDemandRequest,
) -> Result<TransferPeerDemandStatus, String> {
request.validate().map_err(|error| error.to_string())?;
let runtime_id = {
let mut registry = self.capability_registry.lock().await;
let count: usize = registry
.registrations
.values()
.map(|entry| entry.transfer_demands.len())
.sum();
let registration = registry.registrations.get_mut(key).ok_or_else(|| {
"execution_suspended: native capability is not registered".to_string()
})?;
registration.transfer_namespace(capability_token)?;
if let Some(existing) = registration.transfer_demands.get(&request.id) {
if existing.request != request {
return Err(
"transfer peer demand identity conflicts with an accepted request".into(),
);
}
existing.runtime_id.clone()
} else {
if count >= 4096 {
return Err("receiver_busy: native transfer peer demand input is full".into());
}
let runtime_id = uuid::Uuid::new_v4().to_string();
registration.transfer_demands.insert(
request.id.clone(),
NativeTransferPeerDemand {
request: request.clone(),
runtime_id: runtime_id.clone(),
generation: None,
},
);
runtime_id
}
};
let mut runtime_request = request.clone();
runtime_request.id = runtime_id.clone();
let client = self.client();
let accepted = client.request_transfer_peer(runtime_request).await;
let mut status = match accepted {
Ok(status) => status,
Err(error) => {
let mut registry = self.capability_registry.lock().await;
if let Some(registration) = registry.registrations.get_mut(key) {
if registration
.transfer_demands
.get(&request.id)
.is_some_and(|entry| {
entry.runtime_id == runtime_id && entry.generation.is_none()
})
{
registration.transfer_demands.remove(&request.id);
}
}
return Err(error.to_string());
}
};
let current = {
let mut registry = self.capability_registry.lock().await;
registry
.registrations
.get_mut(key)
.filter(|registration| registration.transfer_namespace(capability_token).is_ok())
.and_then(|registration| registration.transfer_demands.get_mut(&request.id))
.filter(|entry| entry.runtime_id == runtime_id)
.map(|entry| {
entry.generation = Some(status.generation);
true
})
.unwrap_or(false)
};
if !current {
let _ = client
.release_transfer_peer(&runtime_id, status.generation)
.await;
return Err(
"execution_suspended: native transfer demand was retired during acceptance".into(),
);
}
status.id = request.id;
Ok(status)
}
pub(super) async fn native_transfer_peer_demand_status(
&self,
key: &str,
capability_token: &str,
id: &str,
generation: u64,
) -> Result<Option<TransferPeerDemandStatus>, String> {
TransferPeerDemandRequest::validate_identity(id).map_err(|error| error.to_string())?;
let runtime_id = {
let registry = self.capability_registry.lock().await;
let registration = registry.registrations.get(key).ok_or_else(|| {
"execution_suspended: native capability is not registered".to_string()
})?;
registration.transfer_namespace(capability_token)?;
let Some(entry) = registration.transfer_demands.get(id) else {
return Ok(None);
};
if entry
.generation
.is_some_and(|accepted| accepted != generation)
{
return Err(
"execution_suspended: transfer peer demand belongs to a retired generation"
.into(),
);
}
entry.runtime_id.clone()
};
let result = self
.client()
.transfer_peer_demand_status(&runtime_id, generation)
.await
.map_err(|error| error.to_string())?;
let registry = self.capability_registry.lock().await;
let current = registry
.registrations
.get(key)
.filter(|registration| registration.transfer_namespace(capability_token).is_ok())
.and_then(|registration| registration.transfer_demands.get(id))
.is_some_and(|entry| {
entry.runtime_id == runtime_id && entry.generation == Some(generation)
});
if !current {
return Err(
"execution_suspended: native transfer demand was retired during status lookup"
.into(),
);
}
Ok(result.map(|mut status| {
status.id = id.to_owned();
status
}))
}
pub(super) async fn release_native_transfer_peer(
&self,
key: &str,
capability_token: &str,
id: &str,
generation: u64,
) -> Result<bool, String> {
TransferPeerDemandRequest::validate_identity(id).map_err(|error| error.to_string())?;
let entry = {
let mut registry = self.capability_registry.lock().await;
let registration = registry.registrations.get_mut(key).ok_or_else(|| {
"execution_suspended: native capability is not registered".to_string()
})?;
registration.transfer_namespace(capability_token)?;
let Some(entry) = registration.transfer_demands.get(id) else {
return Ok(false);
};
if entry
.generation
.is_some_and(|accepted| accepted != generation)
{
return Err(
"execution_suspended: transfer peer demand belongs to a retired generation"
.into(),
);
}
registration
.transfer_demands
.remove(id)
.expect("checked demand")
};
self.client()
.release_transfer_peer(&entry.runtime_id, generation)
.await
.map_err(|error| error.to_string())
}
pub(super) async fn set_native_transfer_peer_budget(
&self,
key: &str,
capability_token: &str,
budget: usize,
generation: u64,
) -> Result<(), String> {
let namespace = {
let registry = self.capability_registry.lock().await;
registry
.registrations
.get(key)
.ok_or_else(|| {
"execution_suspended: native capability is not registered".to_string()
})?
.transfer_namespace(capability_token)?
.to_owned()
};
self.client()
.set_transfer_peer_budget(budget, generation)
.await
.map_err(|error| error.to_string())?;
let registry = self.capability_registry.lock().await;
if !registry.registrations.get(key).is_some_and(|registration| {
registration
.transfer_namespace(capability_token)
.is_ok_and(|current| current == namespace)
}) {
return Err(
"execution_suspended: native capability retired during budget update".into(),
);
}
Ok(())
}
pub(super) async fn release_native_transfer_peer_demands(
&self,
entries: Vec<NativeTransferPeerDemand>,
) {
let client = self.client();
for entry in entries {
if let Some(generation) = entry.generation {
let _ = client
.release_transfer_peer(&entry.runtime_id, generation)
.await;
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use openrtc::client::{TransferPeerDemandPurpose, TransferPeerDemandState};
fn state() -> OpenRtcTauriState {
OpenRtcTauriState::new(super::super::OpenRtcTauriConfig {
api_key: format!("pk_test_{}", "a".repeat(40)),
..Default::default()
})
}
fn request(id: &str) -> TransferPeerDemandRequest {
TransferPeerDemandRequest {
id: id.into(),
device_id: "offline-device".into(),
purpose: TransferPeerDemandPurpose::FileTransfer,
}
}
async fn register(state: &OpenRtcTauriState, kind: &str, id: &str) -> String {
let key = format!("{kind}:{id}");
let mut registry = state.capability_registry.lock().await;
registry
.register(key.clone(), kind.into(), id.into(), false)
.unwrap();
registry.registrations[&key].transfer_namespace.clone()
}
#[tokio::test]
async fn native_transfer_peer_references_are_scoped_and_close_releases_only_its_jobs() {
let state = state();
let first_token = register(&state, "devices", "first").await;
let second_token = register(&state, "devices", "second").await;
state
.client()
.start_external_auto_connect("root-user".into(), "local".into())
.await
.unwrap();
let first = state
.request_native_transfer_peer("devices:first", &first_token, request("job"))
.await
.unwrap();
let second = state
.request_native_transfer_peer("devices:second", &second_token, request("job"))
.await
.unwrap();
assert_eq!(first.state, TransferPeerDemandState::WaitingForPeer);
assert_eq!(
state
.request_native_transfer_peer("devices:first", &first_token, request("job"))
.await
.unwrap(),
first
);
let mut conflicting = request("job");
conflicting.purpose = TransferPeerDemandPurpose::ContentSync;
assert!(state
.request_native_transfer_peer("devices:first", &first_token, conflicting)
.await
.unwrap_err()
.contains("conflicts"));
assert!(state
.release_native_transfer_peer(
"devices:first",
&first_token,
"job",
first.generation + 1
)
.await
.is_err());
assert!(state
.native_transfer_peer_demand_status(
"devices:first",
&first_token,
"job",
first.generation
)
.await
.unwrap()
.is_some());
assert!(state
.release_native_transfer_peer("devices:first", &first_token, "job", first.generation)
.await
.unwrap());
assert!(state
.native_transfer_peer_demand_status(
"devices:first",
&first_token,
"job",
first.generation
)
.await
.unwrap()
.is_none());
assert!(state
.native_transfer_peer_demand_status(
"devices:second",
&second_token,
"job",
second.generation
)
.await
.unwrap()
.is_some());
let closing = state
.request_native_transfer_peer("devices:first", &first_token, request("close-job"))
.await
.unwrap();
let (closing_runtime_id, other_runtime_id) = {
let registry = state.capability_registry.lock().await;
(
registry.registrations["devices:first"].transfer_demands["close-job"]
.runtime_id
.clone(),
registry.registrations["devices:second"].transfer_demands["job"]
.runtime_id
.clone(),
)
};
state
.unregister_native_capability("devices:first".into())
.await
.unwrap();
assert!(state
.client()
.transfer_peer_demand_status(&closing_runtime_id, closing.generation)
.await
.unwrap()
.is_none());
assert!(state
.client()
.transfer_peer_demand_status(&other_runtime_id, second.generation)
.await
.unwrap()
.is_some());
assert!(state
.request_native_transfer_peer("devices:first", &first_token, request("job"))
.await
.is_err());
let reopened_token = register(&state, "devices", "first").await;
let reopened = state
.request_native_transfer_peer("devices:first", &reopened_token, request("job"))
.await
.unwrap();
assert!(
state
.release_native_transfer_peer(
"devices:first",
&first_token,
"job",
first.generation
)
.await
.is_err(),
"a delayed old capability control must not release the reopened job"
);
assert!(state
.native_transfer_peer_demand_status(
"devices:first",
&first_token,
"job",
first.generation
)
.await
.is_err());
assert!(state
.set_native_transfer_peer_budget("devices:first", &first_token, 1, first.generation)
.await
.is_err());
assert!(state
.request_native_transfer_peer("devices:first", &first_token, request("job"))
.await
.is_err());
assert!(state
.native_transfer_peer_demand_status(
"devices:first",
&reopened_token,
"job",
reopened.generation
)
.await
.unwrap()
.is_some());
assert!(state
.release_native_transfer_peer(
"devices:first",
&reopened_token,
"job",
reopened.generation
)
.await
.unwrap());
assert!(state
.native_transfer_peer_demand_status(
"devices:second",
&second_token,
"job",
second.generation
)
.await
.unwrap()
.is_some());
state
.unregister_native_capability("devices:first".into())
.await
.unwrap();
state
.unregister_native_capability("devices:second".into())
.await
.unwrap();
}
#[tokio::test]
async fn native_transfer_peer_rejects_wrong_avenues_retired_sources_and_invalid_ids_without_leaking_references(
) {
let state = state();
let room_token = register(&state, "room", "collaboration").await;
let owned_token = register(&state, "devices", "owned").await;
assert!(state
.request_native_transfer_peer("devices:owned", &owned_token, request("not-started"))
.await
.is_err());
assert!(
state.capability_registry.lock().await.registrations["devices:owned"]
.transfer_demands
.is_empty()
);
state
.client()
.start_external_auto_connect("root-user".into(), "local".into())
.await
.unwrap();
assert!(state
.request_native_transfer_peer(
"room:collaboration",
&room_token,
request("wrong-avenue")
)
.await
.is_err());
assert!(state
.request_native_transfer_peer("devices:owned", &owned_token, request(&"é".repeat(129)))
.await
.is_err());
assert!(
state.capability_registry.lock().await.registrations["devices:owned"]
.transfer_demands
.is_empty()
);
let valid = state
.request_native_transfer_peer("devices:owned", &owned_token, request("valid"))
.await
.unwrap();
for invalid in ["".to_string(), "\n".to_string(), "é".repeat(129)] {
assert!(
state
.native_transfer_peer_demand_status(
"devices:owned",
&owned_token,
&invalid,
valid.generation
)
.await
.is_err(),
"status must reject invalid job IDs"
);
assert!(
state
.release_native_transfer_peer(
"devices:owned",
&owned_token,
&invalid,
valid.generation
)
.await
.is_err(),
"release must reject invalid job IDs"
);
}
assert!(state
.native_transfer_peer_demand_status(
"devices:owned",
&owned_token,
"valid",
valid.generation
)
.await
.unwrap()
.is_some());
state
.capability_registry
.lock()
.await
.registrations
.get_mut("devices:owned")
.unwrap()
.native_source = Some(NativeRosterSource::Retired);
assert!(state
.request_native_transfer_peer("devices:owned", &owned_token, request("retired"))
.await
.is_err());
assert_eq!(state.capability_registry.lock().await.registrations["devices:owned"].transfer_demands.len(), 1,
"a rejected retired-source request must not replace or erase the earlier accepted reference");
state
.unregister_native_capability("devices:owned".into())
.await
.unwrap();
state
.unregister_native_capability("room:collaboration".into())
.await
.unwrap();
}
}