#![allow(unused_variables)]
#![allow(unused_mut)]
#![allow(dead_code)]
#![allow(clippy::incompatible_msrv)]
#![allow(clippy::unnecessary_cast)]
#![allow(clippy::single_match)]
#![allow(clippy::items_after_test_module)]
use std::collections::HashMap;
use std::sync::Arc;
use jni::objects::{GlobalRef, JByteArray, JClass, JString, JValue};
use jni::sys::{jboolean, jint, jstring, JavaVM, JNI_VERSION_1_6};
use jni::JNIEnv;
use std::os::raw::c_void;
use std::sync::{LazyLock, Mutex};
static JAVA_VM: LazyLock<Mutex<Option<jni::JavaVM>>> = LazyLock::new(|| Mutex::new(None));
static PEER_EVENT_MANAGER_CLASS: LazyLock<Mutex<Option<GlobalRef>>> =
LazyLock::new(|| Mutex::new(None));
#[cfg(feature = "sync")]
static DOCUMENT_CHANGE_LISTENER: LazyLock<Mutex<Option<GlobalRef>>> =
LazyLock::new(|| Mutex::new(None));
#[cfg(feature = "sync")]
static DOCUMENT_SUBSCRIPTION_ACTIVE: LazyLock<std::sync::atomic::AtomicBool> =
LazyLock::new(|| std::sync::atomic::AtomicBool::new(false));
#[cfg(feature = "sync")]
static FORCE_STORE_ERROR_FOR_TESTING: std::sync::atomic::AtomicBool =
std::sync::atomic::AtomicBool::new(false);
#[cfg(all(feature = "sync", feature = "bluetooth"))]
static OUTBOUND_FRAME_LISTENER: LazyLock<Mutex<Option<GlobalRef>>> =
LazyLock::new(|| Mutex::new(None));
#[cfg(all(feature = "sync", feature = "bluetooth"))]
static OUTBOUND_FRAME_FANOUT: LazyLock<Mutex<Option<peat_mesh::transport::FanoutHandle>>> =
LazyLock::new(|| Mutex::new(None));
#[cfg(feature = "sync")]
static GLOBAL_NODE_HANDLE: LazyLock<Mutex<i64>> = LazyLock::new(|| Mutex::new(0));
#[cfg(feature = "sync")]
fn set_global_node_handle(node: &Arc<PeatNode>) {
store_owning_node_in_slot(&GLOBAL_NODE_HANDLE, node);
}
#[cfg(feature = "sync")]
fn store_owning_node_in_slot(slot: &Mutex<i64>, node: &Arc<PeatNode>) {
if let Ok(mut g) = slot.lock() {
let prev = std::mem::replace(&mut *g, Arc::into_raw(Arc::clone(node)) as i64);
if prev != 0 {
unsafe { drop(Arc::from_raw(prev as *const PeatNode)) };
}
}
}
#[cfg(feature = "sync")]
fn clear_owning_node_slot(slot: &Mutex<i64>) {
if let Ok(mut g) = slot.lock() {
let prev = std::mem::replace(&mut *g, 0);
if prev != 0 {
unsafe { drop(Arc::from_raw(prev as *const PeatNode)) };
}
}
}
#[cfg(all(feature = "bluetooth", target_os = "android"))]
static ANDROID_BLE_TRANSPORT: LazyLock<
Mutex<Option<Arc<PeatBleTransport<peat_btle::platform::android::AndroidAdapter>>>>,
> = LazyLock::new(|| Mutex::new(None));
use peat_protocol::cot::{
CotEncoder, Position as CotPosition, TrackUpdate, Velocity as CotVelocity,
};
#[cfg(feature = "sync")]
use peat_protocol::network::{IrohTransport, PeerInfo as PeatPeerInfo, TransportPeerEvent};
#[cfg(feature = "sync")]
use peat_protocol::storage::{AutomergeBackend, AutomergeStore, StorageBackend, SyncCapable};
#[cfg(feature = "sync")]
use peat_protocol::sync::automerge::AutomergeIrohBackend;
#[cfg(feature = "sync")]
use peat_protocol::sync::{BackendConfig, DataSyncBackend, TransportConfig};
use peat_mesh::storage::automerge_store::{
ChangeOrigin as _PeatMeshChangeOrigin, DocChange as _PeatMeshDocChange,
};
#[cfg(feature = "sync")]
use peat_mesh::storage::{
BlobMetadata, BlobStore, BlobStoreExt, BlobToken, NetworkedIrohBlobStore,
};
#[cfg(feature = "sync")]
use peat_mesh::IrohConfig as PeatMeshIrohConfig;
#[cfg(all(feature = "sync", feature = "bluetooth"))]
use peat_protocol::transport::btle::PeatBleTransport;
#[cfg(feature = "sync")]
use peat_protocol::transport::{
CollectionRouteTable, IrohMeshTransport, Transport, TransportCapabilities, TransportInstance,
TransportManager, TransportManagerConfig, TransportPolicy, TransportType,
};
#[cfg(feature = "sync")]
use std::net::SocketAddr;
#[cfg(feature = "sync")]
use std::path::PathBuf;
#[cfg(feature = "sync")]
use std::sync::atomic::{AtomicBool, Ordering};
uniffi::setup_scaffolding!();
pub mod dart_ffi;
#[cfg(feature = "sync")]
mod water_counter;
#[cfg(feature = "sync")]
mod crdt_kv;
#[uniffi::export]
pub fn peat_version() -> String {
env!("CARGO_PKG_VERSION").to_string()
}
#[derive(Debug, Clone, uniffi::Record)]
pub struct Position {
pub lat: f64,
pub lon: f64,
pub hae: Option<f64>,
}
#[derive(Debug, Clone, uniffi::Record)]
pub struct Velocity {
pub bearing: f64,
pub speed_mps: f64,
}
#[derive(Debug, Clone, uniffi::Record)]
pub struct TrackData {
pub track_id: String,
pub source_node: String,
pub position: Position,
pub velocity: Option<Velocity>,
pub classification: String,
pub confidence: f64,
pub cell_id: Option<String>,
pub formation_id: Option<String>,
}
#[derive(Debug, thiserror::Error, uniffi::Error)]
pub enum PeatError {
#[error("Encoding error: {msg}")]
EncodingError { msg: String },
#[error("Invalid input: {msg}")]
InvalidInput { msg: String },
#[error("Storage error: {msg}")]
StorageError { msg: String },
#[error("Connection error: {msg}")]
ConnectionError { msg: String },
#[error("Sync error: {msg}")]
SyncError { msg: String },
}
#[uniffi::export]
pub fn encode_track_to_cot(track: TrackData) -> Result<String, PeatError> {
if track.track_id.is_empty() {
return Err(PeatError::InvalidInput {
msg: "track_id cannot be empty".to_string(),
});
}
let position = CotPosition {
lat: track.position.lat,
lon: track.position.lon,
cep_m: None,
hae: track.position.hae,
};
let velocity = track.velocity.map(|v| CotVelocity {
bearing: v.bearing,
speed_mps: v.speed_mps,
});
let track_update = TrackUpdate {
track_id: track.track_id,
source_node: track.source_node,
source_model: "peat-ffi".to_string(),
model_version: peat_version(),
cell_id: track.cell_id,
formation_id: track.formation_id,
timestamp: chrono::Utc::now(),
position,
velocity,
classification: track.classification,
confidence: track.confidence,
attributes: HashMap::new(),
};
let encoder = CotEncoder::new();
let event = encoder
.track_update_to_event(&track_update)
.map_err(|e| PeatError::EncodingError { msg: e.to_string() })?;
event
.to_xml()
.map_err(|e| PeatError::EncodingError { msg: e.to_string() })
}
#[uniffi::export]
pub fn create_position(lat: f64, lon: f64, hae: Option<f64>) -> Position {
Position { lat, lon, hae }
}
#[uniffi::export]
pub fn create_velocity(bearing: f64, speed_mps: f64) -> Velocity {
Velocity { bearing, speed_mps }
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, Default, uniffi::Record)]
pub struct TransportConfigFFI {
pub enable_ble: bool,
pub ble_mesh_id: Option<String>,
pub ble_power_profile: Option<String>,
pub transport_preference: Option<Vec<String>>,
pub collection_routes_json: Option<String>,
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, uniffi::Record)]
pub struct NodeConfig {
pub app_id: String,
pub shared_key: String,
pub bind_address: Option<String>,
pub storage_path: String,
pub transport: Option<TransportConfigFFI>,
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, uniffi::Record)]
pub struct PeerInfo {
pub name: String,
pub node_id: String,
pub addresses: Vec<String>,
pub relay_url: Option<String>,
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, uniffi::Record)]
pub struct SyncStats {
pub sync_active: bool,
pub connected_peers: u32,
pub bytes_sent: u64,
pub bytes_received: u64,
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, uniffi::Record)]
pub struct PeerTransportState {
pub peer_id: String,
pub links: Vec<TransportLink>,
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, uniffi::Record)]
pub struct TransportLink {
pub transport_id: String,
pub transport_type: String,
pub interface: Option<String>,
pub quality: TransportLinkQuality,
pub rtt_ms: Option<u32>,
pub rssi_dbm: Option<i8>,
pub path_kind: Option<TransportPathKind>,
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, Copy, uniffi::Enum)]
pub enum TransportLinkQuality {
Excellent,
Good,
Fair,
Weak,
Unknown,
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, Copy, uniffi::Enum)]
pub enum TransportPathKind {
Direct,
Relay,
}
#[cfg(feature = "sync")]
impl From<peat_mesh::transport::LinkQuality> for TransportLinkQuality {
fn from(q: peat_mesh::transport::LinkQuality) -> Self {
match q {
peat_mesh::transport::LinkQuality::Excellent => TransportLinkQuality::Excellent,
peat_mesh::transport::LinkQuality::Good => TransportLinkQuality::Good,
peat_mesh::transport::LinkQuality::Fair => TransportLinkQuality::Fair,
peat_mesh::transport::LinkQuality::Weak => TransportLinkQuality::Weak,
peat_mesh::transport::LinkQuality::Unknown => TransportLinkQuality::Unknown,
}
}
}
#[cfg(feature = "sync")]
impl From<peat_mesh::transport::PathKind> for TransportPathKind {
fn from(p: peat_mesh::transport::PathKind) -> Self {
match p {
peat_mesh::transport::PathKind::Direct => TransportPathKind::Direct,
peat_mesh::transport::PathKind::Relay => TransportPathKind::Relay,
}
}
}
#[cfg(feature = "sync")]
impl From<peat_mesh::transport::LinkState> for TransportLink {
fn from(s: peat_mesh::transport::LinkState) -> Self {
let transport_type = match s.transport_type {
peat_mesh::transport::TransportType::BluetoothLE => "ble".to_string(),
peat_mesh::transport::TransportType::Quic => "iroh".to_string(),
peat_mesh::transport::TransportType::LoRa => "lora".to_string(),
peat_mesh::transport::TransportType::WifiDirect => "wifi-direct".to_string(),
peat_mesh::transport::TransportType::TacticalRadio => "tactical-radio".to_string(),
peat_mesh::transport::TransportType::Satellite => "satellite".to_string(),
peat_mesh::transport::TransportType::BluetoothClassic => {
"bluetooth-classic".to_string()
}
peat_mesh::transport::TransportType::Custom(n) => format!("custom-{n}"),
};
TransportLink {
transport_id: s.transport_id,
transport_type,
interface: s.interface,
quality: s.quality.into(),
rtt_ms: s.rtt_ms,
rssi_dbm: s.rssi_dbm,
path_kind: s.path_kind.map(Into::into),
}
}
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, uniffi::Enum)]
pub enum ChangeType {
Upsert,
Delete,
}
#[cfg(feature = "sync")]
#[derive(Debug, Clone, uniffi::Record)]
pub struct DocumentChange {
pub collection: String,
pub doc_id: String,
pub change_type: ChangeType,
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[derive(Debug, Clone, uniffi::Record)]
pub struct OutboundFrame {
pub transport_id: String,
pub collection: String,
pub bytes: Vec<u8>,
}
#[cfg(feature = "sync")]
#[uniffi::export(callback_interface)]
pub trait DocumentCallback: Send + Sync {
fn on_change(&self, change: DocumentChange);
fn on_error(&self, message: String);
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[uniffi::export(callback_interface)]
pub trait OutboundFrameCallback: Send + Sync {
fn on_frame(&self, transport_id: String, collection: String, bytes: Vec<u8>);
}
#[cfg(feature = "sync")]
#[derive(uniffi::Object)]
pub struct SubscriptionHandle {
active: Arc<AtomicBool>,
pending: Arc<std::sync::Mutex<std::collections::VecDeque<DocumentChange>>>,
}
#[cfg(feature = "sync")]
impl SubscriptionHandle {
fn new(active: Arc<AtomicBool>) -> Self {
Self {
active,
pending: Arc::new(std::sync::Mutex::new(std::collections::VecDeque::new())),
}
}
fn new_with_queue(
active: Arc<AtomicBool>,
pending: Arc<std::sync::Mutex<std::collections::VecDeque<DocumentChange>>>,
) -> Self {
Self { active, pending }
}
}
#[cfg(feature = "sync")]
#[uniffi::export]
impl SubscriptionHandle {
pub fn is_active(&self) -> bool {
self.active.load(Ordering::SeqCst)
}
pub fn cancel(&self) {
self.active.store(false, Ordering::SeqCst);
}
pub fn poll_changes(&self) -> Vec<DocumentChange> {
self.pending
.lock()
.map(|mut q| q.drain(..).collect())
.unwrap_or_default()
}
}
#[cfg(feature = "sync")]
impl Drop for SubscriptionHandle {
fn drop(&mut self) {
self.active.store(false, Ordering::SeqCst);
}
}
#[cfg(feature = "sync")]
#[derive(uniffi::Object)]
pub struct PeatNode {
sync_backend: Arc<AutomergeIrohBackend>,
storage_backend: Arc<AutomergeBackend>,
#[cfg(feature = "sync")]
node: Arc<peat_mesh::Node>,
#[cfg(all(feature = "sync", feature = "bluetooth"))]
ble_translator: Arc<peat_protocol::sync::ble_translation::BleTranslator>,
transport_manager: TransportManager,
iroh_transport: Arc<IrohTransport>,
store: Arc<AutomergeStore>,
#[allow(dead_code)] storage_path: PathBuf,
runtime: Arc<tokio::runtime::Runtime>,
#[allow(dead_code)]
cleanup_running: Arc<AtomicBool>,
#[cfg(feature = "sync")]
blob_store: std::sync::RwLock<Option<Arc<NetworkedIrohBlobStore>>>,
#[cfg(all(feature = "sync", feature = "bluetooth"))]
outbound_queue: Arc<std::sync::Mutex<std::collections::VecDeque<OutboundFrame>>>,
#[cfg(all(feature = "sync", feature = "bluetooth"))]
outbound_fanout: std::sync::Mutex<Option<peat_mesh::transport::FanoutHandle>>,
#[cfg(all(feature = "sync", feature = "bluetooth"))]
relay_seen: std::sync::Mutex<std::collections::HashMap<u64, std::time::Instant>>,
#[cfg(feature = "sync")]
water_counter: water_counter::WaterCounter,
#[cfg(feature = "sync")]
crdt_kv: crdt_kv::CrdtKvDocs,
}
#[cfg(feature = "sync")]
#[uniffi::export]
impl PeatNode {
pub fn crdt_counter_value(&self) -> i64 {
self.water_counter.value()
}
pub fn crdt_counter_increment(&self, delta: i64) -> String {
hex::encode(self.water_counter.increment(delta))
}
pub fn crdt_counter_merge(&self, hex_doc: String) -> i64 {
match hex::decode(hex_doc.trim()) {
Ok(bytes) => self.water_counter.merge(&bytes),
Err(_) => self.water_counter.value(),
}
}
pub fn crdt_counter_snapshot(&self) -> String {
hex::encode(self.water_counter.snapshot())
}
pub fn crdt_kv_put(&self, collection: String, key: String, value_json: String) -> String {
hex::encode(self.crdt_kv.put(&collection, &key, &value_json))
}
pub fn crdt_kv_all(&self, collection: String) -> String {
self.crdt_kv.all_json(&collection)
}
pub fn crdt_kv_merge(&self, collection: String, hex_doc: String) {
if let Ok(bytes) = hex::decode(hex_doc.trim()) {
self.crdt_kv.merge(&collection, &bytes);
}
}
pub fn crdt_kv_snapshot(&self, collection: String) -> String {
hex::encode(self.crdt_kv.snapshot(&collection))
}
pub fn node_id(&self) -> String {
hex::encode(self.iroh_transport.endpoint_id().as_bytes())
}
pub fn endpoint_addr(&self) -> String {
format!("{:?}", self.iroh_transport.endpoint_addr())
}
pub fn peer_count(&self) -> u32 {
self.iroh_transport.peer_count() as u32
}
pub fn connected_peers(&self) -> Vec<String> {
self.iroh_transport
.connected_peers()
.iter()
.map(|id| hex::encode(id.as_bytes()))
.collect()
}
pub fn endpoint_socket_addr(&self) -> Option<String> {
self.iroh_transport.bound_socket_addr_string()
}
pub fn start_sync(&self) -> Result<(), PeatError> {
#[cfg(target_os = "android")]
android_log("start_sync: called");
let _guard = self.runtime.enter();
#[cfg(target_os = "android")]
android_log("start_sync: runtime entered");
let result = self.runtime.block_on(async {
#[cfg(target_os = "android")]
android_log("start_sync: inside block_on");
self.storage_backend
.start_sync()
.map_err(|e| PeatError::SyncError { msg: e.to_string() })
});
#[cfg(target_os = "android")]
match &result {
Ok(_) => android_log("start_sync: SUCCESS - sync handlers spawned"),
Err(e) => android_log(&format!("start_sync: FAILED - {}", e)),
}
result
}
pub fn stop_sync(&self) -> Result<(), PeatError> {
self.runtime.block_on(async {
self.storage_backend
.stop_sync()
.map_err(|e| PeatError::SyncError { msg: e.to_string() })
})
}
pub fn sync_stats(&self) -> Result<SyncStats, PeatError> {
let stats = self
.storage_backend
.sync_stats()
.map_err(|e| PeatError::SyncError { msg: e.to_string() })?;
Ok(SyncStats {
sync_active: stats.peer_count > 0, connected_peers: self.iroh_transport.peer_count() as u32,
bytes_sent: stats.bytes_sent,
bytes_received: stats.bytes_received,
})
}
pub fn peer_transport_state(&self, peer_id: String) -> Result<PeerTransportState, PeatError> {
let mesh_peer = peat_mesh::NodeId::new(peer_id.clone());
let links = self
.transport_manager
.available_instances_for_peer(&mesh_peer)
.into_iter()
.filter_map(|transport_id| {
let transport = self.transport_manager.get_instance(&transport_id)?;
let mut state = transport.peer_link_state(&mesh_peer)?;
state.transport_id = transport_id;
Some(TransportLink::from(state))
})
.collect();
Ok(PeerTransportState { peer_id, links })
}
pub fn all_peer_transport_states(&self) -> Result<Vec<PeerTransportState>, PeatError> {
let mut peer_ids: Vec<String> = self
.iroh_transport
.connected_peers()
.iter()
.map(|id| hex::encode(id.as_bytes()))
.collect();
peer_ids.sort();
peer_ids.dedup();
let mut out = Vec::with_capacity(peer_ids.len());
for peer_id in peer_ids {
out.push(self.peer_transport_state(peer_id)?);
}
Ok(out)
}
pub fn request_sync(&self) -> Result<(), PeatError> {
if let Some(coordinator) = self.storage_backend.sync_coordinator() {
let peers = self.iroh_transport.connected_peers();
let peer_count = peers.len();
#[cfg(target_os = "android")]
android_log(&format!(
"request_sync: starting with {} connected peer(s)",
peer_count
));
let coord = Arc::clone(coordinator);
self.runtime.block_on(async {
for peer_id in peers {
match coord.sync_all_documents_with_peer(peer_id).await {
Ok(()) => {
#[cfg(target_os = "android")]
{
let peer_hex = hex::encode(peer_id.as_bytes());
android_log(&format!(
"request_sync: pushed to peer {}",
&peer_hex[..16]
));
}
}
Err(_e) => {
#[cfg(target_os = "android")]
{
let peer_hex = hex::encode(peer_id.as_bytes());
android_log(&format!(
"request_sync: FAILED for peer {}: {}",
&peer_hex[..16],
_e
));
}
}
}
}
});
#[cfg(target_os = "android")]
android_log(&format!(
"request_sync: complete ({} peer(s) attempted)",
peer_count
));
}
Ok(())
}
pub fn connect_peer(&self, peer: PeerInfo) -> Result<(), PeatError> {
let peat_peer = PeatPeerInfo {
name: peer.name,
node_id: peer.node_id,
addresses: peer.addresses,
relay_url: peer.relay_url,
};
let _guard = self.runtime.enter();
self.runtime.block_on(async {
let conn_opt = self
.iroh_transport
.connect_peer(&peat_peer)
.await
.map_err(|e| PeatError::ConnectionError { msg: e.to_string() })?;
if let Some(conn) = conn_opt {
let peer_id = conn.remote_id();
if let Some(formation_key) = self.sync_backend.formation_key() {
use peat_protocol::network::perform_initiator_handshake;
match perform_initiator_handshake(&conn, &formation_key).await {
Ok(()) => {
self.iroh_transport.emit_peer_connected(peer_id);
if let Some(coordinator) = self.storage_backend.sync_coordinator() {
let coord = Arc::clone(coordinator);
let sync_peer = peer_id;
tokio::spawn(async move {
tokio::time::sleep(tokio::time::Duration::from_millis(500))
.await;
#[cfg(target_os = "android")]
android_log(&format!(
"Triggering sync_all_documents_with_peer for {:?}",
sync_peer
));
match coord.sync_all_documents_with_peer(sync_peer).await {
Ok(()) => {
#[cfg(target_os = "android")]
android_log("sync_all_documents_with_peer: SUCCESS");
}
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"sync_all_documents_with_peer: FAILED - {}",
e
));
}
}
});
}
}
Err(e) => {
conn.close(1u32.into(), b"authentication failed");
self.iroh_transport.disconnect(&peer_id).ok();
return Err(PeatError::ConnectionError {
msg: format!("Formation handshake failed: {}", e),
});
}
}
} else {
self.iroh_transport.emit_peer_connected(peer_id);
}
}
Ok(())
})
}
pub fn disconnect_peer(&self, node_id: &str) -> Result<(), PeatError> {
let connected = self.iroh_transport.connected_peers();
for endpoint_id in connected {
if hex::encode(endpoint_id.as_bytes()) == node_id {
return self
.iroh_transport
.disconnect(&endpoint_id)
.map_err(|e| PeatError::ConnectionError { msg: e.to_string() });
}
}
Err(PeatError::ConnectionError {
msg: format!("Peer {} not found in connected peers", node_id),
})
}
pub fn put_document(
&self,
collection: &str,
doc_id: &str,
json_data: &str,
) -> Result<(), PeatError> {
let _: serde_json::Value =
serde_json::from_str(json_data).map_err(|e| PeatError::InvalidInput {
msg: format!("Invalid JSON: {}", e),
})?;
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collection);
coll.upsert(doc_id, json_data.as_bytes().to_vec())
.map_err(|e| PeatError::StorageError { msg: e.to_string() })
})
}
pub fn get_document(
&self,
collection: &str,
doc_id: &str,
) -> Result<Option<String>, PeatError> {
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collection);
match coll.get(doc_id) {
Ok(Some(bytes)) => {
let json = String::from_utf8(bytes).map_err(|e| PeatError::StorageError {
msg: format!("Invalid UTF-8: {}", e),
})?;
Ok(Some(json))
}
Ok(None) => Ok(None),
Err(e) => Err(PeatError::StorageError { msg: e.to_string() }),
}
})
}
pub fn delete_document(&self, collection: &str, doc_id: &str) -> Result<(), PeatError> {
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collection);
coll.delete(doc_id)
.map_err(|e| PeatError::StorageError { msg: e.to_string() })
})
}
pub fn list_documents(&self, collection: &str) -> Result<Vec<String>, PeatError> {
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collection);
let docs = coll
.scan()
.map_err(|e| PeatError::StorageError { msg: e.to_string() })?;
Ok(docs.into_iter().map(|(id, _)| id).collect())
})
}
pub fn sync_document(&self, collection: &str, doc_id: &str) -> Result<(), PeatError> {
let doc_key = format!("{}:{}", collection, doc_id);
self.runtime.block_on(async {
let backend = &self.storage_backend;
backend
.sync_document(&doc_key)
.await
.map_err(|e| PeatError::SyncError { msg: e.to_string() })
})
}
pub fn subscribe(
&self,
callback: Box<dyn DocumentCallback>,
) -> Result<Arc<SubscriptionHandle>, PeatError> {
let change_rx = self.store.subscribe_to_changes_with_origin();
let active = Arc::new(AtomicBool::new(true));
let active_clone = Arc::clone(&active);
let callback = Arc::new(callback);
self.runtime.spawn(async move {
let mut rx = change_rx;
while active_clone.load(Ordering::SeqCst) {
tokio::select! {
result = rx.recv() => {
match result {
Ok(doc_change) => {
let doc_key = doc_change.key;
let change = if let Some((collection, doc_id)) = doc_key.split_once(':') {
DocumentChange {
collection: collection.to_string(),
doc_id: doc_id.to_string(),
change_type: ChangeType::Upsert,
}
} else {
DocumentChange {
collection: "default".to_string(),
doc_id: doc_key,
change_type: ChangeType::Upsert,
}
};
callback.on_change(change);
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
callback.on_error(format!("Lagged {} messages", n));
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
callback.on_error("Document change channel closed".to_string());
break;
}
}
}
_ = tokio::time::sleep(tokio::time::Duration::from_millis(100)) => {
if !active_clone.load(Ordering::SeqCst) {
break;
}
}
}
}
});
Ok(Arc::new(SubscriptionHandle::new(active)))
}
pub fn subscribe_poll(&self) -> Result<Arc<SubscriptionHandle>, PeatError> {
let change_rx = self.store.subscribe_to_changes_with_origin();
let active = Arc::new(AtomicBool::new(true));
let active_clone = Arc::clone(&active);
let pending = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::<
DocumentChange,
>::new()));
let pending_clone = Arc::clone(&pending);
self.runtime.spawn(async move {
let mut rx = change_rx;
while active_clone.load(Ordering::SeqCst) {
tokio::select! {
result = rx.recv() => {
match result {
Ok(doc_change) => {
let doc_key = doc_change.key;
let change = if let Some((collection, doc_id)) = doc_key.split_once(':') {
DocumentChange {
collection: collection.to_string(),
doc_id: doc_id.to_string(),
change_type: ChangeType::Upsert,
}
} else {
DocumentChange {
collection: "default".to_string(),
doc_id: doc_key,
change_type: ChangeType::Upsert,
}
};
if let Ok(mut q) = pending_clone.lock() {
q.push_back(change);
}
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
}
}
_ = tokio::time::sleep(tokio::time::Duration::from_millis(100)) => {
if !active_clone.load(Ordering::SeqCst) {
break;
}
}
}
}
});
Ok(Arc::new(SubscriptionHandle::new_with_queue(
active, pending,
)))
}
}
#[cfg(feature = "sync")]
#[uniffi::export]
pub fn create_node(config: NodeConfig) -> Result<Arc<PeatNode>, PeatError> {
use std::time::Instant;
let total_start = Instant::now();
if config.app_id.is_empty() {
return Err(PeatError::InvalidInput {
msg: "app_id cannot be empty".to_string(),
});
}
if config.shared_key.is_empty() {
return Err(PeatError::InvalidInput {
msg: "shared_key cannot be empty".to_string(),
});
}
fn get_rss_kb() -> u64 {
std::fs::read_to_string("/proc/self/status")
.ok()
.and_then(|s| {
s.lines()
.find(|l| l.starts_with("VmRSS:"))
.and_then(|l| l.split_whitespace().nth(1))
.and_then(|v| v.parse().ok())
})
.unwrap_or(0)
}
#[cfg(target_os = "android")]
android_log(&format!("[MEM] Before runtime: {} kB", get_rss_kb()));
let phase_start = Instant::now();
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(4)
.enable_all()
.build()
.map_err(|e| PeatError::SyncError {
msg: format!("Failed to create runtime: {}", e),
})?;
let runtime_ms = phase_start.elapsed().as_millis();
#[cfg(target_os = "android")]
android_log(&format!("[TIMING] Runtime creation: {}ms", runtime_ms));
#[cfg(target_os = "android")]
android_log(&format!("[MEM] After runtime: {} kB", get_rss_kb()));
#[cfg(not(target_os = "android"))]
eprintln!("[Peat TIMING] Runtime creation: {}ms", runtime_ms);
let bind_addr: SocketAddr = config
.bind_address
.as_deref()
.unwrap_or("0.0.0.0:0")
.parse()
.map_err(|e| PeatError::InvalidInput {
msg: format!("Invalid bind address: {}", e),
})?;
let storage_path = PathBuf::from(&config.storage_path);
std::fs::create_dir_all(&storage_path).map_err(|e| PeatError::StorageError {
msg: format!("Failed to create storage directory: {}", e),
})?;
let phase_start = Instant::now();
let seed = format!("{}/{}", config.app_id, config.storage_path);
let storage_path_for_store = storage_path.clone();
let (store, transport, store_ms, transport_ms) = runtime.block_on(async {
let store_start = Instant::now();
let transport_start = Instant::now();
let store_handle = tokio::task::spawn_blocking(move || {
let mut last_err = None;
for _ in 0..60u32 {
match AutomergeStore::open(&storage_path_for_store) {
Ok(s) => return (Ok(s), store_start.elapsed().as_millis()),
Err(e) => {
last_err = Some(e);
std::thread::sleep(std::time::Duration::from_millis(500));
}
}
}
(Err(last_err.unwrap()), store_start.elapsed().as_millis())
});
let transport_future = async {
let result = IrohTransport::from_seed_with_discovery_at_addr(&seed, bind_addr).await;
(result, transport_start.elapsed().as_millis())
};
let (store_result, transport_result) = tokio::join!(store_handle, transport_future);
let (store_inner, store_elapsed) = store_result.map_err(|e| PeatError::StorageError {
msg: format!("Store task panicked: {}", e),
})?;
let store = store_inner.map_err(|e| PeatError::StorageError {
msg: format!("Failed to open store: {}", e),
})?;
#[cfg(target_os = "android")]
android_log(&format!(
"[MEM] After store open: {} kB (store {}ms)",
get_rss_kb(),
store_elapsed
));
let (transport_inner, transport_elapsed) = transport_result;
let transport = transport_inner.map_err(|e| PeatError::ConnectionError {
msg: format!("Failed to create transport with mDNS: {}", e),
})?;
#[cfg(target_os = "android")]
android_log(&format!(
"[MEM] After iroh transport: {} kB (transport {}ms)",
get_rss_kb(),
transport_elapsed
));
Ok::<_, PeatError>((
Arc::new(store),
Arc::new(transport),
store_elapsed,
transport_elapsed,
))
})?;
let parallel_total_ms = phase_start.elapsed().as_millis();
#[cfg(target_os = "android")]
{
android_log(&format!("[TIMING] Store open: {}ms", store_ms));
android_log(&format!(
"[TIMING] Transport create (with mDNS): {}ms",
transport_ms
));
android_log(&format!(
"[TIMING] Parallel total (max of above): {}ms",
parallel_total_ms
));
}
#[cfg(not(target_os = "android"))]
{
eprintln!("[Peat TIMING] Store open: {}ms", store_ms);
eprintln!(
"[Peat TIMING] Transport create (with mDNS): {}ms",
transport_ms
);
eprintln!(
"[Peat TIMING] Parallel total (max of above): {}ms",
parallel_total_ms
);
}
let storage_backend = Arc::new(AutomergeBackend::with_transport(
Arc::clone(&store),
Arc::clone(&transport),
));
let sync_backend = Arc::new(AutomergeIrohBackend::new(
Arc::clone(&storage_backend),
Arc::clone(&transport),
));
let mut event_rx = transport.subscribe_peer_events();
let phase_start = Instant::now();
let backend_config = BackendConfig {
app_id: config.app_id.clone(),
persistence_dir: storage_path.clone(),
shared_key: Some(config.shared_key.clone()),
transport: TransportConfig::default(),
extra: std::collections::HashMap::new(),
};
runtime.block_on(async {
sync_backend
.initialize(backend_config)
.await
.map_err(|e| PeatError::SyncError {
msg: format!("Failed to initialize sync backend: {}", e),
})
})?;
let sync_init_ms = phase_start.elapsed().as_millis();
#[cfg(target_os = "android")]
{
android_log(&format!("[TIMING] Sync backend init: {}ms", sync_init_ms));
android_log("=== sync_backend.initialize() completed successfully ===");
}
#[cfg(not(target_os = "android"))]
eprintln!("[Peat TIMING] Sync backend init: {}ms", sync_init_ms);
let cleanup_running = Arc::new(AtomicBool::new(true));
let cleanup_flag = Arc::clone(&cleanup_running);
let runtime_arc = Arc::new(runtime);
let transport_for_cleanup = Arc::clone(&transport);
#[cfg(target_os = "android")]
android_log("Starting peer event listener task (Issue #275)");
runtime_arc.spawn(async move {
#[cfg(target_os = "android")]
android_log("Peer event listener task running");
while cleanup_flag.load(Ordering::Relaxed) {
tokio::select! {
event_result = event_rx.recv() => {
match event_result {
Some(event) => {
#[cfg(target_os = "android")]
android_log(&format!("Received transport peer event: {:?}", event));
match event {
TransportPeerEvent::Connected { endpoint_id, .. } => {
let peer_id = hex::encode(endpoint_id.as_bytes());
#[cfg(target_os = "android")]
android_log(&format!("Processing Connected event for peer: {}", peer_id));
notify_peer_connected(&peer_id);
}
TransportPeerEvent::Disconnected { endpoint_id, reason } => {
let peer_id = hex::encode(endpoint_id.as_bytes());
#[cfg(target_os = "android")]
android_log(&format!("Processing Disconnected event for peer: {} reason: {}", peer_id, reason));
notify_peer_disconnected(&peer_id, &reason);
}
}
}
None => {
#[cfg(target_os = "android")]
android_log("Event channel closed, exiting peer event listener");
break;
}
}
}
_ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {
let count = transport_for_cleanup.peer_count();
#[cfg(target_os = "android")]
android_log(&format!("Periodic cleanup tick - peer count: {}", count));
}
}
}
#[cfg(target_os = "android")]
android_log("Peer event listener task exiting");
});
let storage_backend = sync_backend.storage_backend();
let mut tm_config = TransportManagerConfig::default();
if let Some(ref transport_config) = config.transport {
if let Some(ref prefs) = transport_config.transport_preference {
let policy = TransportPolicy::new("ffi-config").primary(prefs.clone());
tm_config.default_policy = Some(policy);
}
if let Some(ref routes_json) = transport_config.collection_routes_json {
match serde_json::from_str::<CollectionRouteTable>(routes_json) {
Ok(table) => {
tm_config.collection_routes = table;
}
Err(e) => {
eprintln!("[Peat] Failed to parse collection_routes_json: {}", e);
}
}
}
}
let mut transport_manager = TransportManager::new(tm_config);
let iroh_mesh_transport = Arc::new(IrohMeshTransport::new(Arc::clone(&transport), Vec::new()));
let iroh_as_transport: Arc<dyn Transport> = iroh_mesh_transport.clone();
transport_manager.register(iroh_as_transport.clone());
let iroh_instance = TransportInstance::new(
"iroh-primary",
TransportType::Quic,
TransportCapabilities::quic(),
)
.with_description("Primary Iroh/QUIC transport");
transport_manager.register_instance(iroh_instance, iroh_as_transport);
#[cfg(feature = "bluetooth")]
if let Some(ref transport_config) = config.transport {
if transport_config.enable_ble {
#[cfg(target_os = "android")]
{
use peat_btle::platform::android::AndroidAdapter;
use peat_btle::{BleConfig, BluetoothLETransport};
android_log("BLE transport requested - initializing AndroidAdapter stub");
let iroh_endpoint_id = transport.endpoint_id();
let iroh_key_bytes = iroh_endpoint_id.as_bytes();
let ble_node_id = peat_btle::NodeId::new(u32::from_be_bytes([
iroh_key_bytes[28],
iroh_key_bytes[29],
iroh_key_bytes[30],
iroh_key_bytes[31],
]));
let ble_config = BleConfig::new(ble_node_id);
let adapter = AndroidAdapter::new_stub();
let btle = BluetoothLETransport::new(ble_config, adapter);
let ble_transport = Arc::new(PeatBleTransport::new(btle));
let ble_as_transport: Arc<dyn Transport> = ble_transport.clone();
transport_manager.register(ble_as_transport.clone());
let ble_instance = TransportInstance::new(
"ble-primary",
TransportType::BluetoothLE,
TransportCapabilities::bluetooth_le(),
)
.with_description("Primary BLE transport (Android)");
transport_manager.register_instance(ble_instance, ble_as_transport);
*ANDROID_BLE_TRANSPORT.lock().unwrap() = Some(ble_transport);
android_log("BLE transport registered as PACE instance 'ble-primary'");
}
#[cfg(not(target_os = "android"))]
{
#[cfg(target_os = "linux")]
{
use peat_btle::platform::linux::BluerAdapter;
use peat_btle::{BleAdapter, BleConfig, BluetoothLETransport, PowerProfile};
let power_profile = match transport_config.ble_power_profile.as_deref() {
Some("aggressive") => PowerProfile::Aggressive,
Some("low_power") => PowerProfile::LowPower,
_ => PowerProfile::Balanced,
};
let iroh_endpoint_id = transport.endpoint_id();
let iroh_key_bytes = iroh_endpoint_id.as_bytes();
let ble_node_id = peat_btle::NodeId::new(u32::from_be_bytes([
iroh_key_bytes[28],
iroh_key_bytes[29],
iroh_key_bytes[30],
iroh_key_bytes[31],
]));
let mut ble_config = BleConfig::new(ble_node_id);
ble_config.power_profile = power_profile;
if let Some(ref mesh_id) = transport_config.ble_mesh_id {
ble_config.mesh.mesh_id = mesh_id.clone();
}
use peat_protocol::transport::MeshTransport;
match runtime_arc.block_on(async {
let mut adapter = BluerAdapter::new().await?;
adapter.init(&ble_config).await?;
adapter.register_gatt_service().await?;
let btle = BluetoothLETransport::new(ble_config, adapter);
let ble_transport = Arc::new(PeatBleTransport::new(btle));
ble_transport.start().await.map_err(|e| {
peat_btle::BleError::PlatformError(format!(
"Failed to start BLE transport: {}",
e
))
})?;
Ok::<_, peat_btle::BleError>(ble_transport)
}) {
Ok(ble_transport) => {
let ble_as_transport: Arc<dyn Transport> = ble_transport.clone();
transport_manager.register(ble_as_transport.clone());
let ble_instance = TransportInstance::new(
"ble-primary",
TransportType::BluetoothLE,
TransportCapabilities::bluetooth_le(),
)
.with_description("Primary BLE transport");
transport_manager.register_instance(ble_instance, ble_as_transport);
eprintln!(
"[Peat] BLE transport registered as PACE instance 'ble-primary'"
);
}
Err(e) => {
eprintln!("[Peat] Failed to initialize BLE adapter: {} (continuing without BLE)", e);
}
}
}
#[cfg(not(target_os = "linux"))]
eprintln!(
"[Peat] BLE transport requested but not yet implemented for this platform"
);
}
}
}
let total_ms = total_start.elapsed().as_millis();
#[cfg(target_os = "android")]
android_log(&format!(
"[TIMING] === TOTAL create_node: {}ms ===",
total_ms
));
#[cfg(not(target_os = "android"))]
eprintln!("[Peat TIMING] === TOTAL create_node: {}ms ===", total_ms);
#[cfg(feature = "sync")]
let node = {
use peat_mesh::sync::traits::DataSyncBackend;
let backend_dyn: Arc<dyn DataSyncBackend> = sync_backend.clone();
Arc::new(peat_mesh::Node::new(backend_dyn))
};
#[cfg(all(feature = "sync", feature = "bluetooth"))]
let ble_translator = {
use peat_protocol::sync::ble_translation::BleTranslator;
Arc::new(BleTranslator::with_defaults())
};
let node_arc = Arc::new(PeatNode {
sync_backend,
storage_backend,
#[cfg(feature = "sync")]
node,
#[cfg(all(feature = "sync", feature = "bluetooth"))]
ble_translator,
transport_manager,
iroh_transport: transport,
store,
#[cfg(feature = "sync")]
water_counter: water_counter::WaterCounter::load_or_init(
storage_path.join("water.automerge"),
),
#[cfg(feature = "sync")]
crdt_kv: crdt_kv::CrdtKvDocs::new(storage_path.clone()),
storage_path,
runtime: runtime_arc,
cleanup_running,
#[cfg(feature = "sync")]
blob_store: std::sync::RwLock::new(None),
#[cfg(all(feature = "sync", feature = "bluetooth"))]
outbound_queue: Arc::new(std::sync::Mutex::new(std::collections::VecDeque::new())),
#[cfg(all(feature = "sync", feature = "bluetooth"))]
outbound_fanout: std::sync::Mutex::new(None),
#[cfg(all(feature = "sync", feature = "bluetooth"))]
relay_seen: std::sync::Mutex::new(std::collections::HashMap::new()),
});
#[cfg(target_os = "android")]
set_global_node_handle(&node_arc);
Ok(node_arc)
}
#[cfg(feature = "sync")]
impl From<anyhow::Error> for PeatError {
fn from(e: anyhow::Error) -> Self {
PeatError::SyncError { msg: e.to_string() }
}
}
pub mod collections {
pub const CELLS: &str = "cells";
pub const TRACKS: &str = "tracks";
pub const NODES: &str = "nodes";
pub const CAPABILITIES: &str = "capabilities";
pub const COMMANDS: &str = "commands";
pub const MARKERS: &str = "markers";
}
const TOMBSTONE_PLACEHOLDER_TYPE: &str = "a-u-G";
#[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Enum)]
pub enum CellStatus {
Active,
Forming,
Degraded,
Offline,
}
impl CellStatus {
fn from_str(s: &str) -> Self {
match s.to_uppercase().as_str() {
"ACTIVE" => Self::Active,
"FORMING" => Self::Forming,
"DEGRADED" => Self::Degraded,
"OFFLINE" => Self::Offline,
_ => Self::Offline,
}
}
fn as_str(&self) -> &'static str {
match self {
Self::Active => "ACTIVE",
Self::Forming => "FORMING",
Self::Degraded => "DEGRADED",
Self::Offline => "OFFLINE",
}
}
}
#[derive(Debug, Clone, uniffi::Record)]
pub struct CellInfo {
pub id: String,
pub name: String,
pub status: CellStatus,
pub node_count: u32,
pub center_lat: f64,
pub center_lon: f64,
pub capabilities: Vec<String>,
pub formation_id: Option<String>,
pub leader_id: Option<String>,
pub last_update: i64,
pub scenario_command: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Enum)]
pub enum TrackCategory {
Person,
Vehicle,
Aircraft,
Vessel,
Installation,
Unknown,
}
impl TrackCategory {
fn from_str(s: &str) -> Self {
match s.to_uppercase().as_str() {
"PERSON" => Self::Person,
"VEHICLE" => Self::Vehicle,
"AIRCRAFT" => Self::Aircraft,
"VESSEL" => Self::Vessel,
"INSTALLATION" => Self::Installation,
_ => Self::Unknown,
}
}
fn as_str(&self) -> &'static str {
match self {
Self::Person => "PERSON",
Self::Vehicle => "VEHICLE",
Self::Aircraft => "AIRCRAFT",
Self::Vessel => "VESSEL",
Self::Installation => "INSTALLATION",
Self::Unknown => "UNKNOWN",
}
}
}
#[derive(Debug, Clone, uniffi::Record)]
pub struct TrackInfo {
pub id: String,
pub source_node: String,
pub cell_id: Option<String>,
pub formation_id: Option<String>,
pub lat: f64,
pub lon: f64,
pub hae: Option<f64>,
pub cep: Option<f64>,
pub heading: Option<f64>,
pub speed: Option<f64>,
pub classification: String,
pub confidence: f64,
pub category: TrackCategory,
pub created_at: i64,
pub last_update: i64,
pub attributes: HashMap<String, String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Enum)]
pub enum NodeStatus {
Ready,
Active,
Degraded,
Offline,
Loading,
}
impl NodeStatus {
fn from_str(s: &str) -> Self {
match s.to_uppercase().as_str() {
"READY" => Self::Ready,
"ACTIVE" => Self::Active,
"DEGRADED" => Self::Degraded,
"OFFLINE" => Self::Offline,
"LOADING" => Self::Loading,
_ => Self::Offline,
}
}
pub fn as_str(&self) -> &'static str {
match self {
Self::Ready => "READY",
Self::Active => "ACTIVE",
Self::Degraded => "DEGRADED",
Self::Offline => "OFFLINE",
Self::Loading => "LOADING",
}
}
}
#[derive(Debug, Clone, uniffi::Record)]
pub struct NodeInfo {
pub id: String,
pub node_type: String,
pub name: String,
pub status: NodeStatus,
pub lat: f64,
pub lon: f64,
pub hae: Option<f64>,
pub readiness: f64,
pub capabilities: Vec<String>,
pub cell_id: Option<String>,
pub battery_percent: Option<i32>,
pub heart_rate: Option<i32>,
pub last_heartbeat: i64,
}
#[derive(Debug, Clone, uniffi::Record)]
pub struct MarkerInfo {
pub uid: String,
pub marker_type: String,
pub lat: f64,
pub lon: f64,
pub hae: Option<f64>,
pub ts: i64,
pub callsign: Option<String>,
pub color: Option<i32>,
pub cell_id: Option<String>,
pub deleted: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Enum)]
pub enum CommandStatus {
Pending,
Executing,
Completed,
Failed,
Cancelled,
}
impl CommandStatus {
fn from_str(s: &str) -> Self {
match s.to_uppercase().as_str() {
"PENDING" => Self::Pending,
"EXECUTING" => Self::Executing,
"COMPLETED" => Self::Completed,
"FAILED" => Self::Failed,
"CANCELLED" => Self::Cancelled,
_ => Self::Pending,
}
}
fn as_str(&self) -> &'static str {
match self {
Self::Pending => "PENDING",
Self::Executing => "EXECUTING",
Self::Completed => "COMPLETED",
Self::Failed => "FAILED",
Self::Cancelled => "CANCELLED",
}
}
}
#[derive(Debug, Clone, uniffi::Record)]
pub struct CommandInfo {
pub id: String,
pub command_type: String,
pub target_id: String,
pub parameters: String,
pub priority: u8,
pub status: CommandStatus,
pub originator: String,
pub created_at: i64,
pub last_update: i64,
}
#[cfg(feature = "sync")]
#[uniffi::export]
impl PeatNode {
pub fn get_cells(&self) -> Result<Vec<CellInfo>, PeatError> {
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collections::CELLS);
let docs = coll
.scan()
.map_err(|e| PeatError::StorageError { msg: e.to_string() })?;
let mut cells = Vec::new();
for (id, data) in docs {
if let Ok(json) = String::from_utf8(data) {
if let Ok(cell) = parse_cell_json(&id, &json) {
cells.push(cell);
}
}
}
Ok(cells)
})
}
pub fn get_cell(&self, cell_id: &str) -> Result<Option<CellInfo>, PeatError> {
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collections::CELLS);
match coll.get(cell_id) {
Ok(Some(data)) => {
let json = String::from_utf8(data).map_err(|e| PeatError::StorageError {
msg: format!("Invalid UTF-8: {}", e),
})?;
let cell = parse_cell_json(cell_id, &json)?;
Ok(Some(cell))
}
Ok(None) => Ok(None),
Err(e) => Err(PeatError::StorageError { msg: e.to_string() }),
}
})
}
pub fn put_cell(&self, cell: CellInfo) -> Result<(), PeatError> {
let json = serialize_cell_json(&cell)?;
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collections::CELLS);
coll.upsert(&cell.id, json.into_bytes())
.map_err(|e| PeatError::StorageError { msg: e.to_string() })
})
}
pub fn get_tracks(&self) -> Result<Vec<TrackInfo>, PeatError> {
use peat_mesh::sync::types::Query;
self.runtime.block_on(async {
let docs = self
.node
.query(collections::TRACKS, &Query::All)
.await
.map_err(|e| PeatError::StorageError { msg: e.to_string() })?;
let mut tracks = Vec::with_capacity(docs.len());
for doc in docs {
if let Some(id) = doc.id.clone() {
if let Ok(track) = track_from_document(&id, &doc) {
tracks.push(track);
}
}
}
Ok(tracks)
})
}
pub fn get_track(&self, track_id: &str) -> Result<Option<TrackInfo>, PeatError> {
self.runtime.block_on(async {
let id = track_id.to_string();
match self.node.get(collections::TRACKS, &id).await {
Ok(Some(doc)) => Ok(Some(track_from_document(track_id, &doc)?)),
Ok(None) => Ok(None),
Err(e) => Err(PeatError::StorageError { msg: e.to_string() }),
}
})
}
pub fn put_track(&self, track: TrackInfo) -> Result<(), PeatError> {
let doc = track_to_document(&track)?;
self.runtime.block_on(async {
self.node
.publish(collections::TRACKS, doc)
.await
.map(|_id| ())
.map_err(|e| PeatError::StorageError { msg: e.to_string() })
})
}
pub fn get_nodes(&self) -> Result<Vec<NodeInfo>, PeatError> {
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collections::NODES);
let docs = coll
.scan()
.map_err(|e| PeatError::StorageError { msg: e.to_string() })?;
let mut nodes = Vec::new();
for (id, data) in docs {
if let Ok(json) = String::from_utf8(data) {
if let Ok(node) = parse_node_json(&id, &json) {
nodes.push(node);
}
}
}
Ok(nodes)
})
}
pub fn put_node(&self, node: NodeInfo) -> Result<(), PeatError> {
let json = serialize_node_json(&node)?;
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collections::NODES);
coll.upsert(&node.id, json.into_bytes())
.map_err(|e| PeatError::StorageError { msg: e.to_string() })
})
}
pub fn get_markers(&self) -> Result<Vec<MarkerInfo>, PeatError> {
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collections::MARKERS);
let docs = coll
.scan()
.map_err(|e| PeatError::StorageError { msg: e.to_string() })?;
let mut markers = Vec::new();
for (id, data) in docs {
let json_str = String::from_utf8_lossy(&data);
match parse_marker_publish_json(&id, &json_str) {
Ok(m) => markers.push(m),
Err(_) => {
}
}
}
Ok(markers)
})
}
pub fn put_marker(&self, marker: MarkerInfo) -> Result<(), PeatError> {
let json = serialize_marker_json(&marker)?;
let uid = marker.uid.clone();
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collections::MARKERS);
coll.upsert(&uid, json.into_bytes())
.map_err(|e| PeatError::StorageError { msg: e.to_string() })
})
}
pub fn get_commands(&self) -> Result<Vec<CommandInfo>, PeatError> {
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collections::COMMANDS);
let docs = coll
.scan()
.map_err(|e| PeatError::StorageError { msg: e.to_string() })?;
let mut commands = Vec::new();
for (id, data) in docs {
if let Ok(json) = String::from_utf8(data) {
if let Ok(cmd) = parse_command_json(&id, &json) {
commands.push(cmd);
}
}
}
Ok(commands)
})
}
pub fn put_command(&self, command: CommandInfo) -> Result<(), PeatError> {
let json = serialize_command_json(&command)?;
self.runtime.block_on(async {
let backend = &self.storage_backend;
let coll = backend.collection(collections::COMMANDS);
coll.upsert(&command.id, json.into_bytes())
.map_err(|e| PeatError::StorageError { msg: e.to_string() })
})
}
}
#[cfg(feature = "sync")]
impl PeatNode {
pub fn enable_blob_transfer(
&self,
bind_addr: Option<std::net::SocketAddr>,
) -> Result<(), PeatError> {
let blob_dir = self.storage_path.join("blobs");
std::fs::create_dir_all(&blob_dir).map_err(|e| PeatError::StorageError {
msg: format!("Failed to create blob dir {:?}: {}", blob_dir, e),
})?;
let config = PeatMeshIrohConfig {
bind_addr,
..Default::default()
};
let store = self
.runtime
.block_on(NetworkedIrohBlobStore::from_config(blob_dir, &config))
.map_err(|e| PeatError::SyncError {
msg: format!("Failed to create blob store: {}", e),
})?;
#[cfg(target_os = "android")]
android_log(&format!(
"Blob transfer enabled. EndpointId={}",
store.endpoint_id().fmt_short()
));
let mut slot = self.blob_store.write().map_err(|_| PeatError::SyncError {
msg: "blob_store lock poisoned".to_string(),
})?;
*slot = Some(store);
Ok(())
}
pub fn blob_add_peer(&self, peer_id_hex: &str, address: &str) -> Result<(), PeatError> {
let store_guard = self.blob_store.read().map_err(|_| PeatError::SyncError {
msg: "blob_store lock poisoned".to_string(),
})?;
let store = store_guard.as_ref().ok_or(PeatError::SyncError {
msg: "blob transfer not enabled".to_string(),
})?;
let store_clone = Arc::clone(store);
let hex = peer_id_hex.to_string();
let addr = address.to_string();
self.runtime
.block_on(async move { store_clone.add_peer_from_hex(&hex, &addr).await })
.map_err(|e| PeatError::SyncError {
msg: format!("blob_add_peer: {}", e),
})?;
#[cfg(target_os = "android")]
android_log(&format!(
"Blob peer added: {} at {}",
&peer_id_hex[..16.min(peer_id_hex.len())],
address
));
Ok(())
}
pub fn blob_put(&self, data: &[u8], content_type: &str) -> Result<String, PeatError> {
let store_guard = self.blob_store.read().map_err(|_| PeatError::SyncError {
msg: "blob_store lock poisoned".to_string(),
})?;
let store = store_guard.as_ref().ok_or(PeatError::SyncError {
msg: "blob transfer not enabled".to_string(),
})?;
let metadata = BlobMetadata {
content_type: Some(content_type.to_string()),
name: None,
custom: Default::default(),
};
let store_clone = Arc::clone(store);
let data_vec = data.to_vec();
let token = self
.runtime
.block_on(async move {
store_clone
.create_blob_from_bytes(&data_vec, metadata)
.await
})
.map_err(|e| PeatError::StorageError {
msg: format!("blob put failed: {}", e),
})?;
Ok(token.hash.as_hex().to_string())
}
pub fn blob_get(&self, hash_hex: &str) -> Result<Vec<u8>, PeatError> {
let store_guard = self.blob_store.read().map_err(|_| PeatError::SyncError {
msg: "blob_store lock poisoned".to_string(),
})?;
let store = store_guard.as_ref().ok_or(PeatError::SyncError {
msg: "blob transfer not enabled".to_string(),
})?;
let token = BlobToken {
hash: peat_mesh::storage::BlobHash(hash_hex.to_string()),
size_bytes: 0, metadata: BlobMetadata {
content_type: None,
name: None,
custom: Default::default(),
},
};
let store_clone = Arc::clone(store);
let handle = self
.runtime
.block_on(async move { store_clone.fetch_blob_simple(&token).await })
.map_err(|e| PeatError::StorageError {
msg: format!("blob fetch failed: {}", e),
})?;
std::fs::read(&handle.path).map_err(|e| PeatError::StorageError {
msg: format!("blob read failed: {}", e),
})
}
pub fn blob_exists_locally(&self, hash_hex: &str) -> bool {
let store_guard = match self.blob_store.read() {
Ok(g) => g,
Err(_) => return false,
};
let store = match store_guard.as_ref() {
Some(s) => s,
None => return false,
};
let hash = peat_mesh::storage::BlobHash(hash_hex.to_string());
store.blob_exists_locally(&hash)
}
pub fn blob_endpoint_id(&self) -> Option<String> {
let store_guard = self.blob_store.read().ok()?;
let store = store_guard.as_ref()?;
Some(hex::encode(store.endpoint_id().as_bytes()))
}
pub fn blob_bound_addr(&self) -> Option<String> {
let store_guard = self.blob_store.read().ok()?;
let store = store_guard.as_ref()?;
store.bound_addr_string()
}
}
fn parse_cell_json(id: &str, json: &str) -> Result<CellInfo, PeatError> {
let root: serde_json::Value =
serde_json::from_str(json).map_err(|e| PeatError::InvalidInput {
msg: format!("Invalid JSON: {}", e),
})?;
let v = match root.get("fields") {
Some(f) if f.is_object() => f,
_ => &root,
};
Ok(CellInfo {
id: id.to_string(),
name: v["name"].as_str().unwrap_or(id).to_string(),
status: CellStatus::from_str(v["status"].as_str().unwrap_or("OFFLINE")),
node_count: v["node_count"].as_u64().unwrap_or(0) as u32,
center_lat: v["center_lat"].as_f64().unwrap_or(0.0),
center_lon: v["center_lon"].as_f64().unwrap_or(0.0),
capabilities: v["capabilities"]
.as_array()
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect()
})
.unwrap_or_default(),
formation_id: v["formation_id"].as_str().map(|s| s.to_string()),
leader_id: v["leader_id"].as_str().map(|s| s.to_string()),
last_update: v["last_update"].as_i64().unwrap_or(0),
scenario_command: v["scenario_command"].as_str().map(|s| s.to_string()),
})
}
fn serialize_cell_json(cell: &CellInfo) -> Result<String, PeatError> {
let v = serde_json::json!({
"name": cell.name,
"status": cell.status.as_str(),
"node_count": cell.node_count,
"center_lat": cell.center_lat,
"center_lon": cell.center_lon,
"capabilities": cell.capabilities,
"formation_id": cell.formation_id,
"leader_id": cell.leader_id,
"last_update": cell.last_update,
"scenario_command": cell.scenario_command,
});
serde_json::to_string(&v).map_err(|e| PeatError::EncodingError { msg: e.to_string() })
}
fn track_to_document(track: &TrackInfo) -> Result<peat_mesh::sync::types::Document, PeatError> {
let json = serialize_track_json(track)?;
let value: serde_json::Value =
serde_json::from_str(&json).map_err(|e| PeatError::EncodingError {
msg: format!("track_to_document: re-parse failed: {}", e),
})?;
let fields: std::collections::HashMap<String, serde_json::Value> = match value {
serde_json::Value::Object(map) => map.into_iter().collect(),
_ => std::collections::HashMap::new(),
};
Ok(peat_mesh::sync::types::Document {
id: Some(track.id.clone()),
fields,
updated_at: std::time::SystemTime::now(),
})
}
fn track_from_document(
id: &str,
doc: &peat_mesh::sync::types::Document,
) -> Result<TrackInfo, PeatError> {
let body: serde_json::Map<String, serde_json::Value> = doc
.fields
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let json = serde_json::to_string(&serde_json::Value::Object(body))
.map_err(|e| PeatError::EncodingError { msg: e.to_string() })?;
parse_track_json(id, &json)
}
fn parse_track_json(id: &str, json: &str) -> Result<TrackInfo, PeatError> {
let v: serde_json::Value = serde_json::from_str(json).map_err(|e| PeatError::InvalidInput {
msg: format!("Invalid JSON: {}", e),
})?;
Ok(TrackInfo {
id: id.to_string(),
source_node: v["source_node"].as_str().unwrap_or("unknown").to_string(),
cell_id: v["cell_id"].as_str().map(|s| s.to_string()),
formation_id: v["formation_id"].as_str().map(|s| s.to_string()),
lat: v["lat"].as_f64().unwrap_or(0.0),
lon: v["lon"].as_f64().unwrap_or(0.0),
hae: v["hae"].as_f64(),
cep: v["cep"].as_f64(),
heading: v["heading"].as_f64(),
speed: v["speed"].as_f64(),
classification: v["classification"].as_str().unwrap_or("a-u-G").to_string(),
confidence: v["confidence"].as_f64().unwrap_or(0.5),
category: TrackCategory::from_str(v["category"].as_str().unwrap_or("UNKNOWN")),
created_at: v["created_at"].as_i64().unwrap_or(0),
last_update: v["last_update"].as_i64().unwrap_or(0),
attributes: v["attributes"]
.as_object()
.map(|obj| {
obj.iter()
.map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string()))
.collect()
})
.unwrap_or_default(),
})
}
fn serialize_track_json(track: &TrackInfo) -> Result<String, PeatError> {
let v = serde_json::json!({
"source_node": track.source_node,
"cell_id": track.cell_id,
"formation_id": track.formation_id,
"lat": track.lat,
"lon": track.lon,
"hae": track.hae,
"cep": track.cep,
"heading": track.heading,
"speed": track.speed,
"classification": track.classification,
"confidence": track.confidence,
"category": track.category.as_str(),
"created_at": track.created_at,
"last_update": track.last_update,
"attributes": track.attributes,
});
serde_json::to_string(&v).map_err(|e| PeatError::EncodingError { msg: e.to_string() })
}
fn parse_node_json(id: &str, json: &str) -> Result<NodeInfo, PeatError> {
let root: serde_json::Value =
serde_json::from_str(json).map_err(|e| PeatError::InvalidInput {
msg: format!("Invalid JSON: {}", e),
})?;
let v = match root.get("fields") {
Some(f) if f.is_object() => f,
_ => &root,
};
Ok(NodeInfo {
id: id.to_string(),
node_type: v["node_type"].as_str().unwrap_or("unknown").to_string(),
name: v["name"].as_str().unwrap_or(id).to_string(),
status: NodeStatus::from_str(v["status"].as_str().unwrap_or("OFFLINE")),
lat: v["lat"].as_f64().unwrap_or(0.0),
lon: v["lon"].as_f64().unwrap_or(0.0),
hae: v["hae"].as_f64(),
readiness: v["readiness"].as_f64().unwrap_or(0.0),
capabilities: v["capabilities"]
.as_array()
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect()
})
.unwrap_or_default(),
cell_id: v["cell_id"].as_str().map(|s| s.to_string()),
battery_percent: parse_battery_percent(&v["battery_percent"]),
heart_rate: parse_heart_rate(&v["heart_rate"]),
last_heartbeat: v["last_heartbeat"].as_i64().unwrap_or(0),
})
}
fn parse_node_publish_json(json_str: &str) -> Result<NodeInfo, PeatError> {
let v: serde_json::Value =
serde_json::from_str(json_str).map_err(|e| PeatError::InvalidInput {
msg: format!("publishNode: invalid JSON: {}", e),
})?;
let id = match v["id"].as_str() {
Some(id) if !id.is_empty() => id.to_string(),
_ => {
return Err(PeatError::InvalidInput {
msg: "publishNode: missing or empty 'id' field".to_string(),
});
}
};
Ok(NodeInfo {
id,
node_type: v["node_type"].as_str().unwrap_or("SOLDIER").to_string(),
name: v["name"].as_str().unwrap_or("Unknown").to_string(),
status: NodeStatus::from_str(v["status"].as_str().unwrap_or("ACTIVE")),
lat: v["lat"].as_f64().unwrap_or(0.0),
lon: v["lon"].as_f64().unwrap_or(0.0),
hae: v["hae"].as_f64(),
readiness: v["readiness"].as_f64().unwrap_or(1.0),
capabilities: v["capabilities"]
.as_array()
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(|s| s.to_string()))
.collect()
})
.unwrap_or_else(|| vec!["PLI".to_string()]),
cell_id: v["cell_id"].as_str().map(|s| s.to_string()),
battery_percent: parse_battery_percent(&v["battery_percent"]),
heart_rate: parse_heart_rate(&v["heart_rate"]),
last_heartbeat: parse_publish_last_heartbeat(&v["last_heartbeat"]),
})
}
fn parse_publish_last_heartbeat(v: &serde_json::Value) -> i64 {
let now_ms = chrono::Utc::now().timestamp_millis();
const FUTURE_GRACE_MS: i64 = 60_000;
let max_acceptable = now_ms.saturating_add(FUTURE_GRACE_MS);
match v.as_i64() {
Some(n) if n > max_acceptable => now_ms,
Some(n) if n < 0 => 0,
Some(n) => n,
None => now_ms,
}
}
fn serialize_nodes_get_json(nodes: &[NodeInfo]) -> String {
let json_array: Vec<serde_json::Value> = nodes
.iter()
.map(|p| {
serde_json::json!({
"id": p.id,
"node_type": p.node_type,
"name": p.name,
"status": p.status.as_str(),
"lat": p.lat,
"lon": p.lon,
"hae": p.hae,
"readiness": p.readiness,
"capabilities": p.capabilities,
"cell_id": p.cell_id,
"battery_percent": p.battery_percent,
"heart_rate": p.heart_rate,
"last_heartbeat": p.last_heartbeat,
})
})
.collect();
serde_json::to_string(&json_array).unwrap_or_else(|_| "[]".to_string())
}
fn coerce_json_number_to_i64(v: &serde_json::Value) -> Option<i64> {
if let Some(n) = v.as_i64() {
return Some(n);
}
v.as_f64().map(|f| f.round() as i64)
}
fn parse_battery_percent(v: &serde_json::Value) -> Option<i32> {
let n = coerce_json_number_to_i64(v)?;
Some(n.clamp(0, 100) as i32)
}
fn parse_heart_rate(v: &serde_json::Value) -> Option<i32> {
let n = coerce_json_number_to_i64(v)?;
Some(n.clamp(0, 250) as i32)
}
fn parse_marker_publish_json(id: &str, json_str: &str) -> Result<MarkerInfo, PeatError> {
let v: serde_json::Value =
serde_json::from_str(json_str).map_err(|e| PeatError::InvalidInput {
msg: format!("marker JSON: {}", e),
})?;
let uid = v["uid"]
.as_str()
.map(|s| s.to_string())
.filter(|s| !s.is_empty())
.unwrap_or_else(|| id.to_string());
if uid.is_empty() {
return Err(PeatError::InvalidInput {
msg: "marker missing uid (and no doc-store id supplied)".to_string(),
});
}
let deleted = v["_deleted"].as_bool().unwrap_or(false);
let marker_type = if deleted {
v["type"]
.as_str()
.unwrap_or(TOMBSTONE_PLACEHOLDER_TYPE)
.to_string()
} else {
v["type"]
.as_str()
.ok_or_else(|| PeatError::InvalidInput {
msg: format!("marker {uid} missing CoT type"),
})?
.to_string()
};
let lat = if deleted {
v["lat"].as_f64().unwrap_or(0.0)
} else {
v["lat"].as_f64().ok_or_else(|| PeatError::InvalidInput {
msg: format!("marker {uid} missing lat"),
})?
};
let lon = if deleted {
v["lon"].as_f64().unwrap_or(0.0)
} else {
v["lon"].as_f64().ok_or_else(|| PeatError::InvalidInput {
msg: format!("marker {uid} missing lon"),
})?
};
let hae = v["hae"].as_f64();
let ts = v["ts"].as_i64().unwrap_or(0);
let callsign = v["callsign"]
.as_str()
.filter(|s| !s.is_empty())
.map(|s| s.to_string());
let color = coerce_json_number_to_i64(&v["color"]).map(|n| n as i32);
let cell_id = v["cell_id"]
.as_str()
.filter(|s| !s.is_empty())
.map(|s| s.to_string());
Ok(MarkerInfo {
uid,
marker_type,
lat,
lon,
hae,
ts,
callsign,
color,
cell_id,
deleted,
})
}
fn serialize_markers_get_json(markers: &[MarkerInfo]) -> String {
let json_array: Vec<serde_json::Value> = markers
.iter()
.map(|m| {
let mut obj = serde_json::json!({
"uid": m.uid,
"type": m.marker_type,
"lat": m.lat,
"lon": m.lon,
"hae": m.hae,
"ts": m.ts,
"callsign": m.callsign,
"color": m.color,
"cell_id": m.cell_id,
});
if m.deleted {
obj["_deleted"] = serde_json::Value::Bool(true);
}
obj
})
.collect();
serde_json::to_string(&json_array).unwrap_or_else(|_| "[]".to_string())
}
fn serialize_marker_json(marker: &MarkerInfo) -> Result<String, PeatError> {
let mut v = serde_json::json!({
"uid": marker.uid,
"type": marker.marker_type,
"lat": marker.lat,
"lon": marker.lon,
"hae": marker.hae,
"ts": marker.ts,
"callsign": marker.callsign,
"color": marker.color,
"cell_id": marker.cell_id,
});
if marker.deleted {
v["_deleted"] = serde_json::Value::Bool(true);
}
serde_json::to_string(&v).map_err(|e| PeatError::EncodingError { msg: e.to_string() })
}
fn serialize_node_json(node: &NodeInfo) -> Result<String, PeatError> {
let v = serde_json::json!({
"node_type": node.node_type,
"name": node.name,
"status": node.status.as_str(),
"lat": node.lat,
"lon": node.lon,
"hae": node.hae,
"readiness": node.readiness,
"capabilities": node.capabilities,
"cell_id": node.cell_id,
"battery_percent": node.battery_percent,
"heart_rate": node.heart_rate,
"last_heartbeat": node.last_heartbeat,
});
serde_json::to_string(&v).map_err(|e| PeatError::EncodingError { msg: e.to_string() })
}
fn parse_command_json(id: &str, json: &str) -> Result<CommandInfo, PeatError> {
let root: serde_json::Value =
serde_json::from_str(json).map_err(|e| PeatError::InvalidInput {
msg: format!("Invalid JSON: {}", e),
})?;
let v = match root.get("fields") {
Some(f) if f.is_object() => f,
_ => &root,
};
Ok(CommandInfo {
id: id.to_string(),
command_type: v["command_type"].as_str().unwrap_or("UNKNOWN").to_string(),
target_id: v["target_id"].as_str().unwrap_or("").to_string(),
parameters: v["parameters"].to_string(),
priority: v["priority"].as_u64().unwrap_or(3) as u8,
status: CommandStatus::from_str(v["status"].as_str().unwrap_or("PENDING")),
originator: v["originator"].as_str().unwrap_or("").to_string(),
created_at: v["created_at"].as_i64().unwrap_or(0),
last_update: v["last_update"].as_i64().unwrap_or(0),
})
}
fn serialize_command_json(command: &CommandInfo) -> Result<String, PeatError> {
let params: serde_json::Value =
serde_json::from_str(&command.parameters).unwrap_or(serde_json::json!({}));
let v = serde_json::json!({
"command_type": command.command_type,
"target_id": command.target_id,
"parameters": params,
"priority": command.priority,
"status": command.status.as_str(),
"originator": command.originator,
"created_at": command.created_at,
"last_update": command.last_update,
});
serde_json::to_string(&v).map_err(|e| PeatError::EncodingError { msg: e.to_string() })
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_peat_version() {
let version = peat_version();
assert!(!version.is_empty());
assert!(version.contains('.'));
}
#[test]
fn test_encode_track() {
let track = TrackData {
track_id: "track-001".to_string(),
source_node: "node-1".to_string(),
position: Position {
lat: 34.0522,
lon: -118.2437,
hae: Some(100.0),
},
velocity: Some(Velocity {
bearing: 90.0,
speed_mps: 10.0,
}),
classification: "a-f-G-U-C".to_string(),
confidence: 0.95,
cell_id: Some("cell-1".to_string()),
formation_id: None,
};
let result = encode_track_to_cot(track);
assert!(result.is_ok());
let xml = result.unwrap();
assert!(xml.contains("<event"));
assert!(xml.contains("track-001"));
}
#[test]
fn test_encode_minimal_track() {
let track = TrackData {
track_id: "t1".to_string(),
source_node: "p1".to_string(),
position: Position {
lat: 0.0,
lon: 0.0,
hae: None,
},
velocity: None,
classification: "a-u-G".to_string(),
confidence: 0.5,
cell_id: None,
formation_id: None,
};
let result = encode_track_to_cot(track);
assert!(result.is_ok());
}
#[test]
fn test_invalid_track_id() {
let track = TrackData {
track_id: "".to_string(), source_node: "p1".to_string(),
position: Position {
lat: 0.0,
lon: 0.0,
hae: None,
},
velocity: None,
classification: "a-u-G".to_string(),
confidence: 0.5,
cell_id: None,
formation_id: None,
};
let result = encode_track_to_cot(track);
assert!(result.is_err());
}
#[test]
fn test_helper_functions() {
let pos = create_position(34.0, -118.0, Some(50.0));
assert_eq!(pos.lat, 34.0);
assert_eq!(pos.lon, -118.0);
assert_eq!(pos.hae, Some(50.0));
let vel = create_velocity(45.0, 15.0);
assert_eq!(vel.bearing, 45.0);
assert_eq!(vel.speed_mps, 15.0);
}
#[cfg(feature = "sync")]
mod publish_document_tests {
use super::*;
use peat_mesh::sync::traits::DataSyncBackend;
use peat_mesh::sync::InMemoryBackend;
fn fresh_node() -> peat_mesh::Node {
let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
peat_mesh::Node::new(backend)
}
fn rt() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime")
}
#[test]
fn round_trip_with_explicit_id() {
let rt = rt();
rt.block_on(async {
let node = fresh_node();
let json = r#"{
"id": "chat-001",
"sender": "ALPHA-1",
"text": "hello",
"timestamp": 1700000000000
}"#;
let id = publish_document_into_node(&node, "chats", json)
.await
.expect("publish");
assert_eq!(id, "chat-001");
let got = node
.get("chats", &"chat-001".to_string())
.await
.expect("get")
.expect("found");
assert_eq!(
got.fields.get("sender").and_then(|v| v.as_str()),
Some("ALPHA-1")
);
assert_eq!(
got.fields.get("text").and_then(|v| v.as_str()),
Some("hello")
);
assert!(
!got.fields.contains_key("id"),
"id is hoisted to Document::id, not duplicated in fields"
);
});
}
#[test]
fn id_assignment_when_absent() {
let rt = rt();
rt.block_on(async {
let node = fresh_node();
let json = r#"{"text":"orphan","sender":"BRAVO-2"}"#;
let id = publish_document_into_node(&node, "chats", json)
.await
.expect("publish");
assert!(!id.is_empty(), "backend must assign an id");
let got = node.get("chats", &id).await.expect("get").expect("found");
assert_eq!(
got.fields.get("text").and_then(|v| v.as_str()),
Some("orphan")
);
});
}
#[test]
fn malformed_json_errors() {
let rt = rt();
rt.block_on(async {
let node = fresh_node();
let result = publish_document_into_node(&node, "chats", "not-json").await;
assert!(result.is_err());
});
}
#[test]
fn non_object_json_errors() {
let rt = rt();
rt.block_on(async {
let node = fresh_node();
let result = publish_document_into_node(&node, "chats", "[1, 2, 3]").await;
assert!(result.is_err());
});
}
#[test]
fn non_string_id_falls_back_to_assigned() {
let rt = rt();
rt.block_on(async {
let node = fresh_node();
let json = r#"{"id":42,"text":"weird"}"#;
let id = publish_document_into_node(&node, "chats", json)
.await
.expect("publish");
assert_ne!(id, "42", "non-string id must be discarded, not coerced");
assert!(!id.is_empty());
});
}
#[test]
fn origin_variant_publishes_with_explicit_id() {
let rt = rt();
rt.block_on(async {
let node = fresh_node();
let json = r#"{"id":"ble-decoded-001","sender":"OBS-1","text":"x"}"#;
let id = publish_document_into_node_with_origin(
&node,
"chats",
json,
Some("ble".to_string()),
)
.await
.expect("publish_with_origin");
assert_eq!(id, "ble-decoded-001");
let got = node
.get("chats", &"ble-decoded-001".to_string())
.await
.expect("get")
.expect("found");
assert_eq!(
got.fields.get("sender").and_then(|v| v.as_str()),
Some("OBS-1")
);
});
}
#[test]
fn origin_variant_with_none_matches_plain_publish() {
let rt = rt();
rt.block_on(async {
let node = fresh_node();
let json = r#"{"id":"plain-001","text":"plain"}"#;
let id = publish_document_into_node_with_origin(&node, "chats", json, None)
.await
.expect("publish_with_origin(None)");
assert_eq!(id, "plain-001");
let got = node
.get("chats", &"plain-001".to_string())
.await
.expect("get")
.expect("found");
assert_eq!(
got.fields.get("text").and_then(|v| v.as_str()),
Some("plain")
);
});
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
mod ingest_position_tests {
use super::*;
use peat_mesh::sync::traits::DataSyncBackend;
use peat_mesh::sync::InMemoryBackend;
use peat_protocol::sync::ble_translation::BleTranslator;
struct Fixture {
translator: BleTranslator,
node: peat_mesh::Node,
}
fn fresh_fixture() -> Fixture {
let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
Fixture {
translator: BleTranslator::with_defaults(),
node: peat_mesh::Node::new(backend),
}
}
fn rt() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("runtime")
}
#[test]
fn round_trip_full_envelope() {
let rt = rt();
rt.block_on(async {
let fx = fresh_fixture();
const PERIPHERAL: u32 = 0xCAFE_0001;
let json = format!(
r#"{{
"lat": 40.7128,
"lon": -74.0060,
"altitude": 100.0,
"accuracy": 5.0,
"peripheral_id": {},
"callsign": "SCOUT-CAFE",
"mesh_id": "29C916FA"
}}"#,
PERIPHERAL
);
let id = ingest_position_via_translator(&fx.translator, &fx.node, &json)
.await
.expect("ingest");
assert_eq!(id, format!("ble-{:08X}", PERIPHERAL));
let doc = fx
.node
.get(fx.translator.tracks_collection(), &id)
.await
.expect("get")
.expect("found");
assert_eq!(
doc.fields.get("ble_origin"),
Some(&serde_json::Value::Bool(true)),
"ble_origin marker required for outbound loop suppression"
);
});
}
#[test]
fn omits_optional_fields() {
let rt = rt();
rt.block_on(async {
let fx = fresh_fixture();
let json = r#"{
"lat": 40.7128,
"lon": -74.0060,
"peripheral_id": 1
}"#;
let id = ingest_position_via_translator(&fx.translator, &fx.node, json)
.await
.expect("ingest");
assert_eq!(id, "ble-00000001");
});
}
#[test]
fn missing_required_fields_errors() {
let rt = rt();
rt.block_on(async {
let fx = fresh_fixture();
let json_no_lat = r#"{"lon": -74.0, "peripheral_id": 1}"#;
assert!(
ingest_position_via_translator(&fx.translator, &fx.node, json_no_lat)
.await
.is_err()
);
let json_no_id = r#"{"lat": 40.0, "lon": -74.0}"#;
assert!(
ingest_position_via_translator(&fx.translator, &fx.node, json_no_id)
.await
.is_err()
);
});
}
#[test]
fn malformed_json_errors() {
let rt = rt();
rt.block_on(async {
let fx = fresh_fixture();
let result =
ingest_position_via_translator(&fx.translator, &fx.node, "not-json").await;
assert!(result.is_err());
});
}
#[test]
fn peripheral_id_negative_int_form_recovers_to_same_u32() {
let rt = rt();
rt.block_on(async {
let fx = fresh_fixture();
const POSITIVE: i64 = 3_405_643_777;
const NEGATIVE: i64 = -889_323_519;
let expected_id = "ble-CAFE0001";
let positive_json = format!(
r#"{{ "lat": 40.0, "lon": -74.0, "peripheral_id": {} }}"#,
POSITIVE
);
let negative_json = format!(
r#"{{ "lat": 40.0, "lon": -74.0, "peripheral_id": {} }}"#,
NEGATIVE
);
let id_pos =
ingest_position_via_translator(&fx.translator, &fx.node, &positive_json)
.await
.expect("positive form ingests");
assert_eq!(id_pos, expected_id);
let id_neg =
ingest_position_via_translator(&fx.translator, &fx.node, &negative_json)
.await
.expect("negative (Kotlin Int) form ingests");
assert_eq!(
id_neg, expected_id,
"both forms must yield the same track id"
);
});
}
#[test]
fn peripheral_id_out_of_range_errors() {
let rt = rt();
rt.block_on(async {
let fx = fresh_fixture();
let too_big = r#"{ "lat": 40.0, "lon": -74.0, "peripheral_id": 4294967296 }"#;
assert!(
ingest_position_via_translator(&fx.translator, &fx.node, too_big)
.await
.is_err()
);
let too_small = r#"{ "lat": 40.0, "lon": -74.0, "peripheral_id": -2147483649 }"#;
assert!(
ingest_position_via_translator(&fx.translator, &fx.node, too_small)
.await
.is_err()
);
});
}
#[test]
fn peripheral_id_boundaries_accepted() {
let rt = rt();
rt.block_on(async {
let fx = fresh_fixture();
let max_json = r#"{ "lat": 40.0, "lon": -74.0, "peripheral_id": 4294967295 }"#;
let id = ingest_position_via_translator(&fx.translator, &fx.node, max_json)
.await
.expect("u32::MAX");
assert_eq!(id, "ble-FFFFFFFF");
let min_int_json = r#"{ "lat": 40.0, "lon": -74.0, "peripheral_id": -2147483648 }"#;
let id = ingest_position_via_translator(&fx.translator, &fx.node, min_int_json)
.await
.expect("i32::MIN as Int form");
assert_eq!(id, "ble-80000000");
});
}
#[tokio::test]
async fn ingest_emits_observer_event_with_ble_origin() {
use peat_mesh::sync::types::{ChangeEvent, Query};
let fx = fresh_fixture();
let mut tracks = fx
.node
.observe(fx.translator.tracks_collection(), &Query::All)
.expect("observe");
let json = r#"{
"lat": 40.7,
"lon": -74.0,
"peripheral_id": 1,
"callsign": "SCOUT-1"
}"#;
let _ = ingest_position_via_translator(&fx.translator, &fx.node, json)
.await
.expect("ingest");
loop {
let ev = tracks.receiver.recv().await.expect("event");
if let ChangeEvent::Updated { origin, .. } = ev {
assert_eq!(
origin,
Some("ble".to_string()),
"ingestPositionJni must publish with Some(\"ble\") origin per ADR-059"
);
break;
}
}
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
mod outbound_frame_tests {
use super::*;
use peat_mesh::sync::traits::DataSyncBackend;
use peat_mesh::sync::InMemoryBackend;
use peat_mesh::transport::{
FanoutHandle, OutboundSink, TranslationContext, Translator,
TranslatorRegistrationConfig,
};
use peat_protocol::sync::ble_translation::BleTranslator;
use peat_protocol::transport::{TransportManager, TransportManagerConfig};
use std::sync::Mutex as StdMutex;
use tokio::time::{timeout, Duration};
#[derive(Default)]
struct RecordingSink {
frames: StdMutex<Vec<(String, String, Vec<u8>)>>,
}
#[async_trait::async_trait]
impl OutboundSink for RecordingSink {
async fn send_outbound(
&self,
bytes: Vec<u8>,
ctx: &TranslationContext,
) -> anyhow::Result<()> {
let collection = ctx.collection.clone().unwrap_or_default();
self.frames
.lock()
.unwrap()
.push(("ble".to_string(), collection, bytes));
Ok(())
}
}
impl RecordingSink {
fn snapshot(&self) -> Vec<(String, String, Vec<u8>)> {
self.frames.lock().unwrap().clone()
}
}
struct Fixture {
node: Arc<peat_mesh::Node>,
translator: Arc<BleTranslator>,
transport_manager: TransportManager,
sink: Arc<RecordingSink>,
}
fn fixture() -> Fixture {
let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
Fixture {
node: Arc::new(peat_mesh::Node::new(backend)),
translator: Arc::new(BleTranslator::with_defaults()),
transport_manager: TransportManager::new(TransportManagerConfig::default()),
sink: Arc::new(RecordingSink::default()),
}
}
async fn register_and_start(fx: &Fixture) -> anyhow::Result<FanoutHandle> {
let translator_dyn: Arc<dyn Translator> = fx.translator.clone();
let sink_dyn: Arc<dyn OutboundSink> = fx.sink.clone();
fx.transport_manager
.register_translator(
translator_dyn,
sink_dyn,
TranslatorRegistrationConfig::ble(),
)
.await?;
fx.transport_manager.start_fanout(
Arc::clone(&fx.node),
vec![fx.translator.tracks_collection().to_string()],
)
}
async fn wait_for_frames(sink: &RecordingSink, expected: usize) {
let _ = timeout(Duration::from_secs(1), async {
loop {
if sink.snapshot().len() >= expected {
return;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await;
}
#[tokio::test]
async fn iroh_origin_doc_reaches_ble_sink() {
let fx = fixture();
let _h = register_and_start(&fx).await.expect("register");
let doc = peat_mesh::sync::types::Document::with_id("ble-00000001".to_string(), {
let mut f = std::collections::HashMap::new();
f.insert("lat".to_string(), serde_json::json!(40.0));
f.insert("lon".to_string(), serde_json::json!(-74.0));
f.insert(
"source_node".to_string(),
serde_json::json!("iroh-00000001"),
);
f.insert("hae".to_string(), serde_json::json!(100.0));
f.insert("cep".to_string(), serde_json::json!(5.0));
f.insert("classification".to_string(), serde_json::json!("a-f-G-U-C"));
f.insert("confidence".to_string(), serde_json::json!(0.9));
f.insert("category".to_string(), serde_json::json!("friendly"));
f.insert("callsign".to_string(), serde_json::json!("ALPHA-1"));
f.insert(
"created_at".to_string(),
serde_json::json!(1_700_000_000_000_i64),
);
f.insert(
"last_update".to_string(),
serde_json::json!(1_700_000_000_000_i64),
);
f
});
fx.node.publish("tracks", doc).await.expect("publish");
wait_for_frames(&fx.sink, 1).await;
let frames = fx.sink.snapshot();
assert!(
!frames.is_empty(),
"iroh-origin track must reach ble sink; got 0 frames"
);
let (transport, collection, bytes) = &frames[0];
assert_eq!(transport, "ble");
assert_eq!(collection, "tracks");
assert!(!bytes.is_empty(), "encoded bytes must be non-empty");
}
#[tokio::test]
async fn ble_origin_doc_does_not_re_encode_to_ble_sink() {
let fx = fixture();
let _h = register_and_start(&fx).await.expect("register");
let doc = peat_mesh::sync::types::Document::with_id("ble-CAFE0001".to_string(), {
let mut f = std::collections::HashMap::new();
f.insert("lat".to_string(), serde_json::json!(40.0));
f.insert("lon".to_string(), serde_json::json!(-74.0));
f.insert("ble_origin".to_string(), serde_json::json!(true));
f
});
fx.node
.publish_with_origin("tracks", doc, Some("ble".to_string()))
.await
.expect("publish");
tokio::time::sleep(Duration::from_millis(150)).await;
let frames = fx.sink.snapshot();
assert!(
frames.is_empty(),
"ble-origin doc must be suppressed from outbound BLE \
(ADR-059 same-node echo break); got {} frames",
frames.len()
);
}
#[tokio::test]
async fn drop_handle_stops_subsequent_delivery() {
let fx = fixture();
let h = register_and_start(&fx).await.expect("register");
fx.node
.publish(
"tracks",
peat_mesh::sync::types::Document::with_id("ble-00000001".to_string(), {
let mut f = std::collections::HashMap::new();
f.insert("lat".to_string(), serde_json::json!(40.0));
f.insert("lon".to_string(), serde_json::json!(-74.0));
f.insert("source_node".to_string(), serde_json::json!("iroh-1"));
f.insert("callsign".to_string(), serde_json::json!("A"));
f.insert("hae".to_string(), serde_json::json!(0.0));
f.insert("cep".to_string(), serde_json::json!(0.0));
f.insert("classification".to_string(), serde_json::json!("a-f-G-U-C"));
f.insert("confidence".to_string(), serde_json::json!(0.5));
f.insert("category".to_string(), serde_json::json!("friendly"));
f.insert(
"created_at".to_string(),
serde_json::json!(1_700_000_000_000_i64),
);
f.insert(
"last_update".to_string(),
serde_json::json!(1_700_000_000_000_i64),
);
f
}),
)
.await
.expect("publish-1");
wait_for_frames(&fx.sink, 1).await;
let pre_drop_count = fx.sink.snapshot().len();
assert!(pre_drop_count >= 1);
drop(h);
tokio::time::sleep(Duration::from_millis(50)).await;
fx.node
.publish(
"tracks",
peat_mesh::sync::types::Document::with_id("ble-00000002".to_string(), {
let mut f = std::collections::HashMap::new();
f.insert("lat".to_string(), serde_json::json!(41.0));
f.insert("lon".to_string(), serde_json::json!(-75.0));
f.insert("source_node".to_string(), serde_json::json!("iroh-2"));
f.insert("callsign".to_string(), serde_json::json!("B"));
f.insert("hae".to_string(), serde_json::json!(0.0));
f.insert("cep".to_string(), serde_json::json!(0.0));
f.insert("classification".to_string(), serde_json::json!("a-f-G-U-C"));
f.insert("confidence".to_string(), serde_json::json!(0.5));
f.insert("category".to_string(), serde_json::json!("friendly"));
f.insert(
"created_at".to_string(),
serde_json::json!(1_700_000_000_001_i64),
);
f.insert(
"last_update".to_string(),
serde_json::json!(1_700_000_000_001_i64),
);
f
}),
)
.await
.expect("publish-2");
tokio::time::sleep(Duration::from_millis(200)).await;
let post_drop_count = fx.sink.snapshot().len();
assert_eq!(
post_drop_count, pre_drop_count,
"no frames must arrive after FanoutHandle drop"
);
}
#[tokio::test]
async fn re_register_after_unregister_succeeds() {
let fx = fixture();
let h = register_and_start(&fx).await.expect("register-1");
drop(h);
fx.transport_manager
.unregister_translator("ble")
.await
.expect("unregister");
let _h2 = register_and_start(&fx).await.expect("register-2");
}
#[tokio::test]
async fn double_register_rejects() {
let fx = fixture();
let _h = register_and_start(&fx).await.expect("register-1");
let result = register_and_start(&fx).await;
assert!(
result.is_err(),
"second register on same transport_id must error"
);
}
#[tokio::test]
async fn queue_sink_enqueues_frames() {
let queue = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::<
OutboundFrame,
>::new()));
let sink = QueueOutboundSink {
transport_id: "ble",
queue: Arc::clone(&queue),
};
let ctx = TranslationContext::inbound("ble").with_collection("tracks");
sink.send_outbound(vec![0xAA, 0xBB], &ctx).await.unwrap();
sink.send_outbound(vec![0xCC], &ctx).await.unwrap();
let frames: Vec<OutboundFrame> = queue.lock().unwrap().drain(..).collect();
assert_eq!(frames.len(), 2);
assert_eq!(frames[0].transport_id, "ble");
assert_eq!(frames[0].collection, "tracks");
assert_eq!(frames[0].bytes, vec![0xAA, 0xBB]);
assert_eq!(frames[1].bytes, vec![0xCC]);
}
#[tokio::test]
async fn queue_sink_receives_fanned_out_doc() {
let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
let node = Arc::new(peat_mesh::Node::new(backend));
let translator = Arc::new(BleTranslator::with_defaults());
let tm = TransportManager::new(TransportManagerConfig::default());
let queue = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::<
OutboundFrame,
>::new()));
let sink: Arc<dyn OutboundSink> = Arc::new(QueueOutboundSink {
transport_id: "ble",
queue: Arc::clone(&queue),
});
let translator_dyn: Arc<dyn Translator> = translator.clone();
tm.register_translator(translator_dyn, sink, TranslatorRegistrationConfig::ble())
.await
.expect("register");
let _h = tm
.start_fanout(
Arc::clone(&node),
vec![translator.tracks_collection().to_string()],
)
.expect("start_fanout");
let doc = peat_mesh::sync::types::Document::with_id("q-00000001".to_string(), {
let mut f = std::collections::HashMap::new();
f.insert("lat".to_string(), serde_json::json!(51.5));
f.insert("lon".to_string(), serde_json::json!(-0.1));
f.insert("source_platform".to_string(), serde_json::json!("iroh-q01"));
f.insert("hae".to_string(), serde_json::json!(10.0));
f.insert("cep".to_string(), serde_json::json!(2.0));
f.insert("classification".to_string(), serde_json::json!("a-f-G-U-C"));
f.insert("confidence".to_string(), serde_json::json!(0.8));
f.insert("category".to_string(), serde_json::json!("friendly"));
f.insert("callsign".to_string(), serde_json::json!("BRAVO-1"));
f.insert(
"created_at".to_string(),
serde_json::json!(1_700_000_001_000_i64),
);
f
});
node.publish(translator.tracks_collection(), doc)
.await
.expect("publish");
let _ = timeout(Duration::from_secs(1), async {
loop {
if !queue.lock().unwrap().is_empty() {
return;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await;
let frames: Vec<OutboundFrame> = queue.lock().unwrap().drain(..).collect();
assert!(
!frames.is_empty(),
"queue sink must receive at least one frame"
);
assert_eq!(frames[0].transport_id, "ble");
assert_eq!(frames[0].collection, translator.tracks_collection());
}
#[tokio::test]
async fn ingest_inbound_frame_roundtrip_publishes_with_ble_origin() {
let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
let node = Arc::new(peat_mesh::Node::new(backend));
let translator = Arc::new(BleTranslator::with_defaults());
let outbound_doc =
peat_mesh::sync::types::Document::with_id("enc-00000001".to_string(), {
let mut f = std::collections::HashMap::new();
f.insert("lat".to_string(), serde_json::json!(48.858));
f.insert("lon".to_string(), serde_json::json!(2.294));
f.insert(
"source_platform".to_string(),
serde_json::json!("iroh-enc01"),
);
f.insert("hae".to_string(), serde_json::json!(50.0));
f.insert("cep".to_string(), serde_json::json!(3.0));
f.insert("classification".to_string(), serde_json::json!("a-f-G-U-C"));
f.insert("confidence".to_string(), serde_json::json!(0.9));
f.insert("category".to_string(), serde_json::json!("friendly"));
f.insert("callsign".to_string(), serde_json::json!("DELTA-1"));
f.insert(
"created_at".to_string(),
serde_json::json!(1_700_000_002_000_i64),
);
f
});
let encode_ctx = TranslationContext::inbound("ble")
.with_collection(translator.tracks_collection().to_string());
let postcard_bytes = translator
.encode_outbound(&outbound_doc, &encode_ctx)
.await
.expect("encode_outbound should produce Some bytes for a tracks doc");
let decode_ctx = TranslationContext::inbound("ble")
.with_collection(translator.tracks_collection().to_string());
let decoded = translator
.decode_inbound(&postcard_bytes, &decode_ctx)
.await
.expect("decode_inbound")
.expect("should produce a document for tracks");
let id = node
.publish_with_origin(
translator.tracks_collection(),
decoded,
Some("ble".to_string()),
)
.await
.expect("publish");
let stored = node
.get(translator.tracks_collection(), &id)
.await
.expect("get")
.expect("doc must be present after ingest");
assert!(
stored.fields.contains_key("lat"),
"decoded document must contain lat field"
);
}
}
#[cfg(all(feature = "sync", feature = "bluetooth", feature = "lite-bridge"))]
mod lite_bridge_outbound_frame_tests {
use super::*;
use peat_mesh::sync::traits::DataSyncBackend;
use peat_mesh::sync::InMemoryBackend;
use peat_mesh::transport::{
FanoutHandle, OutboundSink, TranslationContext, Translator,
TranslatorRegistrationConfig, BLE_LITE_BRIDGE,
};
use peat_protocol::sync::ble_translation::BleTranslator;
use peat_protocol::transport::{TransportManager, TransportManagerConfig};
use std::sync::Mutex as StdMutex;
use tokio::time::{timeout, Duration};
struct TaggedRecordingSink {
transport_id: &'static str,
frames: StdMutex<Vec<(String, String, Vec<u8>)>>,
}
#[async_trait::async_trait]
impl OutboundSink for TaggedRecordingSink {
async fn send_outbound(
&self,
bytes: Vec<u8>,
ctx: &TranslationContext,
) -> anyhow::Result<()> {
let collection = ctx.collection.clone().unwrap_or_default();
self.frames.lock().unwrap().push((
self.transport_id.to_string(),
collection,
bytes,
));
Ok(())
}
}
impl TaggedRecordingSink {
fn new(transport_id: &'static str) -> Arc<Self> {
Arc::new(Self {
transport_id,
frames: StdMutex::new(Vec::new()),
})
}
fn snapshot(&self) -> Vec<(String, String, Vec<u8>)> {
self.frames.lock().unwrap().clone()
}
}
async fn wait_for_any(sinks: &[&Arc<TaggedRecordingSink>], min_total: usize) {
let _ = timeout(Duration::from_secs(1), async {
loop {
let total: usize = sinks.iter().map(|s| s.snapshot().len()).sum();
if total >= min_total {
return;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await;
}
struct CoexistenceFixture {
node: Arc<peat_mesh::Node>,
transport_manager: TransportManager,
ble_sink: Arc<TaggedRecordingSink>,
lite_sink: Arc<TaggedRecordingSink>,
}
async fn coexistence_fixture() -> (CoexistenceFixture, FanoutHandle) {
let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
let node = Arc::new(peat_mesh::Node::new(backend));
let mgr = TransportManager::new(TransportManagerConfig::default());
let ble_translator = Arc::new(BleTranslator::with_defaults());
let ble_sink = TaggedRecordingSink::new("ble");
let ble_translator_dyn: Arc<dyn Translator> = ble_translator.clone();
let ble_sink_dyn: Arc<dyn OutboundSink> = ble_sink.clone();
mgr.register_translator(
ble_translator_dyn,
ble_sink_dyn,
TranslatorRegistrationConfig::ble(),
)
.await
.expect("register typed BLE");
let lite_translator: Arc<dyn Translator> = Arc::new(
CollectionGatedLiteBridge::for_ble_with_collections(LITE_BRIDGE_COLLECTIONS),
);
let lite_sink = TaggedRecordingSink::new(BLE_LITE_BRIDGE);
let lite_sink_dyn: Arc<dyn OutboundSink> = lite_sink.clone();
mgr.register_translator(
lite_translator,
lite_sink_dyn,
TranslatorRegistrationConfig::ble(),
)
.await
.expect("register lite-bridge");
let mut collections = vec![
ble_translator.tracks_collection().to_string(),
ble_translator.nodes_collection().to_string(),
];
for c in LITE_BRIDGE_COLLECTIONS {
collections.push((*c).to_string());
}
let handle = mgr
.start_fanout(Arc::clone(&node), collections)
.expect("start_fanout");
(
CoexistenceFixture {
node,
transport_manager: mgr,
ble_sink,
lite_sink,
},
handle,
)
}
fn marker_doc(uuid: &str) -> peat_mesh::sync::types::Document {
let mut fields = std::collections::HashMap::new();
fields.insert("type".to_string(), serde_json::json!("a-f-G-U-C"));
fields.insert("lat".to_string(), serde_json::json!(33.71));
fields.insert("lon".to_string(), serde_json::json!(-84.41));
peat_mesh::sync::types::Document::with_id(uuid.to_string(), fields)
}
fn track_doc(uuid: &str) -> peat_mesh::sync::types::Document {
let mut f = std::collections::HashMap::new();
f.insert("lat".to_string(), serde_json::json!(40.0));
f.insert("lon".to_string(), serde_json::json!(-74.0));
f.insert("source_node".to_string(), serde_json::json!("iroh-1"));
f.insert("hae".to_string(), serde_json::json!(0.0));
f.insert("cep".to_string(), serde_json::json!(0.0));
f.insert("classification".to_string(), serde_json::json!("a-f-G-U-C"));
f.insert("confidence".to_string(), serde_json::json!(0.5));
f.insert("category".to_string(), serde_json::json!("friendly"));
f.insert("callsign".to_string(), serde_json::json!("ALPHA-1"));
f.insert(
"created_at".to_string(),
serde_json::json!(1_700_000_000_000_i64),
);
f.insert(
"last_update".to_string(),
serde_json::json!(1_700_000_000_000_i64),
);
peat_mesh::sync::types::Document::with_id(uuid.to_string(), f)
}
#[tokio::test]
async fn marker_publish_reaches_only_lite_bridge_sink() {
let (fx, _h) = coexistence_fixture().await;
let doc = marker_doc("marker-uuid-001");
let original_fields = doc.fields.clone();
fx.node
.publish_with_origin("markers", doc, Some("self".to_string()))
.await
.expect("publish marker");
wait_for_any(&[&fx.ble_sink, &fx.lite_sink], 1).await;
let ble_frames = fx.ble_sink.snapshot();
let lite_frames = fx.lite_sink.snapshot();
assert!(
ble_frames.is_empty(),
"typed BLE sink MUST decline 'markers' (unknown collection); \
got {} frames",
ble_frames.len()
);
assert_eq!(
lite_frames.len(),
1,
"lite-bridge sink should see exactly one envelope for the marker"
);
let (transport_id, collection, bytes) = &lite_frames[0];
assert_eq!(transport_id, BLE_LITE_BRIDGE);
assert_eq!(collection, "markers");
let (envelope_collection, decoded) =
peat_mesh::transport::document_codec::decode_document(bytes)
.expect("decode envelope");
assert_eq!(envelope_collection, "markers");
assert_eq!(decoded.id.as_deref(), Some("marker-uuid-001"));
assert_eq!(decoded.fields, original_fields);
}
#[tokio::test]
async fn marker_tombstone_publish_reaches_lite_bridge_sink_with_deleted_flag() {
let (fx, _h) = coexistence_fixture().await;
let mut fields = std::collections::HashMap::new();
fields.insert("_deleted".to_string(), serde_json::json!(true));
fields.insert("ts".to_string(), serde_json::json!(1_700_000_000_000_i64));
let doc = peat_mesh::sync::types::Document::with_id(
"marker-tombstone-001".to_string(),
fields.clone(),
);
fx.node
.publish_with_origin("markers", doc, Some("self".to_string()))
.await
.expect("publish tombstone");
wait_for_any(&[&fx.ble_sink, &fx.lite_sink], 1).await;
let ble_frames = fx.ble_sink.snapshot();
let lite_frames = fx.lite_sink.snapshot();
assert!(
ble_frames.is_empty(),
"typed BLE sink MUST decline 'markers' tombstone (unknown collection)"
);
assert_eq!(
lite_frames.len(),
1,
"lite-bridge sink should see exactly one envelope for the tombstone"
);
let (_, collection, bytes) = &lite_frames[0];
assert_eq!(collection, "markers");
let (envelope_collection, decoded) =
peat_mesh::transport::document_codec::decode_document(bytes)
.expect("decode tombstone envelope");
assert_eq!(envelope_collection, "markers");
assert_eq!(decoded.id.as_deref(), Some("marker-tombstone-001"));
assert_eq!(
decoded.fields.get("_deleted"),
Some(&serde_json::json!(true)),
"tombstone _deleted: true must survive the BLE wire round-trip"
);
}
#[tokio::test]
async fn track_publish_reaches_only_typed_ble_sink() {
let (fx, _h) = coexistence_fixture().await;
let doc = track_doc("ble-CAFE0001");
fx.node.publish("tracks", doc).await.expect("publish track");
wait_for_any(&[&fx.ble_sink, &fx.lite_sink], 1).await;
let ble_frames = fx.ble_sink.snapshot();
let lite_frames = fx.lite_sink.snapshot();
assert_eq!(
ble_frames.len(),
1,
"typed BLE sink should see the track frame"
);
assert!(
lite_frames.is_empty(),
"lite-bridge sink MUST decline 'tracks' (not in \
LITE_BRIDGE_COLLECTIONS allow-list); got {} frames",
lite_frames.len()
);
}
#[tokio::test]
async fn ble_lite_origin_marker_does_not_re_emit_to_lite_bridge() {
let (fx, _h) = coexistence_fixture().await;
let skip_doc = marker_doc("marker-skip");
fx.node
.publish_with_origin("markers", skip_doc, Some(BLE_LITE_BRIDGE.to_string()))
.await
.expect("publish skip");
let barrier_doc = marker_doc("marker-barrier");
fx.node
.publish_with_origin("markers", barrier_doc, Some("self".to_string()))
.await
.expect("publish barrier");
wait_for_any(&[&fx.lite_sink], 1).await;
let lite_frames = fx.lite_sink.snapshot();
assert_eq!(
lite_frames.len(),
1,
"lite-bridge sink MUST receive only the barrier doc; \
the BLE_LITE_BRIDGE-origin doc must be suppressed by \
origin-skip (echo-loop break)"
);
let bytes = &lite_frames[0].2;
let (_collection, decoded) =
peat_mesh::transport::document_codec::decode_document(bytes)
.expect("decode envelope");
assert_eq!(decoded.id.as_deref(), Some("marker-barrier"));
}
#[tokio::test]
async fn re_register_with_lite_bridge_after_unregister_succeeds() {
let (fx, h1) = coexistence_fixture().await;
drop(h1);
fx.transport_manager
.unregister_translator(BLE_LITE_BRIDGE)
.await
.expect("unregister lite-bridge");
fx.transport_manager
.unregister_translator("ble")
.await
.expect("unregister typed BLE");
let ble_translator = Arc::new(BleTranslator::with_defaults());
let ble_sink = TaggedRecordingSink::new("ble");
let ble_translator_dyn: Arc<dyn Translator> = ble_translator.clone();
let ble_sink_dyn: Arc<dyn OutboundSink> = ble_sink.clone();
fx.transport_manager
.register_translator(
ble_translator_dyn,
ble_sink_dyn,
TranslatorRegistrationConfig::ble(),
)
.await
.expect("re-register typed BLE");
let lite_translator: Arc<dyn Translator> = Arc::new(
CollectionGatedLiteBridge::for_ble_with_collections(LITE_BRIDGE_COLLECTIONS),
);
let lite_sink = TaggedRecordingSink::new(BLE_LITE_BRIDGE);
let lite_sink_dyn: Arc<dyn OutboundSink> = lite_sink.clone();
fx.transport_manager
.register_translator(
lite_translator,
lite_sink_dyn,
TranslatorRegistrationConfig::ble(),
)
.await
.expect("re-register lite-bridge");
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
mod poll_api_wrapper_tests {
use super::*;
fn test_cfg(storage_path: &str) -> NodeConfig {
NodeConfig {
app_id: "poll-wrapper-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: storage_path.to_string(),
transport: None,
}
}
#[test]
fn subscribe_poll_drain_and_cancel() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(test_cfg(tmp.path().to_str().unwrap())).expect("create_node");
let handle = node.subscribe_poll().expect("subscribe_poll");
let mesh_node = Arc::clone(&node.node);
node.runtime
.block_on(publish_document_into_node(
&mesh_node,
"test",
r#"{"id":"doc-001","x":1}"#,
))
.expect("publish_document_into_node");
std::thread::sleep(std::time::Duration::from_millis(100));
let changes = handle.poll_changes();
assert!(
!changes.is_empty(),
"poll_changes must return changes after publish_document_into_node"
);
assert!(
changes.iter().any(|c| c.collection == "test"),
"change must be for the 'test' collection; got: {changes:?}"
);
assert!(
handle.poll_changes().is_empty(),
"second poll must be empty after drain"
);
handle.cancel();
handle.cancel();
}
#[test]
fn outbound_frames_start_poll_ingest_stop_restart() {
let tmp_a = tempfile::tempdir().unwrap();
let tmp_b = tempfile::tempdir().unwrap();
let node_a = create_node(test_cfg(tmp_a.path().to_str().unwrap())).expect("node_a");
let node_b = create_node(test_cfg(tmp_b.path().to_str().unwrap())).expect("node_b");
node_a.start_outbound_frames().expect("start 1");
node_a
.start_outbound_frames()
.expect("start 2 (idempotent no-op)");
let tracks_json = r#"{
"id": "track-wrap-001",
"lat": 51.5, "lon": -0.1,
"source_platform": "test-01",
"hae": 10.0, "cep": 2.0,
"classification": "a-f-G-U-C",
"confidence": 0.9,
"category": "friendly",
"callsign": "ALPHA-1",
"created_at": 1700000001000
}"#;
let mesh_a = Arc::clone(&node_a.node);
node_a
.runtime
.block_on(publish_document_into_node(&mesh_a, "tracks", tracks_json))
.expect("publish tracks");
let mut frames = Vec::new();
for _ in 0..40 {
frames = node_a.poll_outbound_frames();
if !frames.is_empty() {
break;
}
std::thread::sleep(std::time::Duration::from_millis(25));
}
assert!(
!frames.is_empty(),
"outbound frames must appear after publishing to 'tracks'"
);
assert_eq!(frames[0].transport_id, "ble");
assert_eq!(frames[0].collection, "tracks");
let doc_id = node_b
.ingest_inbound_frame("tracks".to_string(), frames[0].bytes.clone())
.expect("ingest_inbound_frame must not error")
.expect("must return a doc_id for a valid tracks frame");
assert!(!doc_id.is_empty(), "ingested doc_id must be non-empty");
let stored = node_b
.runtime
.block_on(Arc::clone(&node_b.node).get("tracks", &doc_id))
.expect("get must not error")
.expect("ingested document must be in node_b's store");
assert!(
stored.fields.contains_key("lat"),
"decoded track must carry lat field"
);
node_a.stop_outbound_frames();
node_a
.start_outbound_frames()
.expect("re-start after stop must succeed");
node_a.stop_outbound_frames(); }
#[cfg(feature = "lite-bridge")]
#[test]
fn lite_outbound_poll_ingest_converges_without_echo() {
let tmp_a = tempfile::tempdir().unwrap();
let tmp_b = tempfile::tempdir().unwrap();
let node_a = create_node(test_cfg(tmp_a.path().to_str().unwrap())).expect("node_a");
let node_b = create_node(test_cfg(tmp_b.path().to_str().unwrap())).expect("node_b");
node_a.start_outbound_frames().expect("start a");
node_b.start_outbound_frames().expect("start b");
let demo_json = r#"{"id":"counter-demo-lite","inc":3,"dec":1,"by":"BRAVO"}"#;
let mesh_a = Arc::clone(&node_a.node);
node_a
.runtime
.block_on(publish_document_into_node(&mesh_a, "demo", demo_json))
.expect("publish demo");
let mut lite = None;
for _ in 0..40 {
if let Some(f) = node_a
.poll_outbound_frames()
.into_iter()
.find(|f| f.transport_id == "ble-lite" && f.collection == "demo")
{
lite = Some(f);
break;
}
std::thread::sleep(std::time::Duration::from_millis(25));
}
let lite = lite.expect("a ble-lite frame must appear for the 'demo' doc");
let _ = node_b.poll_outbound_frames();
let doc_id = node_b
.ingest_inbound_lite_frame("demo".to_string(), lite.bytes.clone())
.expect("ingest_inbound_lite_frame must not error")
.expect("must return a doc_id for a valid demo lite frame");
assert!(!doc_id.is_empty(), "ingested doc_id must be non-empty");
let stored = node_b
.runtime
.block_on(Arc::clone(&node_b.node).get("demo", &doc_id))
.expect("get must not error")
.expect("ingested demo doc must be in node_b's store");
assert_eq!(
stored.fields.get("inc").and_then(|v| v.as_i64()),
Some(3),
"decoded demo doc must carry inc=3"
);
assert_eq!(
stored.fields.get("by").and_then(|v| v.as_str()),
Some("BRAVO"),
"decoded demo doc must carry the 'by' field"
);
let mut echoed = false;
for _ in 0..16 {
if node_b
.poll_outbound_frames()
.iter()
.any(|f| f.transport_id == "ble-lite" && f.collection == "demo")
{
echoed = true;
break;
}
std::thread::sleep(std::time::Duration::from_millis(25));
}
assert!(
!echoed,
"ingested ble-lite doc must NOT be re-emitted on ble-lite \
(origin-skip / echo-suppression)"
);
node_a.stop_outbound_frames();
node_b.stop_outbound_frames();
}
#[test]
fn owning_node_slot_store_then_clear_drops_exactly_one_ref() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(test_cfg(tmp.path().to_str().unwrap())).expect("node");
let slot = std::sync::Mutex::new(0i64);
let before = Arc::strong_count(&node);
store_owning_node_in_slot(&slot, &node);
assert_ne!(
*slot.lock().unwrap(),
0,
"store must stash a non-zero owning pointer"
);
assert_eq!(
Arc::strong_count(&node),
before + 1,
"store must add exactly one owning reference"
);
clear_owning_node_slot(&slot);
assert_eq!(*slot.lock().unwrap(), 0, "clear must zero the slot");
assert_eq!(
Arc::strong_count(&node),
before,
"clear must drop exactly the one stored reference (no leak/double-free)"
);
}
}
mod doc_shape_parse_tests {
use super::*;
fn wrap(fields_json: &str) -> String {
String::from(r#"{"id":"x","fields":"#)
+ fields_json
+ r#","updated_at":{"secs_since_epoch":1730000000,"nanos_since_epoch":0}}"#
}
#[test]
fn parse_node_json_wrapped_equals_flat() {
let flat = r#"{"node_type":"peat-flutter","name":"Kilo","status":"ACTIVE","readiness":1.0,"capabilities":["comms","leader"],"last_heartbeat":1730000000000}"#;
let a = parse_node_json("n1", flat).expect("flat parse");
let b = parse_node_json("n1", &wrap(flat)).expect("wrapped parse");
assert_eq!(
b.name, "Kilo",
"wrapped name must come from fields, not the id"
);
assert_eq!(b.name, a.name);
assert_eq!(b.node_type, a.node_type);
assert_eq!(b.capabilities, a.capabilities);
assert_eq!(
b.capabilities,
vec!["comms".to_string(), "leader".to_string()]
);
assert_eq!(b.last_heartbeat, a.last_heartbeat);
assert_eq!(b.last_heartbeat, 1730000000000);
}
#[test]
fn parse_cell_json_wrapped_equals_flat() {
let flat = r#"{"name":"Alpha Cell","status":"ACTIVE","node_count":2,"capabilities":["comms"],"leader_id":"n1","last_update":1730000000000}"#;
let a = parse_cell_json("alpha", flat).expect("flat parse");
let b = parse_cell_json("alpha", &wrap(flat)).expect("wrapped parse");
assert_eq!(b.name, "Alpha Cell");
assert_eq!(b.node_count, 2);
assert_eq!(b.node_count, a.node_count);
assert_eq!(b.leader_id, a.leader_id);
assert_eq!(b.capabilities, a.capabilities);
}
#[test]
fn parse_command_json_wrapped_equals_flat() {
let flat = r#"{"command_type":"WATER_REQUEST","target_id":"leader","parameters":{"quantity":5,"from":"Kilo"},"priority":1,"status":"PENDING","originator":"n1","created_at":1730000000000,"last_update":1730000000000}"#;
let a = parse_command_json("req-1", flat).expect("flat parse");
let b = parse_command_json("req-1", &wrap(flat)).expect("wrapped parse");
assert_eq!(b.command_type, "WATER_REQUEST");
assert_eq!(b.command_type, a.command_type);
assert_eq!(b.originator, a.originator);
assert_eq!(b.target_id, a.target_id);
assert_eq!(b.parameters, a.parameters);
}
}
#[cfg(feature = "sync")]
mod blob_tests {
use super::*;
fn generate_test_image(label: &str, width: u32, height: u32, hue_shift: u8) -> Vec<u8> {
let body_len = (width as usize * height as usize) / 4;
let mut buf = Vec::with_capacity(body_len + label.len() + 8);
buf.extend_from_slice(&[0xFF, 0xD8]); buf.extend_from_slice(label.as_bytes());
buf.push(hue_shift);
buf.extend(std::iter::repeat(hue_shift.wrapping_mul(3)).take(body_len));
buf.extend_from_slice(&[0xFF, 0xD9]); buf
}
fn test_node_config(storage_path: &str) -> NodeConfig {
NodeConfig {
app_id: "blob-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: storage_path.to_string(),
transport: None,
}
}
#[test]
fn test_blob_put_get_local_roundtrip() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(test_node_config(tmp.path().to_str().unwrap()))
.expect("create_node failed");
node.enable_blob_transfer(None)
.expect("enable_blob_transfer failed");
assert!(
node.blob_endpoint_id().is_some(),
"blob endpoint should be initialized"
);
let test_data = b"SKUNK-1 image chip placeholder";
let hash = node
.blob_put(test_data, "image/jpeg")
.expect("blob_put failed");
assert!(!hash.is_empty(), "hash should be non-empty");
assert!(
node.blob_exists_locally(&hash),
"blob should exist locally after put"
);
let retrieved = node.blob_get(&hash).expect("blob_get failed");
assert_eq!(retrieved, test_data, "retrieved bytes must match original");
}
#[test]
fn test_blob_get_nonexistent_returns_error() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(test_node_config(tmp.path().to_str().unwrap()))
.expect("create_node failed");
node.enable_blob_transfer(None)
.expect("enable_blob_transfer failed");
let fake_hash = "0000000000000000000000000000000000000000000000000000000000000000";
assert!(
!node.blob_exists_locally(fake_hash),
"nonexistent hash should not be local"
);
let result = node.blob_get(fake_hash);
assert!(result.is_err(), "fetching nonexistent blob should error");
}
#[test]
fn test_blob_transfer_disabled_errors() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(test_node_config(tmp.path().to_str().unwrap()))
.expect("create_node failed");
assert!(node.blob_endpoint_id().is_none());
assert!(node.blob_put(b"data", "text/plain").is_err());
assert!(node.blob_get("abc").is_err());
assert!(!node.blob_exists_locally("abc"));
}
#[test]
fn test_blob_cross_node_transfer() {
let tmp_a = tempfile::tempdir().unwrap();
let tmp_b = tempfile::tempdir().unwrap();
let node_a = create_node(NodeConfig {
app_id: "blob-xfer-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: tmp_a.path().to_str().unwrap().to_string(),
transport: None,
})
.expect("create node A");
let node_b = create_node(NodeConfig {
app_id: "blob-xfer-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: tmp_b.path().to_str().unwrap().to_string(),
transport: None,
})
.expect("create node B");
node_a
.enable_blob_transfer(Some("127.0.0.1:0".parse().unwrap()))
.expect("enable blob A");
node_b
.enable_blob_transfer(Some("127.0.0.1:0".parse().unwrap()))
.expect("enable blob B");
let a_endpoint_id = node_a.blob_endpoint_id().expect("A blob endpoint");
let a_addr = node_a.blob_bound_addr().expect("A bound addr");
node_b
.blob_add_peer(&a_endpoint_id, &a_addr)
.expect("add peer");
let test_data = b"cross-node image chip test payload 1234567890";
let hash = node_a.blob_put(test_data, "image/jpeg").expect("put on A");
let retrieved = node_b.blob_get(&hash).expect("get from B");
assert_eq!(
retrieved, test_data,
"cross-node blob transfer: bytes must match"
);
}
#[test]
fn test_e2e_contact_report_with_image_chip() {
let tmp_sim = tempfile::tempdir().unwrap();
let tmp_tablet = tempfile::tempdir().unwrap();
let sim = create_node(NodeConfig {
app_id: "e2e-contact-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: tmp_sim.path().to_str().unwrap().to_string(),
transport: None,
})
.expect("create sim node");
let tablet = create_node(NodeConfig {
app_id: "e2e-contact-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: tmp_tablet.path().to_str().unwrap().to_string(),
transport: None,
})
.expect("create tablet node");
sim.enable_blob_transfer(Some("127.0.0.1:0".parse().unwrap()))
.expect("sim blob");
tablet
.enable_blob_transfer(Some("127.0.0.1:0".parse().unwrap()))
.expect("tablet blob");
let sim_blob_id = sim.blob_endpoint_id().unwrap();
let sim_blob_addr = sim.blob_bound_addr().unwrap();
tablet
.blob_add_peer(&sim_blob_id, &sim_blob_addr)
.expect("tablet add sim as blob peer");
let sim_sync_id = sim.node_id();
let sim_sync_addr = format!("{:?}", sim.iroh_transport.endpoint_addr());
let sim_peer = PeerInfo {
name: "sim".to_string(),
node_id: sim_sync_id.clone(),
addresses: vec![],
relay_url: None,
};
let sim_clone = Arc::clone(&sim);
let tablet_clone = Arc::clone(&tablet);
tablet.runtime.block_on(async {
tablet_clone
.iroh_transport
.connect_peer(&peat_protocol::network::PeerInfo {
name: "sim".to_string(),
node_id: sim_sync_id,
addresses: vec![sim_clone
.iroh_transport
.endpoint_addr()
.addrs
.iter()
.next()
.map(|a| format!("{}", a))
.unwrap_or_default()],
relay_url: None,
})
.await
.ok();
});
let fake_jpeg = b"\xFF\xD8\xFF\xE0fake-jpeg-contact-report-image-chip-data";
let image_hash = sim.blob_put(fake_jpeg, "image/jpeg").expect("sim blob put");
let track_json = serde_json::json!({
"id": "red-track-1",
"source_node": "sensor-node-3",
"source_model": "FLIR Vue Pro R 640",
"model_version": "1.0",
"cell_id": "company-CHARLIE",
"lat": 32.655,
"lon": -117.245,
"heading": 0.0,
"speed": 7.7,
"classification": "a-h-S",
"confidence": 0.82,
"category": "VESSEL",
"attributes": {
"callsign": "SKUNK-1",
"speed_kts": "15",
"vehicle_class": "fast attack craft",
"reporter": "sensor-node-3",
"distance_to_reporter_m": "800",
"image_chip_hash": &image_hash,
},
"last_update": std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64,
});
let sim_backend = &sim.storage_backend;
let tracks_coll = sim_backend.collection("tracks");
tracks_coll
.upsert("red-track-1", track_json.to_string().into_bytes())
.expect("sim upsert track");
std::thread::sleep(std::time::Duration::from_secs(2));
let tablet_tracks = tablet_clone.storage_backend.collection("tracks");
let track_doc = tablet_tracks.scan().expect("tablet scan tracks");
if let Some((_id, data)) = track_doc.into_iter().find(|(id, _)| id == "red-track-1") {
let parsed: serde_json::Value = serde_json::from_slice(&data).expect("valid JSON");
assert_eq!(parsed["source_node"], "sensor-node-3");
assert_eq!(parsed["classification"], "a-h-S");
assert_eq!(parsed["attributes"]["callsign"], "SKUNK-1");
assert_eq!(parsed["attributes"]["image_chip_hash"], image_hash);
let chip_hash = parsed["attributes"]["image_chip_hash"]
.as_str()
.expect("hash is string");
let chip_bytes = tablet.blob_get(chip_hash).expect("tablet blob get");
assert_eq!(
chip_bytes, fake_jpeg,
"image chip bytes must match across mesh"
);
eprintln!("E2E PASS: contact report + image chip transferred through mesh");
} else {
eprintln!(
"E2E SKIP: doc sync didn't complete in 2s (blob transfer \
validated separately). Re-run if you want full chain coverage."
);
}
}
#[test]
fn test_blob_transfer_with_synthetic_image() {
let tmp_a = tempfile::tempdir().unwrap();
let tmp_b = tempfile::tempdir().unwrap();
let node_a = create_node(NodeConfig {
app_id: "img-xfer-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: tmp_a.path().to_str().unwrap().to_string(),
transport: None,
})
.expect("create node A");
let node_b = create_node(NodeConfig {
app_id: "img-xfer-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: tmp_b.path().to_str().unwrap().to_string(),
transport: None,
})
.expect("create node B");
node_a
.enable_blob_transfer(Some("127.0.0.1:0".parse().unwrap()))
.expect("enable A");
node_b
.enable_blob_transfer(Some("127.0.0.1:0".parse().unwrap()))
.expect("enable B");
let a_id = node_a.blob_endpoint_id().unwrap();
let a_addr = node_a.blob_bound_addr().unwrap();
node_b.blob_add_peer(&a_id, &a_addr).expect("add peer");
let images = vec![
(
"distant",
generate_test_image("SKUNK-1 DISTANT", 160, 120, 40),
),
(
"approach",
generate_test_image("SKUNK-1 APPROACH", 160, 120, 80),
),
("close", generate_test_image("SKUNK-1 CLOSE", 160, 120, 160)),
("id", generate_test_image("SKUNK-1 ID", 160, 120, 220)),
];
for (label, jpeg_bytes) in &images {
assert!(jpeg_bytes.len() > 100, "{} should be a real JPEG", label);
assert!(
jpeg_bytes.len() < 80_000,
"{} should be under 80KB (got {})",
label,
jpeg_bytes.len()
);
assert_eq!(jpeg_bytes[0], 0xFF);
assert_eq!(jpeg_bytes[1], 0xD8);
}
let mut hashes = Vec::new();
for (label, jpeg_bytes) in &images {
let hash = node_a
.blob_put(jpeg_bytes, "image/jpeg")
.unwrap_or_else(|e| panic!("put {label}: {e}"));
hashes.push((label.to_string(), hash));
}
for (label, hash) in &hashes {
let fetched = node_b
.blob_get(hash)
.unwrap_or_else(|e| panic!("get {label}: {e}"));
let original = &images.iter().find(|(l, _)| l == label).unwrap().1;
assert_eq!(
fetched.len(),
original.len(),
"{}: fetched size must match",
label
);
assert_eq!(
fetched, *original,
"{}: fetched bytes must match original",
label
);
}
eprintln!(
"IMAGE TRANSFER PASS: 4 synthetic JPEGs transferred cross-node ({} total bytes)",
images.iter().map(|(_, b)| b.len()).sum::<usize>()
);
}
}
#[cfg(feature = "sync")]
mod m4_endpoint_and_get_document_tests {
use super::*;
fn test_node_config(storage_path: &str) -> NodeConfig {
NodeConfig {
app_id: "m4-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: storage_path.to_string(),
transport: None,
}
}
#[test]
fn endpoint_socket_addr_returns_parseable_loopback_addr() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(test_node_config(tmp.path().to_str().unwrap()))
.expect("create_node failed");
let addr_str = node
.endpoint_socket_addr()
.expect("a bound node must report at least one IP address");
let parsed: std::net::SocketAddr = addr_str.parse().unwrap_or_else(|e| {
panic!("endpoint_socket_addr returned '{addr_str}' which doesn't parse as SocketAddr: {e}")
});
assert!(
parsed.port() > 0,
"port must be nonzero for a bound socket, got {parsed}"
);
}
#[test]
fn document_layer_round_trip_publish_then_get() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(test_node_config(tmp.path().to_str().unwrap()))
.expect("create_node failed");
let collection = "markers";
let doc_id = "M-RT-1";
let body = format!(r#"{{"id":"{doc_id}","name":"alpha","severity":3}}"#);
let mesh_node = Arc::clone(&node.node);
let returned_id = node
.runtime
.block_on(publish_document_into_node(&mesh_node, collection, &body))
.expect("publish_document_into_node");
assert_eq!(returned_id, doc_id);
let fetched = node
.runtime
.block_on(mesh_node.get(collection, &doc_id.to_string()))
.expect("get must not Err")
.expect("doc must be present on the publishing node");
assert_eq!(
fetched.id.as_deref(),
Some(doc_id),
"published id must round-trip through Document::id"
);
assert_eq!(
fetched.fields.get("name").and_then(|v| v.as_str()),
Some("alpha")
);
assert_eq!(
fetched.fields.get("severity").and_then(|v| v.as_i64()),
Some(3)
);
}
#[test]
fn jni_serializer_reinserts_id_alongside_fields() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(test_node_config(tmp.path().to_str().unwrap()))
.expect("create_node failed");
let collection = "markers";
let doc_id = "M-RT-1";
let body = format!(r#"{{"id":"{doc_id}","name":"alpha","severity":3}}"#);
let mesh_node = Arc::clone(&node.node);
let _ = node
.runtime
.block_on(publish_document_into_node(&mesh_node, collection, &body))
.expect("publish");
let fetched = node
.runtime
.block_on(mesh_node.get(collection, &doc_id.to_string()))
.expect("get must not Err")
.expect("doc must be present");
let json = serialize_document_for_get_jni(&fetched);
let parsed: serde_json::Value =
serde_json::from_str(&json).expect("JNI output must parse as JSON");
assert!(
parsed.is_object(),
"output must be a JSON object, got {parsed:?}"
);
assert_eq!(parsed["id"], doc_id, "id must be reinserted");
assert_eq!(parsed["name"], "alpha");
assert_eq!(parsed["severity"], 3);
assert_eq!(
parsed.as_object().unwrap().len(),
3,
"unexpected extra fields in JNI serialization: {parsed}"
);
}
#[test]
fn jni_serializer_omits_id_when_none() {
let doc = peat_mesh::sync::Document {
id: None,
fields: {
let mut m = std::collections::HashMap::new();
m.insert("k".to_string(), serde_json::Value::String("v".into()));
m
},
updated_at: std::time::SystemTime::now(),
};
let json = serialize_document_for_get_jni(&doc);
let parsed: serde_json::Value = serde_json::from_str(&json).expect("parseable JSON");
assert!(
parsed.get("id").is_none(),
"expected id absent (not null) when Document::id is None, got {json}"
);
assert_eq!(parsed["k"], "v");
}
#[test]
fn document_layer_get_returns_none_for_missing_doc() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(test_node_config(tmp.path().to_str().unwrap()))
.expect("create_node failed");
let mesh_node = Arc::clone(&node.node);
let result = node
.runtime
.block_on(mesh_node.get("markers", &"never-published".to_string()))
.expect("get must not Err");
assert!(
result.is_none(),
"expected None for a never-published doc, got {result:?}"
);
}
}
#[cfg(feature = "sync")]
mod node_tests {
use super::*;
fn fixture(battery: Option<i32>, heart: Option<i32>) -> NodeInfo {
NodeInfo {
id: "ANDROID-fixture".to_string(),
node_type: "SOLDIER".to_string(),
name: "HOBO".to_string(),
status: NodeStatus::Active,
lat: 33.71576,
lon: -84.41152,
hae: Some(305.0),
readiness: 1.0,
capabilities: vec!["PLI".to_string()],
cell_id: Some("BRAVO".to_string()),
battery_percent: battery,
heart_rate: heart,
last_heartbeat: 1_700_000_000_000,
}
}
#[test]
fn every_optional_field_round_trips_through_storage_codec() {
let original = fixture(Some(85), Some(72));
let json = serialize_node_json(&original).expect("serialize");
let parsed = parse_node_json(&original.id, &json).expect("parse");
assert_eq!(parsed.id, original.id);
assert_eq!(parsed.node_type, original.node_type);
assert_eq!(parsed.name, original.name);
assert_eq!(parsed.lat, original.lat);
assert_eq!(parsed.lon, original.lon);
assert_eq!(parsed.hae, original.hae);
assert_eq!(parsed.readiness, original.readiness);
assert_eq!(parsed.capabilities, original.capabilities);
assert_eq!(parsed.cell_id, original.cell_id);
assert_eq!(parsed.battery_percent, original.battery_percent);
assert_eq!(parsed.heart_rate, original.heart_rate);
assert_eq!(parsed.last_heartbeat, original.last_heartbeat);
}
#[test]
fn battery_none_round_trips_as_none() {
let original = fixture(None, None);
let json = serialize_node_json(&original).expect("serialize");
let parsed = parse_node_json(&original.id, &json).expect("parse");
assert!(parsed.battery_percent.is_none());
assert!(parsed.heart_rate.is_none());
}
#[test]
fn legacy_json_without_battery_or_heart_parses_with_none() {
let legacy_json = serde_json::json!({
"node_type": "SOLDIER",
"name": "LEGACY-PEER",
"status": "ACTIVE",
"lat": 33.71,
"lon": -84.41,
"hae": null,
"readiness": 1.0,
"capabilities": ["PLI"],
"cell_id": "BRAVO",
"last_heartbeat": 1_700_000_000_000_i64,
})
.to_string();
let parsed =
parse_node_json("LEGACY-PEER", &legacy_json).expect("legacy json must parse");
assert!(parsed.battery_percent.is_none());
assert!(parsed.heart_rate.is_none());
assert_eq!(parsed.cell_id.as_deref(), Some("BRAVO"));
}
#[test]
fn put_node_get_nodes_preserves_battery_and_heart() {
let tmp = tempfile::tempdir().unwrap();
let node = create_node(NodeConfig {
app_id: "node-rt-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: tmp.path().to_str().unwrap().to_string(),
transport: None,
})
.expect("create_node");
let original = fixture(Some(85), Some(72));
node.put_node(original.clone()).expect("put_node");
let listed = node.get_nodes().expect("get_nodes");
let found = listed
.iter()
.find(|p| p.id == original.id)
.expect("published node must appear in get_nodes");
assert_eq!(
found.battery_percent,
Some(85),
"battery_percent dropped between put_node and get_nodes"
);
assert_eq!(
found.heart_rate,
Some(72),
"heart_rate dropped between put_node and get_nodes"
);
assert_eq!(found.cell_id.as_deref(), Some("BRAVO"));
}
#[test]
fn publish_json_inline_parser_extracts_battery_and_heart() {
let json = r#"{
"id": "ANDROID-abc123",
"name": "HOBO",
"node_type": "SOLDIER",
"lat": 33.71576,
"lon": -84.41152,
"hae": 305.0,
"status": "ACTIVE",
"capabilities": ["PLI"],
"readiness": 1.0,
"cell_id": "BRAVO",
"battery_percent": 85,
"heart_rate": 72
}"#;
let parsed = parse_node_publish_json(json).expect("parse");
assert_eq!(parsed.id, "ANDROID-abc123");
assert_eq!(parsed.battery_percent, Some(85));
assert_eq!(parsed.heart_rate, Some(72));
assert_eq!(parsed.cell_id.as_deref(), Some("BRAVO"));
assert!(parsed.capabilities.contains(&"PLI".to_string()));
}
#[test]
fn publish_json_rejects_missing_id() {
let json = r#"{"name":"HOBO","node_type":"SOLDIER","lat":33.7,"lon":-84.4}"#;
assert!(parse_node_publish_json(json).is_err());
let empty_id = r#"{"id":"","name":"HOBO","lat":33.7,"lon":-84.4}"#;
assert!(parse_node_publish_json(empty_id).is_err());
}
#[test]
fn battery_and_heart_clamp_out_of_range_numbers() {
let high = serde_json::json!(9999);
assert_eq!(parse_battery_percent(&high), Some(100));
let neg = serde_json::json!(-50);
assert_eq!(parse_battery_percent(&neg), Some(0));
let huge = serde_json::json!(i64::MAX);
assert_eq!(parse_battery_percent(&huge), Some(100));
let bpm_high = serde_json::json!(500);
assert_eq!(parse_heart_rate(&bpm_high), Some(250));
let bpm_neg = serde_json::json!(-50);
assert_eq!(parse_heart_rate(&bpm_neg), Some(0));
let bpm_low_real = serde_json::json!(10);
assert_eq!(parse_heart_rate(&bpm_low_real), Some(10));
}
#[test]
fn battery_and_heart_reject_non_numeric() {
let s = serde_json::json!("85");
assert!(parse_battery_percent(&s).is_none());
assert!(parse_heart_rate(&s).is_none());
let null = serde_json::Value::Null;
assert!(parse_battery_percent(&null).is_none());
assert!(parse_heart_rate(&null).is_none());
let arr = serde_json::json!([85]);
assert!(parse_battery_percent(&arr).is_none());
}
#[test]
fn parse_silently_drops_unknown_future_fields() {
let json = r#"{
"node_type": "SOLDIER",
"name": "FUTURE-PEER",
"status": "ACTIVE",
"lat": 33.71,
"lon": -84.41,
"readiness": 1.0,
"capabilities": ["PLI"],
"cell_id": "BRAVO",
"battery_percent": 90,
"last_heartbeat": 1700000000000,
"future_v2_field_one": "should be ignored",
"future_v2_struct": { "nested": 42 },
"future_v2_array": [1, 2, 3]
}"#;
let parsed =
parse_node_json("FUTURE-PEER", json).expect("future-shaped json must parse");
assert_eq!(parsed.battery_percent, Some(90));
assert_eq!(parsed.cell_id.as_deref(), Some("BRAVO"));
}
#[test]
fn battery_accepts_float_form() {
assert_eq!(parse_battery_percent(&serde_json::json!(85.0)), Some(85));
assert_eq!(parse_battery_percent(&serde_json::json!(85.7)), Some(86));
assert_eq!(parse_battery_percent(&serde_json::json!(85.4)), Some(85));
assert_eq!(parse_battery_percent(&serde_json::json!(150.0)), Some(100));
assert_eq!(parse_battery_percent(&serde_json::json!(-10.5)), Some(0));
}
#[test]
fn heart_rate_accepts_float_form() {
assert_eq!(parse_heart_rate(&serde_json::json!(72.0)), Some(72));
assert_eq!(parse_heart_rate(&serde_json::json!(72.6)), Some(73));
assert_eq!(parse_heart_rate(&serde_json::json!(300.0)), Some(250));
}
#[test]
fn heart_rate_preserves_bradycardia_below_30() {
assert_eq!(parse_heart_rate(&serde_json::json!(25)), Some(25));
assert_eq!(parse_heart_rate(&serde_json::json!(0)), Some(0));
assert_eq!(parse_heart_rate(&serde_json::json!(-5)), Some(0));
}
#[test]
fn serialize_nodes_get_json_round_trips_through_parser() {
let original = NodeInfo {
id: "ANDROID-emit".to_string(),
node_type: "SOLDIER".to_string(),
name: "EMIT-TEST".to_string(),
status: NodeStatus::Active,
lat: 33.71576,
lon: -84.41152,
hae: Some(305.0),
readiness: 1.0,
capabilities: vec!["PLI".to_string()],
cell_id: Some("BRAVO".to_string()),
battery_percent: Some(85),
heart_rate: Some(72),
last_heartbeat: 1_700_000_000_000,
};
let emitted = serialize_nodes_get_json(std::slice::from_ref(&original));
let arr: Vec<serde_json::Value> = serde_json::from_str(&emitted).expect("array");
assert_eq!(arr.len(), 1);
let obj_str = serde_json::to_string(&arr[0]).expect("obj");
let parsed = parse_node_json(&original.id, &obj_str).expect("parse");
assert_eq!(parsed.battery_percent, Some(85));
assert_eq!(parsed.heart_rate, Some(72));
assert_eq!(parsed.cell_id.as_deref(), Some("BRAVO"));
assert_eq!(parsed.last_heartbeat, 1_700_000_000_000);
}
#[test]
fn publish_json_honors_wire_last_heartbeat() {
let supplied: i64 = 1_700_000_123_456;
let json = format!(
r#"{{
"id": "ANDROID-replay",
"name": "REPLAY",
"node_type": "SOLDIER",
"lat": 0.0, "lon": 0.0,
"status": "ACTIVE",
"last_heartbeat": {}
}}"#,
supplied
);
let parsed = parse_node_publish_json(&json).expect("parse");
assert_eq!(parsed.last_heartbeat, supplied);
}
#[test]
fn publish_json_stamps_now_when_last_heartbeat_absent() {
let before = chrono::Utc::now().timestamp_millis();
let json = r#"{
"id": "ANDROID-no-stamp",
"name": "FRESH",
"node_type": "SOLDIER",
"lat": 0.0, "lon": 0.0,
"status": "ACTIVE"
}"#;
let parsed = parse_node_publish_json(json).expect("parse");
let after = chrono::Utc::now().timestamp_millis();
assert!(
parsed.last_heartbeat >= before && parsed.last_heartbeat <= after,
"last_heartbeat ({}) should be in [{}, {}]",
parsed.last_heartbeat,
before,
after
);
}
#[test]
fn publish_json_preserves_wire_last_heartbeat_zero_as_stale_marker() {
let json = r#"{
"id": "ANDROID-stale",
"name": "STALE",
"node_type": "SOLDIER",
"lat": 0.0, "lon": 0.0,
"status": "ACTIVE",
"last_heartbeat": 0
}"#;
let parsed = parse_node_publish_json(json).expect("parse");
assert_eq!(
parsed.last_heartbeat, 0,
"wire `last_heartbeat: 0` must pass through as the stale-record sentinel"
);
}
#[test]
fn publish_json_preserves_small_positive_last_heartbeat() {
for wire in [1_i64, 12_345, 1_700_000_000_000] {
let json = format!(
r#"{{"id":"ANDROID-{w}","name":"X","node_type":"SOLDIER","lat":0.0,"lon":0.0,"status":"ACTIVE","last_heartbeat":{w}}}"#,
w = wire,
);
let parsed = parse_node_publish_json(&json).expect("parse");
assert_eq!(
parsed.last_heartbeat, wire,
"wire `{}` must round-trip",
wire
);
}
}
#[test]
fn publish_json_clamps_far_future_last_heartbeat_to_now() {
let json = r#"{
"id": "ANDROID-malicious",
"name": "MALICIOUS",
"node_type": "SOLDIER",
"lat": 0.0, "lon": 0.0,
"status": "ACTIVE",
"last_heartbeat": 9223372036854775807
}"#;
let before = chrono::Utc::now().timestamp_millis();
let parsed = parse_node_publish_json(json).expect("parse");
let after = chrono::Utc::now().timestamp_millis();
assert!(
parsed.last_heartbeat >= before && parsed.last_heartbeat <= after,
"i64::MAX must clamp to now(), got {}",
parsed.last_heartbeat
);
}
#[test]
fn publish_json_clamps_negative_last_heartbeat_to_zero() {
for wire in [-1_i64, -1_700_000_000_000, i64::MIN] {
let json = format!(
r#"{{"id":"ANDROID-neg-{w}","name":"NEG","node_type":"SOLDIER","lat":0.0,"lon":0.0,"status":"ACTIVE","last_heartbeat":{w}}}"#,
w = wire,
);
let parsed = parse_node_publish_json(&json)
.unwrap_or_else(|e| panic!("wire {} must parse: {:?}", wire, e));
assert_eq!(
parsed.last_heartbeat, 0,
"negative wire `{}` must collapse to stale-marker `0`",
wire
);
}
}
#[test]
fn publish_json_within_grace_window_passes_through_then_clamps_beyond() {
let now = chrono::Utc::now().timestamp_millis();
let in_grace = now + 30_000;
let json = format!(
r#"{{"id":"ANDROID-grace","name":"G","node_type":"SOLDIER","lat":0.0,"lon":0.0,"status":"ACTIVE","last_heartbeat":{}}}"#,
in_grace
);
let parsed = parse_node_publish_json(&json).expect("parse");
assert_eq!(parsed.last_heartbeat, in_grace);
let beyond = chrono::Utc::now().timestamp_millis() + 5 * 60 * 1000;
let json2 = format!(
r#"{{"id":"ANDROID-skew","name":"S","node_type":"SOLDIER","lat":0.0,"lon":0.0,"status":"ACTIVE","last_heartbeat":{}}}"#,
beyond
);
let parsed2 = parse_node_publish_json(&json2).expect("parse");
assert!(
parsed2.last_heartbeat < beyond,
"5min-future must clamp ({} should be << {})",
parsed2.last_heartbeat,
beyond
);
}
#[test]
fn battery_percent_rounds_halves_away_from_zero() {
assert_eq!(parse_battery_percent(&serde_json::json!(85.5)), Some(86));
assert_eq!(parse_battery_percent(&serde_json::json!(84.5)), Some(85));
assert_eq!(parse_battery_percent(&serde_json::json!(0.5)), Some(1));
}
#[test]
fn publish_json_silently_drops_unknown_future_fields() {
let json = r#"{
"id": "ANDROID-future",
"name": "FUTURE",
"node_type": "SOLDIER",
"lat": 33.71, "lon": -84.41,
"status": "ACTIVE",
"battery_percent": 90,
"future_v2_field_one": "should be ignored",
"future_v2_struct": { "nested": 42 },
"future_v2_array": [1, 2, 3]
}"#;
let parsed = parse_node_publish_json(json).expect("future-shaped publish must parse");
assert_eq!(parsed.battery_percent, Some(90));
assert_eq!(parsed.id, "ANDROID-future");
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
mod track_tests {
use super::*;
use peat_protocol::sync::ble_translation::{
value_to_mesh_document, BlePosition, BleTranslator,
};
struct TrackFixture {
node: Arc<PeatNode>,
#[allow(dead_code)]
_tmp: tempfile::TempDir,
}
fn ingest_position_test_node() -> TrackFixture {
let tmp = tempfile::tempdir().expect("tempdir");
let path = tmp.path().to_str().expect("tempdir path utf-8").to_string();
let node = create_node(NodeConfig {
app_id: "track-rt-test".to_string(),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: path,
transport: None,
})
.expect("create_node");
TrackFixture { node, _tmp: tmp }
}
#[test]
fn put_track_get_tracks_preserves_body() {
let fx = ingest_position_test_node();
let pn = &fx.node;
let original = TrackInfo {
id: "manual-001".to_string(),
source_node: "ANDROID-tablet".to_string(),
cell_id: Some("BRAVO".to_string()),
formation_id: None,
lat: 33.71576,
lon: -84.41152,
hae: Some(305.0),
cep: Some(5.0),
heading: Some(87.5),
speed: Some(1.2),
classification: "a-f-G-U-C-I".to_string(),
confidence: 0.9,
category: TrackCategory::Person,
created_at: 1_700_000_000_000,
last_update: 1_700_000_000_000,
attributes: std::collections::HashMap::new(),
};
pn.put_track(original.clone()).expect("put_track");
let listed = pn.get_tracks().expect("get_tracks");
let found = listed
.iter()
.find(|t| t.id == "manual-001")
.expect("track must appear");
assert!(
(found.lat - original.lat).abs() < 1e-9,
"lat dropped via put_track/get_tracks: got {}",
found.lat
);
assert!(
(found.lon - original.lon).abs() < 1e-9,
"lon dropped via put_track/get_tracks: got {}",
found.lon
);
assert_eq!(found.cell_id.as_deref(), Some("BRAVO"));
assert_eq!(found.source_node, original.source_node);
assert_eq!(found.classification, original.classification);
}
#[test]
fn ingest_position_via_translator_then_get_tracks_preserves_body() {
let fx = ingest_position_test_node();
let pn = &fx.node;
let translator = BleTranslator::with_defaults();
const PERIPHERAL: u32 = 0xCAFE_0001;
let position = BlePosition {
latitude: 33.71576,
longitude: -84.41152,
altitude: Some(305.0),
accuracy: Some(5.0),
};
let value = translator.position_to_track_in_cell(
&position,
PERIPHERAL,
Some("SCOUT-CAFE"),
Some("BRAVO"),
);
let doc = value_to_mesh_document(value);
pn.runtime.block_on(async {
pn.node
.publish_with_origin(
translator.tracks_collection(),
doc,
Some("ble".to_string()),
)
.await
.expect("publish_with_origin");
});
let tracks = pn.get_tracks().expect("get_tracks");
let found = tracks
.iter()
.find(|t| t.id.contains("CAFE0001"))
.expect("BLE-bridged track must appear in get_tracks output");
assert!(
(found.lat - 33.71576).abs() < 1e-4,
"peat#832: lat dropped — got {} (expected ~33.71576)",
found.lat
);
assert!(
(found.lon - (-84.41152)).abs() < 1e-4,
"peat#832: lon dropped — got {} (expected ~-84.41152)",
found.lon
);
assert_eq!(
found.cell_id.as_deref(),
Some("BRAVO"),
"peat#832: cell_id dropped"
);
assert!(
!found.source_node.is_empty() && found.source_node != "unknown",
"peat#832: source_node reverted to default — got {:?}",
found.source_node
);
assert_ne!(
found.classification, "a-u-G",
"peat#832: classification reverted to default a-u-G"
);
}
#[test]
fn ingest_position_then_get_track_single_id_preserves_body() {
let fx = ingest_position_test_node();
let pn = &fx.node;
let translator = BleTranslator::with_defaults();
const PERIPHERAL: u32 = 0xCAFE_0002;
let position = BlePosition {
latitude: 33.71576,
longitude: -84.41152,
altitude: Some(305.0),
accuracy: Some(5.0),
};
let value = translator.position_to_track_in_cell(
&position,
PERIPHERAL,
Some("SCOUT-ID-2"),
Some("BRAVO"),
);
let track_id = value
.get("id")
.and_then(|v| v.as_str())
.expect("translator stamps id")
.to_string();
let doc = value_to_mesh_document(value);
pn.runtime.block_on(async {
pn.node
.publish_with_origin(
translator.tracks_collection(),
doc,
Some("ble".to_string()),
)
.await
.expect("publish_with_origin");
});
let single = pn
.get_track(&track_id)
.expect("get_track")
.expect("track must exist for known id");
assert!((single.lat - 33.71576).abs() < 1e-4);
assert!((single.lon - (-84.41152)).abs() < 1e-4);
assert_eq!(single.cell_id.as_deref(), Some("BRAVO"));
assert_eq!(single.id, track_id);
}
#[test]
fn pre_fix_flat_json_entries_are_silently_dropped_not_crashed() {
let fx = ingest_position_test_node();
let pn = &fx.node;
let legacy = serde_json::json!({
"source_node": "ble-DEAD0001",
"lat": 33.0,
"lon": -84.0,
"classification": "a-f-G-U-C-I",
"confidence": 0.9,
"category": "PERSON",
"created_at": 1_700_000_000_000_i64,
"last_update": 1_700_000_000_000_i64,
})
.to_string()
.into_bytes();
let coll = pn.storage_backend.collection(collections::TRACKS);
coll.upsert("legacy-track-DEAD0001", legacy)
.expect("legacy upsert must succeed");
let listed = pn.get_tracks().expect("get_tracks must not panic");
assert!(
listed.iter().all(|t| t.id != "legacy-track-DEAD0001"),
"pre-fix legacy entry must be silently invisible after migration: {:?}",
listed.iter().map(|t| &t.id).collect::<Vec<_>>()
);
}
}
mod marker_tombstone {
use super::*;
#[test]
fn parse_minimal_tombstone() {
let json = r#"{"uid":"abc-123","_deleted":true,"ts":1700000000000}"#;
let m = parse_marker_publish_json("", json).expect("minimal tombstone parses");
assert!(m.deleted, "deleted flag set");
assert_eq!(m.uid, "abc-123");
assert_eq!(m.ts, 1700000000000);
}
#[test]
fn parse_live_marker_requires_geo() {
let no_type = r#"{"uid":"x","lat":1.0,"lon":2.0}"#;
assert!(parse_marker_publish_json("", no_type).is_err());
let no_lat = r#"{"uid":"x","type":"a-f-G","lon":2.0}"#;
assert!(parse_marker_publish_json("", no_lat).is_err());
let no_lon = r#"{"uid":"x","type":"a-f-G","lat":1.0}"#;
assert!(parse_marker_publish_json("", no_lon).is_err());
let ok = r#"{"uid":"x","type":"a-f-G","lat":1.0,"lon":2.0}"#;
let m = parse_marker_publish_json("", ok).expect("live marker parses");
assert!(!m.deleted);
}
#[test]
fn serialize_tombstone_includes_deleted_key() {
let m = MarkerInfo {
uid: "abc-123".to_string(),
marker_type: "a-u-G".to_string(),
lat: 0.0,
lon: 0.0,
hae: None,
ts: 1700000000000,
callsign: None,
color: None,
cell_id: None,
deleted: true,
};
let json = serialize_marker_json(&m).expect("serializes");
assert!(
json.contains("\"_deleted\":true"),
"tombstone serialization must include _deleted key, got: {json}"
);
}
#[test]
fn serialize_live_marker_omits_deleted_key() {
let m = MarkerInfo {
uid: "abc-123".to_string(),
marker_type: "a-f-G-U-C".to_string(),
lat: 33.71,
lon: -84.41,
hae: Some(312.4),
ts: 1700000000000,
callsign: Some("ALPHA-1".to_string()),
color: Some(-65536),
cell_id: None,
deleted: false,
};
let json = serialize_marker_json(&m).expect("serializes");
assert!(
!json.contains("_deleted"),
"live marker must not emit _deleted key, got: {json}"
);
}
#[test]
fn scan_serializes_tombstones_in_array() {
let live = MarkerInfo {
uid: "live".to_string(),
marker_type: "a-f-G".to_string(),
lat: 1.0,
lon: 2.0,
hae: None,
ts: 1,
callsign: None,
color: None,
cell_id: None,
deleted: false,
};
let dead = MarkerInfo {
deleted: true,
..live.clone()
};
let mut dead = dead;
dead.uid = "dead".to_string();
let json = serialize_markers_get_json(&[live, dead]);
let arr: serde_json::Value = serde_json::from_str(&json).unwrap();
let arr = arr.as_array().unwrap();
assert_eq!(arr.len(), 2);
let live_obj = arr.iter().find(|v| v["uid"] == "live").unwrap();
let dead_obj = arr.iter().find(|v| v["uid"] == "dead").unwrap();
assert!(
live_obj.get("_deleted").is_none(),
"live entry has no _deleted"
);
assert_eq!(
dead_obj["_deleted"].as_bool(),
Some(true),
"dead entry has _deleted: true"
);
}
#[test]
fn tombstone_round_trip_is_stable() {
let m = MarkerInfo {
uid: "round-trip-uid".to_string(),
marker_type: "a-u-G".to_string(),
lat: 0.0,
lon: 0.0,
hae: None,
ts: 1700000000000,
callsign: None,
color: None,
cell_id: None,
deleted: true,
};
let s1 = serialize_marker_json(&m).unwrap();
let parsed = parse_marker_publish_json("", &s1).expect("parses tombstone");
assert!(parsed.deleted, "deleted flag preserved through round-trip");
assert_eq!(parsed.uid, m.uid);
let s2 = serialize_marker_json(&parsed).unwrap();
assert_eq!(s1, s2, "round-trip must produce byte-identical output");
}
}
#[cfg(feature = "sync")]
mod marker_tests {
use super::*;
fn live_marker(uid: &str) -> MarkerInfo {
MarkerInfo {
uid: uid.to_string(),
marker_type: "a-f-G-U-C".to_string(),
lat: 33.71576,
lon: -84.41152,
hae: Some(312.4),
ts: 1_700_000_000_000,
callsign: Some("ALPHA-1".to_string()),
color: Some(-65536),
cell_id: Some("BRAVO".to_string()),
deleted: false,
}
}
fn tombstone_marker(uid: &str) -> MarkerInfo {
MarkerInfo {
uid: uid.to_string(),
marker_type: TOMBSTONE_PLACEHOLDER_TYPE.to_string(),
lat: 0.0,
lon: 0.0,
hae: None,
ts: 1_700_000_000_000,
callsign: None,
color: None,
cell_id: None,
deleted: true,
}
}
fn make_node(label: &str) -> Arc<PeatNode> {
let tmp = tempfile::tempdir().expect("tempdir");
create_node(NodeConfig {
app_id: format!("marker-rt-{label}"),
shared_key: "dGVzdC1rZXktMTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0".to_string(),
bind_address: Some("127.0.0.1:0".to_string()),
storage_path: tmp.path().to_str().unwrap().to_string(),
transport: None,
})
.expect("create_node")
}
#[test]
fn put_marker_get_markers_preserves_live_fields() {
let node = make_node("live");
let original = live_marker("marker-live-001");
node.put_marker(original.clone()).expect("put_marker");
let listed = node.get_markers().expect("get_markers");
let found = listed
.iter()
.find(|m| m.uid == original.uid)
.expect("published marker must appear in get_markers");
assert_eq!(found.marker_type, original.marker_type);
assert_eq!(found.lat, original.lat);
assert_eq!(found.lon, original.lon);
assert_eq!(found.hae, original.hae);
assert_eq!(found.ts, original.ts);
assert_eq!(found.callsign, original.callsign);
assert_eq!(found.color, original.color);
assert_eq!(found.cell_id, original.cell_id);
assert!(!found.deleted, "live marker must not arrive deleted");
}
#[test]
fn put_marker_get_markers_preserves_tombstone() {
let node = make_node("tomb");
let original = tombstone_marker("marker-tomb-001");
node.put_marker(original.clone()).expect("put_marker");
let listed = node.get_markers().expect("get_markers");
let found = listed
.iter()
.find(|m| m.uid == original.uid)
.expect("published tombstone must appear in get_markers");
assert!(found.deleted, "tombstone must round-trip with deleted=true");
assert_eq!(found.uid, original.uid);
assert_eq!(found.ts, original.ts);
}
#[test]
fn tombstone_upserts_over_live_marker() {
let node = make_node("upsert");
let uid = "marker-upsert-001";
node.put_marker(live_marker(uid)).expect("put live");
node.put_marker(tombstone_marker(uid)).expect("put tomb");
let listed = node.get_markers().expect("get_markers");
let matching: Vec<_> = listed.iter().filter(|m| m.uid == uid).collect();
assert_eq!(
matching.len(),
1,
"upsert must produce exactly one entry per uid, got {}",
matching.len()
);
assert!(matching[0].deleted, "tombstone must win over prior live");
}
#[test]
fn publish_json_inline_parser_extracts_live_marker_fields() {
let json = r#"{
"uid": "marker-jni-001",
"type": "a-f-G-U-C",
"lat": 33.71576,
"lon": -84.41152,
"hae": 312.4,
"ts": 1700000000000,
"callsign": "ALPHA-1",
"color": -65536,
"cell_id": "BRAVO"
}"#;
let parsed = parse_marker_publish_json("", json).expect("parse");
assert_eq!(parsed.uid, "marker-jni-001");
assert_eq!(parsed.marker_type, "a-f-G-U-C");
assert_eq!(parsed.lat, 33.71576);
assert_eq!(parsed.lon, -84.41152);
assert_eq!(parsed.hae, Some(312.4));
assert_eq!(parsed.callsign.as_deref(), Some("ALPHA-1"));
assert_eq!(parsed.color, Some(-65536));
assert_eq!(parsed.cell_id.as_deref(), Some("BRAVO"));
assert!(!parsed.deleted);
}
#[test]
fn publish_json_inline_parser_accepts_stripped_tombstone() {
let json = r#"{"uid":"marker-jni-tomb-001","_deleted":true,"ts":1700000000000}"#;
let parsed = parse_marker_publish_json("", json).expect("parse stripped tombstone");
assert!(parsed.deleted);
assert_eq!(parsed.uid, "marker-jni-tomb-001");
assert_eq!(parsed.ts, 1_700_000_000_000);
assert_eq!(
parsed.marker_type, TOMBSTONE_PLACEHOLDER_TYPE,
"absent type must resolve to the named placeholder, not a magic literal"
);
}
#[test]
fn get_markers_jni_serialized_shape_re_parses_cleanly() {
let node = make_node("getjni");
node.put_marker(live_marker("marker-getjni-001"))
.expect("put live");
node.put_marker(tombstone_marker("marker-getjni-002"))
.expect("put tomb");
let listed = node.get_markers().expect("get_markers");
let json = serialize_markers_get_json(&listed);
let arr: serde_json::Value = serde_json::from_str(&json).expect("valid JSON");
for obj in arr.as_array().expect("array").iter() {
let body = serde_json::to_string(obj).unwrap();
let parsed = parse_marker_publish_json("", &body).expect("get-side body re-parses");
if parsed.uid == "marker-getjni-002" {
assert!(parsed.deleted, "tombstone preserved in scan output");
} else {
assert!(!parsed.deleted, "live preserved in scan output");
}
}
}
}
}
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_peatVersion(
mut env: JNIEnv,
_class: JClass,
) -> jstring {
let version = peat_version();
env.new_string(&version)
.expect("Failed to create Java string")
.into_raw()
}
#[cfg(target_os = "android")]
static ANDROID_CONTEXT_GLOBAL_REF: std::sync::Mutex<Option<jni::objects::GlobalRef>> =
std::sync::Mutex::new(None);
#[cfg(target_os = "android")]
static IROH_STARTED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
#[cfg(target_os = "android")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_setAndroidContextJni(
env: JNIEnv,
_class: JClass,
context: jni::objects::JObject,
) {
use std::sync::atomic::Ordering;
if IROH_STARTED.load(Ordering::Acquire) {
android_log(
"setAndroidContextJni: ignoring — iroh already started; \
call this from Application.onCreate BEFORE createNodeJni. \
See PeatJni.kt KDoc.",
);
return;
}
let global_ref = match env.new_global_ref(&context) {
Ok(gref) => gref,
Err(e) => {
android_log(&format!(
"setAndroidContextJni: env.new_global_ref(context) failed: {}",
e
));
return;
}
};
let vm_ptr = match env.get_java_vm() {
Ok(vm) => vm.get_java_vm_pointer() as *mut c_void,
Err(_) => {
android_log("setAndroidContextJni: env.get_java_vm() failed");
return;
}
};
let ctx_ptr = global_ref.as_raw() as *mut c_void;
let mut slot = ANDROID_CONTEXT_GLOBAL_REF.lock().unwrap();
unsafe {
ndk_context::release_android_context();
ndk_context::initialize_android_context(vm_ptr, ctx_ptr);
}
*slot = Some(global_ref);
drop(slot);
android_log(
"setAndroidContextJni: ndk_context re-initialized with non-null Context (GlobalRef pinned)",
);
}
#[cfg(target_os = "android")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_verifyAndroidContextJni(
_env: JNIEnv,
_class: JClass,
) -> jni::sys::jboolean {
let stored = ndk_context::android_context().context();
if stored.is_null() {
jni::sys::JNI_FALSE
} else {
jni::sys::JNI_TRUE
}
}
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_testJni(
mut env: JNIEnv,
_class: JClass,
) -> jstring {
let msg = "JNI bindings working! Peat FFI loaded successfully.";
env.new_string(msg)
.expect("Failed to create Java string")
.into_raw()
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_createNodeJni(
mut env: JNIEnv,
_class: JClass,
app_id: JString,
shared_key: JString,
storage_path: JString,
) -> i64 {
let app_id: String = match env.get_string(&app_id) {
Ok(s) => s.into(),
Err(_) => return 0,
};
let shared_key: String = match env.get_string(&shared_key) {
Ok(s) => s.into(),
Err(_) => return 0,
};
let storage_path: String = match env.get_string(&storage_path) {
Ok(s) => s.into(),
Err(_) => return 0,
};
#[cfg(target_os = "android")]
android_log(&format!(
"createNodeJni: app_id={}, storage_path={}",
app_id, storage_path
));
let config = NodeConfig {
app_id,
shared_key,
bind_address: None,
storage_path,
transport: None,
};
match create_node(config) {
Ok(node) => {
#[cfg(target_os = "android")]
android_log("createNodeJni: Node created successfully");
#[cfg(target_os = "android")]
IROH_STARTED.store(true, std::sync::atomic::Ordering::Release);
set_global_node_handle(&node);
let handle = Arc::into_raw(node) as i64;
#[cfg(target_os = "android")]
android_log(&format!("createNodeJni: Stored global handle: {}", handle));
handle
}
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("createNodeJni: Error creating node: {:?}", e));
0
}
}
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_createNodeWithConfigJni(
mut env: JNIEnv,
_class: JClass,
app_id: JString,
shared_key: JString,
storage_path: JString,
enable_ble: jboolean,
ble_power_profile: JString,
) -> i64 {
let app_id: String = match env.get_string(&app_id) {
Ok(s) => s.into(),
Err(_) => return 0,
};
let shared_key: String = match env.get_string(&shared_key) {
Ok(s) => s.into(),
Err(_) => return 0,
};
let storage_path: String = match env.get_string(&storage_path) {
Ok(s) => s.into(),
Err(_) => return 0,
};
let power_profile: Option<String> = env.get_string(&ble_power_profile).ok().and_then(|s| {
let s: String = s.into();
if s.is_empty() {
None
} else {
Some(s)
}
});
#[cfg(target_os = "android")]
android_log(&format!(
"createNodeWithConfigJni: app_id={}, storage_path={}, enable_ble={}, power_profile={:?}",
app_id,
storage_path,
enable_ble != 0,
power_profile
));
let transport_config = if enable_ble != 0 {
Some(TransportConfigFFI {
enable_ble: true,
ble_mesh_id: None, ble_power_profile: power_profile,
transport_preference: None,
collection_routes_json: None,
})
} else {
None
};
let config = NodeConfig {
app_id,
shared_key,
bind_address: None,
storage_path,
transport: transport_config,
};
match create_node(config) {
Ok(node) => {
#[cfg(target_os = "android")]
android_log("createNodeWithConfigJni: Node created successfully");
#[cfg(target_os = "android")]
IROH_STARTED.store(true, std::sync::atomic::Ordering::Release);
set_global_node_handle(&node);
let handle = Arc::into_raw(node) as i64;
#[cfg(target_os = "android")]
android_log(&format!(
"createNodeWithConfigJni: Stored global handle: {}",
handle
));
handle
}
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"createNodeWithConfigJni: Error creating node: {:?}",
e
));
0
}
}
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_getGlobalNodeHandleJni(
_env: JNIEnv,
_class: JClass,
) -> i64 {
match GLOBAL_NODE_HANDLE.lock() {
Ok(handle) => {
let h = *handle;
#[cfg(target_os = "android")]
android_log(&format!("getGlobalNodeHandleJni: Returning handle: {}", h));
h
}
Err(_) => 0,
}
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_clearGlobalNodeHandleJni(
_env: JNIEnv,
_class: JClass,
) {
clear_owning_node_slot(&GLOBAL_NODE_HANDLE);
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_nodeIdJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
) -> jstring {
if handle == 0 {
return env
.new_string("")
.expect("Failed to create Java string")
.into_raw();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let node_id = node.node_id();
std::mem::forget(node);
env.new_string(&node_id)
.expect("Failed to create Java string")
.into_raw()
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_peerCountJni(
_env: JNIEnv,
_class: JClass,
handle: i64,
) -> i32 {
if handle == 0 {
return 0;
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let count = node.peer_count() as i32;
std::mem::forget(node);
count
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_requestSyncJni(
_env: JNIEnv,
_class: JClass,
handle: i64,
) -> jboolean {
if handle == 0 {
return 0;
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = node.request_sync().is_ok();
std::mem::forget(node);
result as jboolean
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_endpointSocketAddrJni(
env: JNIEnv,
_class: JClass,
handle: i64,
) -> jstring {
if handle == 0 {
return std::ptr::null_mut();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let addr = node.endpoint_socket_addr();
std::mem::forget(node);
match addr {
Some(s) => env
.new_string(s)
.map(|js| js.into_raw())
.unwrap_or(std::ptr::null_mut()),
None => std::ptr::null_mut(),
}
}
#[cfg(feature = "sync")]
fn serialize_document_for_get_jni(doc: &peat_mesh::sync::Document) -> String {
let mut obj = serde_json::Map::new();
for (k, v) in &doc.fields {
obj.insert(k.clone(), v.clone());
}
if let Some(id) = &doc.id {
obj.insert("id".to_string(), serde_json::Value::String(id.clone()));
}
serde_json::Value::Object(obj).to_string()
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_getDocumentJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
collection: JString,
doc_id: JString,
) -> jstring {
if handle == 0 {
return std::ptr::null_mut();
}
if FORCE_STORE_ERROR_FOR_TESTING.swap(false, std::sync::atomic::Ordering::SeqCst) {
let _ = env.throw_new(
"java/lang/RuntimeException",
"getDocumentJni: forced store error (test fault injection)",
);
return std::ptr::null_mut();
}
let collection_str: String = match env.get_string(&collection) {
Ok(s) => s.into(),
Err(_) => return std::ptr::null_mut(),
};
let doc_id_str: String = match env.get_string(&doc_id) {
Ok(s) => s.into(),
Err(_) => return std::ptr::null_mut(),
};
let node_owner = unsafe { Arc::from_raw(handle as *const PeatNode) };
let mesh_node = Arc::clone(&node_owner.node);
let runtime = Arc::clone(&node_owner.runtime);
std::mem::forget(node_owner);
let result = runtime.block_on(mesh_node.get(&collection_str, &doc_id_str));
match result {
Ok(Some(doc)) => {
let json = serialize_document_for_get_jni(&doc);
env.new_string(json)
.map(|js| js.into_raw())
.unwrap_or(std::ptr::null_mut())
}
Ok(None) => std::ptr::null_mut(),
Err(e) => {
let msg = format!("getDocumentJni: store read failed: {e}");
let _ = env.throw_new("java/lang/RuntimeException", &msg);
std::ptr::null_mut()
}
}
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_forceStoreErrorForTestingJni(
_env: JNIEnv,
_class: JClass,
handle: i64,
) -> jboolean {
if handle == 0 {
return 0;
}
FORCE_STORE_ERROR_FOR_TESTING.store(true, std::sync::atomic::Ordering::SeqCst);
1
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_connectedPeersJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
) -> jstring {
if handle == 0 {
return env
.new_string("[]")
.expect("Failed to create Java string")
.into_raw();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let peers = node.connected_peers();
let result = serde_json::to_string(&peers).unwrap_or_else(|_| "[]".to_string());
std::mem::forget(node);
env.new_string(&result)
.expect("Failed to create Java string")
.into_raw()
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_startSyncJni(
_env: JNIEnv,
_class: JClass,
handle: i64,
) -> bool {
eprintln!("startSyncJni: CALLED with handle={}", handle);
#[cfg(target_os = "android")]
android_log(&format!("startSyncJni: ENTERED with handle={}", handle));
if handle == 0 {
#[cfg(target_os = "android")]
android_log("startSyncJni: handle is 0, returning false");
return false;
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
#[cfg(target_os = "android")]
android_log("startSyncJni: calling node.start_sync()");
let result = match node.start_sync() {
Ok(()) => {
#[cfg(target_os = "android")]
android_log("startSyncJni: start_sync succeeded");
true
}
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("startSyncJni: start_sync failed: {}", e));
false
}
};
std::mem::forget(node);
result
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_freeNodeJni(
_env: JNIEnv,
_class: JClass,
handle: i64,
) {
if handle != 0 {
#[cfg(target_os = "android")]
android_log(&format!("freeNodeJni: Freeing node handle {}", handle));
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
node.cleanup_running.store(false, Ordering::SeqCst);
#[cfg(target_os = "android")]
android_log("freeNodeJni: Signaled cleanup task to stop");
std::thread::sleep(std::time::Duration::from_millis(100));
#[cfg(all(feature = "bluetooth", target_os = "android"))]
{
*ANDROID_BLE_TRANSPORT.lock().unwrap() = None;
android_log("freeNodeJni: Cleared ANDROID_BLE_TRANSPORT");
}
drop(node);
#[cfg(target_os = "android")]
android_log("freeNodeJni: Node dropped");
}
}
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_bleSetStartedJni(
_env: JNIEnv,
_class: JClass,
handle: i64,
started: jboolean,
) {
if handle == 0 {
android_log("bleSetStartedJni: Invalid handle (0)");
return;
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
use peat_protocol::transport::MeshTransport;
let guard = ANDROID_BLE_TRANSPORT.lock().unwrap();
if let Some(ref ble_transport) = *guard {
if started != 0 {
match node.runtime.block_on(ble_transport.start()) {
Ok(()) => android_log("bleSetStartedJni: BLE transport started"),
Err(e) => android_log(&format!("bleSetStartedJni: start failed: {}", e)),
}
} else {
match node.runtime.block_on(ble_transport.stop()) {
Ok(()) => android_log("bleSetStartedJni: BLE transport stopped"),
Err(e) => android_log(&format!("bleSetStartedJni: stop failed: {}", e)),
}
}
} else {
android_log("bleSetStartedJni: No BLE transport registered");
}
drop(guard);
std::mem::forget(node);
}
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_bleAddPeerJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
peer_id: JString,
) {
if handle == 0 {
android_log("bleAddPeerJni: Invalid handle (0)");
return;
}
let peer_id_str: String = match env.get_string(&peer_id) {
Ok(s) => s.into(),
Err(_) => {
android_log("bleAddPeerJni: Failed to get peer_id string");
return;
}
};
android_log(&format!("bleAddPeerJni: Adding peer {}", peer_id_str));
let guard = ANDROID_BLE_TRANSPORT.lock().unwrap();
if let Some(ref ble_transport) = *guard {
use peat_protocol::transport::NodeId;
ble_transport.add_reachable_peer(NodeId::new(peer_id_str));
} else {
android_log("bleAddPeerJni: No BLE transport registered");
}
}
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_bleRemovePeerJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
peer_id: JString,
) {
if handle == 0 {
android_log("bleRemovePeerJni: Invalid handle (0)");
return;
}
let peer_id_str: String = match env.get_string(&peer_id) {
Ok(s) => s.into(),
Err(_) => {
android_log("bleRemovePeerJni: Failed to get peer_id string");
return;
}
};
android_log(&format!("bleRemovePeerJni: Removing peer {}", peer_id_str));
let guard = ANDROID_BLE_TRANSPORT.lock().unwrap();
if let Some(ref ble_transport) = *guard {
use peat_protocol::transport::NodeId;
ble_transport.remove_reachable_peer(&NodeId::new(peer_id_str));
} else {
android_log("bleRemovePeerJni: No BLE transport registered");
}
}
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_bleIsAvailableJni(
_env: JNIEnv,
_class: JClass,
handle: i64,
) -> jboolean {
if handle == 0 {
android_log("bleIsAvailableJni: Invalid handle (0)");
return 0;
}
use peat_protocol::transport::Transport;
let guard = ANDROID_BLE_TRANSPORT.lock().unwrap();
let result = match guard.as_ref() {
Some(t) => {
if t.is_available() {
1
} else {
0
}
}
None => 0,
};
android_log(&format!("bleIsAvailableJni: {}", result != 0));
result
}
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_blePeerCountJni(
_env: JNIEnv,
_class: JClass,
handle: i64,
) -> jint {
if handle == 0 {
android_log("blePeerCountJni: Invalid handle (0)");
return 0;
}
let guard = ANDROID_BLE_TRANSPORT.lock().unwrap();
let count = match guard.as_ref() {
Some(t) => t.reachable_peer_count() as jint,
None => 0,
};
android_log(&format!("blePeerCountJni: {}", count));
count
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_getCellsJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
) -> jstring {
if handle == 0 {
return env
.new_string("[]")
.expect("Failed to create Java string")
.into_raw();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = match node.get_cells() {
Ok(cells) => {
let json_array: Vec<serde_json::Value> = cells
.iter()
.map(|c| {
serde_json::json!({
"id": c.id,
"name": c.name,
"status": c.status.as_str(),
"node_count": c.node_count,
"center_lat": c.center_lat,
"center_lon": c.center_lon,
"capabilities": c.capabilities,
"formation_id": c.formation_id,
"leader_id": c.leader_id,
"last_update": c.last_update,
"scenario_command": c.scenario_command,
})
})
.collect();
serde_json::to_string(&json_array).unwrap_or_else(|_| "[]".to_string())
}
Err(_) => "[]".to_string(),
};
std::mem::forget(node);
env.new_string(&result)
.expect("Failed to create Java string")
.into_raw()
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_getTracksJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
) -> jstring {
if handle == 0 {
return env
.new_string("[]")
.expect("Failed to create Java string")
.into_raw();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = match node.get_tracks() {
Ok(tracks) => {
let json_array: Vec<serde_json::Value> = tracks
.iter()
.map(|t| {
serde_json::json!({
"id": t.id,
"source_node": t.source_node,
"cell_id": t.cell_id,
"formation_id": t.formation_id,
"lat": t.lat,
"lon": t.lon,
"hae": t.hae,
"cep": t.cep,
"heading": t.heading,
"speed": t.speed,
"classification": t.classification,
"confidence": t.confidence,
"category": t.category.as_str(),
"created_at": t.created_at,
"last_update": t.last_update,
"attributes": t.attributes,
})
})
.collect();
serde_json::to_string(&json_array).unwrap_or_else(|_| "[]".to_string())
}
Err(_) => "[]".to_string(),
};
std::mem::forget(node);
env.new_string(&result)
.expect("Failed to create Java string")
.into_raw()
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_getNodesJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
) -> jstring {
if handle == 0 {
return env
.new_string("[]")
.expect("Failed to create Java string")
.into_raw();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = match node.get_nodes() {
Ok(nodes) => serialize_nodes_get_json(&nodes),
Err(_) => "[]".to_string(),
};
std::mem::forget(node);
env.new_string(&result)
.expect("Failed to create Java string")
.into_raw()
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_getCommandsJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
) -> jstring {
if handle == 0 {
return env
.new_string("[]")
.expect("Failed to create Java string")
.into_raw();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = match node.get_commands() {
Ok(commands) => {
let json_array: Vec<serde_json::Value> = commands
.iter()
.map(|c| {
serde_json::json!({
"id": c.id,
"command_type": c.command_type,
"target_id": c.target_id,
"parameters": c.parameters,
"priority": c.priority,
"status": c.status.as_str(),
"originator": c.originator,
"created_at": c.created_at,
"last_update": c.last_update,
})
})
.collect();
serde_json::to_string(&json_array).unwrap_or_else(|_| "[]".to_string())
}
Err(_) => "[]".to_string(),
};
std::mem::forget(node);
env.new_string(&result)
.expect("Failed to create Java string")
.into_raw()
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_publishNodeJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
node_json: JString,
) -> jboolean {
if handle == 0 {
#[cfg(target_os = "android")]
android_log("publishNodeJni: Invalid handle (0)");
return 0; }
let json_str: String = match env.get_string(&node_json) {
Ok(s) => s.into(),
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"publishNodeJni: Failed to get JSON string: {:?}",
e
));
return 0; }
};
#[cfg(target_os = "android")]
android_log(&format!("publishNodeJni: Received JSON: {}", json_str));
let node: NodeInfo = match parse_node_publish_json(&json_str) {
Ok(p) => p,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("publishNodeJni: {}", e));
return 0; }
};
#[cfg(target_os = "android")]
android_log(&format!(
"publishNodeJni: Publishing node id={}, name={}, lat={}, lon={}",
node.id, node.name, node.lat, node.lon
));
let peat_node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = match peat_node.put_node(node) {
Ok(_) => {
#[cfg(target_os = "android")]
android_log("publishNodeJni: Node published successfully");
1 }
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("publishNodeJni: Failed to publish: {:?}", e));
0 }
};
std::mem::forget(peat_node);
result
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_getMarkersJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
) -> jstring {
if handle == 0 {
return env
.new_string("[]")
.expect("Failed to create Java string")
.into_raw();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = match node.get_markers() {
Ok(markers) => serialize_markers_get_json(&markers),
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("getMarkersJni: get_markers failed: {:?}", e));
let _ = e;
"[]".to_string()
}
};
std::mem::forget(node);
env.new_string(&result)
.expect("Failed to create Java string")
.into_raw()
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_publishMarkerJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
marker_json: JString,
) -> jboolean {
if handle == 0 {
#[cfg(target_os = "android")]
android_log("publishMarkerJni: Invalid handle (0)");
return 0;
}
let json_str: String = match env.get_string(&marker_json) {
Ok(s) => s.into(),
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"publishMarkerJni: Failed to get JSON string: {:?}",
e
));
let _ = e;
return 0;
}
};
#[cfg(target_os = "android")]
android_log(&format!("publishMarkerJni: Received JSON: {}", json_str));
let marker: MarkerInfo = match parse_marker_publish_json("", &json_str) {
Ok(m) => m,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("publishMarkerJni: parse error: {:?}", e));
let _ = e;
return 0;
}
};
#[cfg(target_os = "android")]
if marker.deleted {
android_log(&format!(
"publishMarkerJni: Publishing TOMBSTONE for uid={}",
marker.uid
));
} else {
android_log(&format!(
"publishMarkerJni: Publishing marker uid={}, type={}, lat={}, lon={}",
marker.uid, marker.marker_type, marker.lat, marker.lon
));
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = match node.put_marker(marker) {
Ok(_) => {
#[cfg(target_os = "android")]
android_log("publishMarkerJni: Marker published successfully");
1
}
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("publishMarkerJni: Failed to publish: {:?}", e));
let _ = e;
0
}
};
std::mem::forget(node);
result
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_publishDocumentJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
collection: JString,
json: JString,
) -> jstring {
let result_str: String = if handle == 0 {
#[cfg(target_os = "android")]
android_log("publishDocumentJni: Invalid handle (0)");
String::new()
} else {
match (env.get_string(&collection), env.get_string(&json)) {
(Ok(c), Ok(j)) => {
let collection_str: String = c.into();
let json_str: String = j.into();
let node_owner = unsafe { Arc::from_raw(handle as *const PeatNode) };
let mesh_node = Arc::clone(&node_owner.node);
let runtime = Arc::clone(&node_owner.runtime);
std::mem::forget(node_owner);
#[allow(clippy::manual_unwrap_or_default)]
match runtime.block_on(publish_document_into_node(
&mesh_node,
&collection_str,
&json_str,
)) {
Ok(id) => id,
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!("publishDocumentJni: publish failed: {}", _e));
String::new()
}
}
}
(Err(_e), _) | (_, Err(_e)) => {
#[cfg(target_os = "android")]
android_log(&format!(
"publishDocumentJni: failed to read args: {:?}",
_e
));
String::new()
}
}
};
env.new_string(result_str)
.map(|s| s.into_raw())
.unwrap_or(std::ptr::null_mut())
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_publishDocumentWithOriginJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
collection: JString,
json: JString,
origin: JString,
) -> jstring {
let result_str: String = if handle == 0 {
#[cfg(target_os = "android")]
android_log("publishDocumentWithOriginJni: Invalid handle (0)");
String::new()
} else {
match (
env.get_string(&collection),
env.get_string(&json),
env.get_string(&origin),
) {
(Ok(c), Ok(j), Ok(o)) => {
let collection_str: String = c.into();
let json_str: String = j.into();
let origin_str: String = o.into();
let origin_opt = if origin_str.is_empty() {
None
} else {
Some(origin_str)
};
let node_owner = unsafe { Arc::from_raw(handle as *const PeatNode) };
let mesh_node = Arc::clone(&node_owner.node);
let runtime = Arc::clone(&node_owner.runtime);
std::mem::forget(node_owner);
#[allow(clippy::manual_unwrap_or_default)]
match runtime.block_on(publish_document_into_node_with_origin(
&mesh_node,
&collection_str,
&json_str,
origin_opt,
)) {
Ok(id) => id,
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"publishDocumentWithOriginJni: publish failed: {}",
_e
));
String::new()
}
}
}
(Err(_e), _, _) | (_, Err(_e), _) | (_, _, Err(_e)) => {
#[cfg(target_os = "android")]
android_log(&format!(
"publishDocumentWithOriginJni: failed to read args: {:?}",
_e
));
String::new()
}
}
};
env.new_string(result_str)
.map(|s| s.into_raw())
.unwrap_or(std::ptr::null_mut())
}
#[cfg(feature = "sync")]
async fn publish_document_into_node(
node: &peat_mesh::Node,
collection: &str,
json: &str,
) -> anyhow::Result<String> {
publish_document_into_node_with_origin(node, collection, json, None).await
}
#[cfg(feature = "sync")]
async fn publish_document_into_node_with_origin(
node: &peat_mesh::Node,
collection: &str,
json: &str,
origin: Option<String>,
) -> anyhow::Result<String> {
use peat_mesh::sync::types::Document;
use serde_json::Value;
let value: Value =
serde_json::from_str(json).map_err(|e| anyhow::anyhow!("invalid JSON: {}", e))?;
let mut obj = match value {
Value::Object(map) => map,
other => {
return Err(anyhow::anyhow!(
"document JSON must be an object, got {:?}",
other
))
}
};
let id = obj.remove("id").and_then(|v| match v {
Value::String(s) => Some(s),
_ => None,
});
let fields = obj.into_iter().collect();
let document = match id {
Some(id) => Document::with_id(id, fields),
None => Document::new(fields),
};
match origin {
Some(o) => {
node.publish_with_origin(collection, document, Some(o))
.await
}
None => node.publish(collection, document).await,
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_ingestPositionJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
json: JString,
) -> jstring {
let result_str: String = if handle == 0 {
#[cfg(target_os = "android")]
android_log("ingestPositionJni: Invalid handle (0)");
String::new()
} else {
match env.get_string(&json) {
Ok(j) => {
let json_str: String = j.into();
let node_owner = unsafe { Arc::from_raw(handle as *const PeatNode) };
let translator = Arc::clone(&node_owner.ble_translator);
let node = Arc::clone(&node_owner.node);
let runtime = Arc::clone(&node_owner.runtime);
std::mem::forget(node_owner);
#[allow(clippy::manual_unwrap_or_default)]
match runtime.block_on(ingest_position_via_translator(
&translator,
&node,
&json_str,
)) {
Ok(id) => id,
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!("ingestPositionJni: ingest failed: {}", _e));
String::new()
}
}
}
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!("ingestPositionJni: failed to read json: {:?}", _e));
String::new()
}
}
};
env.new_string(result_str)
.map(|s| s.into_raw())
.unwrap_or(std::ptr::null_mut())
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_ingestInboundFrameJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
collection: JString,
postcard_bytes: JByteArray,
) -> jstring {
if handle == 0 {
#[cfg(target_os = "android")]
android_log("ingestInboundFrameJni: Invalid handle (0)");
return std::ptr::null_mut();
}
let collection_str: String = match env.get_string(&collection) {
Ok(s) => s.into(),
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"ingestInboundFrameJni: failed to read collection: {:?}",
_e
));
return std::ptr::null_mut();
}
};
let bytes: Vec<u8> = match env.convert_byte_array(&postcard_bytes) {
Ok(b) => b,
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"ingestInboundFrameJni: failed to read bytes: {:?}",
_e
));
return std::ptr::null_mut();
}
};
let node_owner = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = node_owner.ingest_inbound_frame(collection_str, bytes);
std::mem::forget(node_owner);
match result {
Ok(Some(id)) => env
.new_string(id)
.map(|s| s.into_raw())
.unwrap_or(std::ptr::null_mut()),
Ok(None) => std::ptr::null_mut(),
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!("ingestInboundFrameJni: ingest failed: {}", _e));
std::ptr::null_mut()
}
}
}
#[cfg(all(feature = "sync", feature = "bluetooth", feature = "lite-bridge"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_ingestInboundLiteFrameJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
collection: JString,
envelope_bytes: JByteArray,
) -> jstring {
if handle == 0 {
#[cfg(target_os = "android")]
android_log("ingestInboundLiteFrameJni: Invalid handle (0)");
return std::ptr::null_mut();
}
let collection_str: String = match env.get_string(&collection) {
Ok(s) => s.into(),
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"ingestInboundLiteFrameJni: failed to read collection: {:?}",
_e
));
return std::ptr::null_mut();
}
};
let bytes: Vec<u8> = match env.convert_byte_array(&envelope_bytes) {
Ok(b) => b,
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"ingestInboundLiteFrameJni: failed to read bytes: {:?}",
_e
));
return std::ptr::null_mut();
}
};
let node_owner = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = node_owner.ingest_inbound_lite_frame(collection_str, bytes);
std::mem::forget(node_owner);
match result {
Ok(Some(id)) => env
.new_string(id)
.map(|s| s.into_raw())
.unwrap_or(std::ptr::null_mut()),
Ok(None) => std::ptr::null_mut(),
Err(_e) => {
#[cfg(target_os = "android")]
android_log(&format!("ingestInboundLiteFrameJni: ingest failed: {}", _e));
std::ptr::null_mut()
}
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_ingestCrdtFrameJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
collection: JString,
hex_bytes: JByteArray,
) -> i64 {
if handle == 0 {
return -1;
}
let collection_str: String = match env.get_string(&collection) {
Ok(s) => s.into(),
Err(_) => return -1,
};
let bytes: Vec<u8> = match env.convert_byte_array(&hex_bytes) {
Ok(b) => b,
Err(_) => return -1,
};
let hex = match String::from_utf8(bytes) {
Ok(s) => s,
Err(_) => return -1,
};
let node_owner = unsafe { Arc::from_raw(handle as *const PeatNode) };
let v = if collection_str == "supply" {
node_owner.crdt_counter_merge(hex)
} else {
node_owner.crdt_kv_merge(collection_str, hex);
0
};
std::mem::forget(node_owner);
v
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
async fn ingest_position_via_translator(
translator: &peat_protocol::sync::ble_translation::BleTranslator,
node: &peat_mesh::Node,
json: &str,
) -> anyhow::Result<String> {
use peat_protocol::sync::ble_translation::{value_to_mesh_document, BlePosition};
use serde_json::Value;
let value: Value = serde_json::from_str(json)
.map_err(|e| anyhow::anyhow!("invalid ingest-position JSON: {}", e))?;
let obj = value
.as_object()
.ok_or_else(|| anyhow::anyhow!("ingest-position JSON must be an object"))?;
let lat = obj
.get("lat")
.and_then(Value::as_f64)
.ok_or_else(|| anyhow::anyhow!("ingest-position: missing or non-numeric `lat`"))?
as f32;
let lon = obj
.get("lon")
.and_then(Value::as_f64)
.ok_or_else(|| anyhow::anyhow!("ingest-position: missing or non-numeric `lon`"))?
as f32;
let peripheral_id = parse_peripheral_id(obj.get("peripheral_id"))?;
let altitude = obj
.get("altitude")
.and_then(Value::as_f64)
.map(|v| v as f32);
let accuracy = obj
.get("accuracy")
.and_then(Value::as_f64)
.map(|v| v as f32);
let callsign = obj
.get("callsign")
.and_then(Value::as_str)
.map(str::to_string);
let mesh_id = obj
.get("mesh_id")
.and_then(Value::as_str)
.map(str::to_string);
let position = BlePosition {
latitude: lat,
longitude: lon,
altitude,
accuracy,
};
let value = translator.position_to_track_in_cell(
&position,
peripheral_id,
callsign.as_deref(),
mesh_id.as_deref(),
);
let doc = value_to_mesh_document(value);
node.publish_with_origin(translator.tracks_collection(), doc, Some("ble".to_string()))
.await
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
fn parse_peripheral_id(value: Option<&serde_json::Value>) -> anyhow::Result<u32> {
let raw = value.and_then(serde_json::Value::as_i64).ok_or_else(|| {
anyhow::anyhow!("ingest-position: missing or non-integer `peripheral_id`")
})?;
if (0..=u32::MAX as i64).contains(&raw) {
Ok(raw as u32)
} else if (i32::MIN as i64..=-1).contains(&raw) {
Ok((raw as i32) as u32)
} else {
Err(anyhow::anyhow!(
"ingest-position: `peripheral_id` {} out of u32 range \
(accepts [i32::MIN, u32::MAX] to handle both Kotlin Int and Long callers)",
raw
))
}
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_connectPeerJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
node_id: JString,
address: JString,
) -> jboolean {
if handle == 0 {
#[cfg(target_os = "android")]
android_log("connectPeerJni: Invalid handle (0)");
return 0;
}
let node_id_str: String = match env.get_string(&node_id) {
Ok(s) => s.into(),
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("connectPeerJni: Failed to get nodeId: {:?}", e));
return 0;
}
};
let addr_str: String = match env.get_string(&address) {
Ok(s) => s.into(),
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("connectPeerJni: Failed to get address: {:?}", e));
return 0;
}
};
#[cfg(target_os = "android")]
android_log(&format!(
"connectPeerJni: Connecting to node={}, addr={}",
node_id_str, addr_str
));
let peer_info = PeerInfo {
name: "quic-peer".to_string(),
node_id: node_id_str,
addresses: vec![addr_str],
relay_url: None,
};
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = match node.connect_peer(peer_info) {
Ok(()) => {
#[cfg(target_os = "android")]
android_log("connectPeerJni: Connected successfully");
1
}
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("connectPeerJni: Failed to connect: {:?}", e));
0
}
};
std::mem::forget(node);
result
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_subscribeDocumentChangesJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
listener: jni::objects::JObject,
) -> jboolean {
use std::sync::atomic::Ordering;
if handle == 0 {
#[cfg(target_os = "android")]
android_log("subscribeDocumentChangesJni: Invalid handle (0)");
return 0;
}
let listener_global = match env.new_global_ref(&listener) {
Ok(g) => g,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"subscribeDocumentChangesJni: new_global_ref failed: {:?}",
e
));
return 0;
}
};
{
let mut slot = DOCUMENT_CHANGE_LISTENER.lock().unwrap();
*slot = Some(listener_global);
}
DOCUMENT_SUBSCRIPTION_ACTIVE.store(false, Ordering::SeqCst);
DOCUMENT_SUBSCRIPTION_ACTIVE.store(true, Ordering::SeqCst);
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let store = Arc::clone(&node.store);
let runtime = Arc::clone(&node.runtime);
std::mem::forget(node);
runtime.spawn(async move {
let mut rx = store.subscribe_to_changes();
while DOCUMENT_SUBSCRIPTION_ACTIVE.load(Ordering::SeqCst) {
tokio::select! {
result = rx.recv() => {
match result {
Ok(doc_key) => {
let (collection, doc_id) = doc_key
.split_once(':')
.map(|(c, d)| (c.to_string(), d.to_string()))
.unwrap_or_else(|| ("default".to_string(), doc_key.clone()));
dispatch_document_change(&collection, &doc_id);
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
dispatch_document_error(&format!("lagged {} messages", n));
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
dispatch_document_error("change channel closed");
break;
}
}
}
_ = tokio::time::sleep(tokio::time::Duration::from_millis(200)) => {
}
}
}
if !DOCUMENT_SUBSCRIPTION_ACTIVE.load(Ordering::SeqCst) {
let mut slot = DOCUMENT_CHANGE_LISTENER.lock().unwrap();
*slot = None;
}
});
1 }
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_unsubscribeDocumentChangesJni(
_env: JNIEnv,
_class: JClass,
) {
use std::sync::atomic::Ordering;
DOCUMENT_SUBSCRIPTION_ACTIVE.store(false, Ordering::SeqCst);
#[cfg(target_os = "android")]
android_log("unsubscribeDocumentChangesJni: subscription marked inactive");
}
#[cfg(feature = "sync")]
fn clone_listener(slot: &Mutex<Option<GlobalRef>>) -> Option<GlobalRef> {
slot.lock().ok()?.as_ref().cloned()
}
#[cfg(feature = "sync")]
fn clone_java_vm() -> Option<jni::JavaVM> {
let raw_ptr = {
let guard = JAVA_VM.lock().ok()?;
guard.as_ref()?.get_java_vm_pointer()
};
unsafe { jni::JavaVM::from_raw(raw_ptr) }.ok()
}
#[cfg(feature = "sync")]
fn dispatch_document_change(collection: &str, doc_id: &str) {
let Some(listener) = clone_listener(&DOCUMENT_CHANGE_LISTENER) else {
return;
};
let Some(java_vm) = clone_java_vm() else {
return;
};
let mut env = match java_vm.attach_current_thread() {
Ok(e) => e,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("dispatch_document_change: attach failed: {:?}", e));
let _ = e;
return;
}
};
let collection_jstr = match env.new_string(collection) {
Ok(s) => s,
Err(_) => return,
};
let doc_id_jstr = match env.new_string(doc_id) {
Ok(s) => s,
Err(_) => return,
};
if let Err(e) = env.call_method(
&listener,
"onChange",
"(Ljava/lang/String;Ljava/lang/String;)V",
&[
JValue::Object(&collection_jstr),
JValue::Object(&doc_id_jstr),
],
) {
#[cfg(target_os = "android")]
android_log(&format!(
"dispatch_document_change: call_method failed: {:?}",
e
));
let _ = e;
let _ = env.exception_describe();
let _ = env.exception_clear();
}
}
#[cfg(feature = "sync")]
fn dispatch_document_error(message: &str) {
let Some(listener) = clone_listener(&DOCUMENT_CHANGE_LISTENER) else {
return;
};
let Some(java_vm) = clone_java_vm() else {
return;
};
let mut env = match java_vm.attach_current_thread() {
Ok(e) => e,
Err(_) => return,
};
let msg_jstr = match env.new_string(message) {
Ok(s) => s,
Err(_) => return,
};
if let Err(e) = env.call_method(
&listener,
"onError",
"(Ljava/lang/String;)V",
&[JValue::Object(&msg_jstr)],
) {
#[cfg(target_os = "android")]
android_log(&format!(
"dispatch_document_error: call_method failed: {:?}",
e
));
let _ = e;
let _ = env.exception_describe();
let _ = env.exception_clear();
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
struct QueueOutboundSink {
transport_id: &'static str,
queue: Arc<std::sync::Mutex<std::collections::VecDeque<OutboundFrame>>>,
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[async_trait::async_trait]
impl peat_mesh::transport::OutboundSink for QueueOutboundSink {
async fn send_outbound(
&self,
bytes: Vec<u8>,
ctx: &peat_mesh::transport::TranslationContext,
) -> anyhow::Result<()> {
let collection = ctx.collection.clone().unwrap_or_default();
self.queue
.lock()
.map_err(|e| anyhow::anyhow!("outbound_queue poisoned: {e}"))?
.push_back(OutboundFrame {
transport_id: self.transport_id.to_string(),
collection,
bytes,
});
Ok(())
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
impl PeatNode {
fn register_ble_fanout(
&self,
sink_factory: impl Fn(&'static str) -> Arc<dyn peat_mesh::transport::OutboundSink>,
) -> anyhow::Result<peat_mesh::transport::FanoutHandle> {
let translator_dyn: Arc<dyn peat_mesh::transport::Translator> = self.ble_translator.clone();
let ble_sink = sink_factory("ble");
let collections = vec![
self.ble_translator.tracks_collection().to_string(),
self.ble_translator.nodes_collection().to_string(),
self.ble_translator.alerts_collection().to_string(),
self.ble_translator.canned_messages_collection().to_string(),
];
#[cfg(feature = "lite-bridge")]
let lite_bridge_translator_id = peat_mesh::transport::BLE_LITE_BRIDGE;
#[cfg(feature = "lite-bridge")]
let mut collections = collections;
#[cfg(feature = "lite-bridge")]
for c in LITE_BRIDGE_COLLECTIONS {
if !collections.iter().any(|existing| existing == c) {
collections.push((*c).to_string());
}
}
let collections = collections;
self.runtime.block_on(async {
self.transport_manager
.register_translator(
translator_dyn,
ble_sink,
peat_mesh::transport::TranslatorRegistrationConfig::ble(),
)
.await?;
#[cfg(feature = "lite-bridge")]
{
let lite_translator: Arc<dyn peat_mesh::transport::Translator> = Arc::new(
CollectionGatedLiteBridge::for_ble_with_collections(LITE_BRIDGE_COLLECTIONS),
);
let lite_sink = sink_factory(lite_bridge_translator_id);
if let Err(e) = self
.transport_manager
.register_translator(
lite_translator,
lite_sink,
peat_mesh::transport::TranslatorRegistrationConfig::ble(),
)
.await
{
let _ = self.transport_manager.unregister_translator("ble").await;
return Err(e);
}
}
match self
.transport_manager
.start_fanout(Arc::clone(&self.node), collections)
{
Ok(handle) => Ok(handle),
Err(e) => {
#[cfg(feature = "lite-bridge")]
{
let _ = self
.transport_manager
.unregister_translator(lite_bridge_translator_id)
.await;
}
let _ = self.transport_manager.unregister_translator("ble").await;
Err(e)
}
}
})
}
fn relay_ble_frame(&self, transport_id: &str, collection: &str, bytes: &[u8]) {
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::time::{Duration, Instant};
const RELAY_DEDUP_TTL: Duration = Duration::from_secs(5);
const RELAY_SEEN_CAP: usize = 2048;
if collection == "nodes" {
return;
}
let active = match self.outbound_fanout.lock() {
Ok(g) => g.is_some(),
Err(e) => e.into_inner().is_some(),
};
if !active {
return;
}
let mut h = DefaultHasher::new();
transport_id.hash(&mut h);
collection.hash(&mut h);
bytes.hash(&mut h);
let key = h.finish();
let now = Instant::now();
{
let mut seen = self.relay_seen.lock().unwrap_or_else(|e| e.into_inner());
seen.retain(|_, t| now.duration_since(*t) < RELAY_DEDUP_TTL);
if seen.contains_key(&key) {
return; }
if seen.len() >= RELAY_SEEN_CAP {
seen.clear();
}
seen.insert(key, now);
}
self.outbound_queue
.lock()
.unwrap_or_else(|e| e.into_inner())
.push_back(OutboundFrame {
transport_id: transport_id.to_string(),
collection: collection.to_string(),
bytes: bytes.to_vec(),
});
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[uniffi::export]
impl PeatNode {
pub fn start_outbound_frames(&self) -> Result<(), PeatError> {
{
let guard = self
.outbound_fanout
.lock()
.map_err(|_| PeatError::SyncError {
msg: "outbound_fanout poisoned".to_string(),
})?;
if guard.is_some() {
return Ok(()); }
}
let queue = Arc::clone(&self.outbound_queue);
let handle = self
.register_ble_fanout(move |tid| {
Arc::new(QueueOutboundSink {
transport_id: tid,
queue: Arc::clone(&queue),
})
})
.map_err(|e| PeatError::SyncError { msg: e.to_string() })?;
*self
.outbound_fanout
.lock()
.map_err(|_| PeatError::SyncError {
msg: "outbound_fanout poisoned".to_string(),
})? = Some(handle);
Ok(())
}
pub fn poll_outbound_frames(&self) -> Vec<OutboundFrame> {
let mut q = self
.outbound_queue
.lock()
.unwrap_or_else(|e| e.into_inner());
q.drain(..).collect()
}
pub fn stop_outbound_frames(&self) {
let handle = self
.outbound_fanout
.lock()
.unwrap_or_else(|e| e.into_inner())
.take();
drop(handle);
self.outbound_queue
.lock()
.unwrap_or_else(|e| e.into_inner())
.clear();
self.runtime.block_on(async {
#[cfg(feature = "lite-bridge")]
{
let _ = self
.transport_manager
.unregister_translator(peat_mesh::transport::BLE_LITE_BRIDGE)
.await;
}
let _ = self.transport_manager.unregister_translator("ble").await;
});
}
pub fn ingest_inbound_frame(
&self,
collection: String,
postcard_bytes: Vec<u8>,
) -> Result<Option<String>, PeatError> {
use peat_mesh::transport::{TranslationContext, Translator};
let ctx = TranslationContext::inbound("ble").with_collection(collection);
let doc = self
.runtime
.block_on(self.ble_translator.decode_inbound(&postcard_bytes, &ctx))
.map_err(|e| PeatError::SyncError { msg: e.to_string() })?;
let Some(mesh_doc) = doc else {
return Ok(None);
};
let collection_name = ctx.collection.unwrap_or_default();
let id = self
.runtime
.block_on(self.node.publish_with_origin(
&collection_name,
mesh_doc,
Some("ble".to_string()),
))
.map_err(|e| PeatError::SyncError { msg: e.to_string() })?;
self.relay_ble_frame("ble", &collection_name, &postcard_bytes);
Ok(Some(id.to_string()))
}
#[cfg(all(feature = "sync", feature = "bluetooth", feature = "lite-bridge"))]
pub fn ingest_inbound_lite_frame(
&self,
collection: String,
envelope_bytes: Vec<u8>,
) -> Result<Option<String>, PeatError> {
use peat_mesh::transport::{TranslationContext, Translator, BLE_LITE_BRIDGE};
let bridge = CollectionGatedLiteBridge::for_ble_with_collections(LITE_BRIDGE_COLLECTIONS);
let ctx = TranslationContext::inbound(BLE_LITE_BRIDGE).with_collection(collection);
let doc = self
.runtime
.block_on(bridge.decode_inbound(&envelope_bytes, &ctx))
.map_err(|e| PeatError::SyncError { msg: e.to_string() })?;
let Some(mesh_doc) = doc else {
return Ok(None);
};
let collection_name = ctx.collection.unwrap_or_default();
let id = self
.runtime
.block_on(self.node.publish_with_origin(
&collection_name,
mesh_doc,
Some(BLE_LITE_BRIDGE.to_string()),
))
.map_err(|e| PeatError::SyncError { msg: e.to_string() })?;
self.relay_ble_frame(BLE_LITE_BRIDGE, &collection_name, &envelope_bytes);
Ok(Some(id.to_string()))
}
#[cfg(feature = "sync")]
pub fn publish_document(&self, collection: String, json: String) -> Result<String, PeatError> {
self.runtime
.block_on(publish_document_into_node(&self.node, &collection, &json))
.map_err(|e| PeatError::SyncError { msg: e.to_string() })
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
struct JniOutboundSink {
transport_id: &'static str,
}
#[cfg(all(feature = "sync", feature = "bluetooth", feature = "lite-bridge"))]
struct CollectionGatedLiteBridge {
inner: peat_mesh::transport::LiteBridgeTranslator,
allowed: std::collections::HashSet<&'static str>,
}
#[cfg(all(feature = "sync", feature = "bluetooth", feature = "lite-bridge"))]
impl CollectionGatedLiteBridge {
fn for_ble_with_collections(collections: &'static [&'static str]) -> Self {
Self {
inner: peat_mesh::transport::LiteBridgeTranslator::for_ble(),
allowed: collections.iter().copied().collect(),
}
}
}
#[cfg(all(feature = "sync", feature = "bluetooth", feature = "lite-bridge"))]
#[async_trait::async_trait]
impl peat_mesh::transport::Translator for CollectionGatedLiteBridge {
fn transport_id(&self) -> &'static str {
self.inner.transport_id()
}
async fn encode_outbound(
&self,
doc: &peat_mesh::sync::types::Document,
ctx: &peat_mesh::transport::TranslationContext,
) -> Option<Vec<u8>> {
let collection = ctx.collection.as_deref()?;
if !self.allowed.contains(collection) {
return None;
}
self.inner.encode_outbound(doc, ctx).await
}
async fn decode_inbound(
&self,
bytes: &[u8],
ctx: &peat_mesh::transport::TranslationContext,
) -> anyhow::Result<Option<peat_mesh::sync::types::Document>> {
self.inner.decode_inbound(bytes, ctx).await
}
}
#[cfg(all(feature = "sync", feature = "bluetooth", feature = "lite-bridge"))]
const LITE_BRIDGE_COLLECTIONS: &[&str] =
&["markers", "demo", "nodes", "mission", "cells", "commands"];
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[async_trait::async_trait]
impl peat_mesh::transport::OutboundSink for JniOutboundSink {
async fn send_outbound(
&self,
bytes: Vec<u8>,
ctx: &peat_mesh::transport::TranslationContext,
) -> anyhow::Result<()> {
let collection = ctx.collection.as_deref().unwrap_or("");
dispatch_outbound_frame(self.transport_id, collection, &bytes);
Ok(())
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_subscribeOutboundFramesJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
listener: jni::objects::JObject,
) -> jboolean {
if handle == 0 {
#[cfg(target_os = "android")]
android_log("subscribeOutboundFramesJni: Invalid handle (0)");
return 0;
}
let listener_global = match env.new_global_ref(&listener) {
Ok(g) => g,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"subscribeOutboundFramesJni: new_global_ref failed: {:?}",
e
));
let _ = e;
return 0;
}
};
*OUTBOUND_FRAME_LISTENER.lock().unwrap() = Some(listener_global);
{
let handle_slot = OUTBOUND_FRAME_FANOUT.lock().unwrap();
if handle_slot.is_some() {
return 1;
}
}
let node_owner = unsafe { Arc::from_raw(handle as *const PeatNode) };
let final_result =
node_owner.register_ble_fanout(|tid| Arc::new(JniOutboundSink { transport_id: tid }));
std::mem::forget(node_owner);
match final_result {
Ok(fanout_handle) => {
*OUTBOUND_FRAME_FANOUT.lock().unwrap() = Some(fanout_handle);
1
}
Err(_e) => {
*OUTBOUND_FRAME_LISTENER.lock().unwrap() = None;
#[cfg(target_os = "android")]
android_log(&format!(
"subscribeOutboundFramesJni: register/start_fanout failed: {}",
_e
));
0
}
}
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_unsubscribeOutboundFramesJni(
_env: JNIEnv,
_class: JClass,
handle: i64,
) {
let _ = OUTBOUND_FRAME_FANOUT.lock().unwrap().take();
if handle != 0 {
let node_owner = unsafe { Arc::from_raw(handle as *const PeatNode) };
node_owner.runtime.block_on(async {
#[cfg(feature = "lite-bridge")]
{
let _ = node_owner
.transport_manager
.unregister_translator(peat_mesh::transport::BLE_LITE_BRIDGE)
.await;
}
let _ = node_owner
.transport_manager
.unregister_translator("ble")
.await;
});
std::mem::forget(node_owner);
}
*OUTBOUND_FRAME_LISTENER.lock().unwrap() = None;
#[cfg(target_os = "android")]
android_log("unsubscribeOutboundFramesJni: subscription torn down");
}
#[cfg(all(feature = "sync", feature = "bluetooth"))]
fn dispatch_outbound_frame(transport_id: &str, collection: &str, bytes: &[u8]) {
let Some(listener) = clone_listener(&OUTBOUND_FRAME_LISTENER) else {
return;
};
let Some(java_vm) = clone_java_vm() else {
return;
};
let mut env = match java_vm.attach_current_thread() {
Ok(e) => e,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("dispatch_outbound_frame: attach failed: {:?}", e));
let _ = e;
return;
}
};
let transport_jstr = match env.new_string(transport_id) {
Ok(s) => s,
Err(_) => return,
};
let collection_jstr = match env.new_string(collection) {
Ok(s) => s,
Err(_) => return,
};
let bytes_jarr = match env.byte_array_from_slice(bytes) {
Ok(a) => a,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"dispatch_outbound_frame: byte_array_from_slice failed: {:?}",
e
));
let _ = e;
return;
}
};
if let Err(e) = env.call_method(
&listener,
"onFrame",
"(Ljava/lang/String;Ljava/lang/String;[B)V",
&[
JValue::Object(&transport_jstr),
JValue::Object(&collection_jstr),
JValue::Object(&bytes_jarr),
],
) {
#[cfg(target_os = "android")]
android_log(&format!(
"dispatch_outbound_frame: call_method failed: {:?}",
e
));
let _ = e;
let _ = env.exception_describe();
let _ = env.exception_clear();
}
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_enableBlobTransferJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
bind_addr: JString,
) -> jboolean {
if handle == 0 {
return 0;
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let addr_str: Option<String> = if bind_addr.is_null() {
None
} else {
env.get_string(&bind_addr).ok().map(|s| s.into())
};
let bind: Option<std::net::SocketAddr> =
addr_str.and_then(|s| if s.is_empty() { None } else { s.parse().ok() });
let result = match node.enable_blob_transfer(bind) {
Ok(()) => 1,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("enableBlobTransferJni: {}", e));
0
}
};
std::mem::forget(node);
result
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_blobAddPeerJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
peer_id_hex: JString,
address: JString,
) -> jboolean {
if handle == 0 {
return 0;
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let peer_hex: String = match env.get_string(&peer_id_hex) {
Ok(s) => s.into(),
Err(_) => {
std::mem::forget(node);
return 0;
}
};
let addr: String = match env.get_string(&address) {
Ok(s) => s.into(),
Err(_) => {
std::mem::forget(node);
return 0;
}
};
let result = match node.blob_add_peer(&peer_hex, &addr) {
Ok(()) => 1,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("blobAddPeerJni: {}", e));
0
}
};
std::mem::forget(node);
result
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_blobPutJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
data: jni::objects::JByteArray,
content_type: JString,
) -> jstring {
if handle == 0 {
return std::ptr::null_mut();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let bytes = match env.convert_byte_array(&data) {
Ok(b) => b,
Err(_) => {
std::mem::forget(node);
return std::ptr::null_mut();
}
};
let ct: String = match env.get_string(&content_type) {
Ok(s) => s.into(),
Err(_) => {
std::mem::forget(node);
return std::ptr::null_mut();
}
};
let result = match node.blob_put(&bytes, &ct) {
Ok(hash) => env.new_string(&hash).ok().map(|s| s.into_raw()),
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("blobPutJni: {}", e));
None
}
};
std::mem::forget(node);
result.unwrap_or(std::ptr::null_mut())
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_blobGetJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
hash_hex: JString,
) -> jni::objects::JByteArray<'static> {
if handle == 0 {
return JByteArray::default();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let hash: String = match env.get_string(&hash_hex) {
Ok(s) => s.into(),
Err(_) => {
std::mem::forget(node);
return JByteArray::default();
}
};
let result = match node.blob_get(&hash) {
Ok(bytes) => env.byte_array_from_slice(&bytes).ok(),
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!("blobGetJni: {}", e));
None
}
};
std::mem::forget(node);
result
.map(|arr| unsafe { std::mem::transmute(arr) })
.unwrap_or(JByteArray::default())
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_blobExistsLocallyJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
hash_hex: JString,
) -> jboolean {
if handle == 0 {
return 0;
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let hash: String = match env.get_string(&hash_hex) {
Ok(s) => s.into(),
Err(_) => {
std::mem::forget(node);
return 0;
}
};
let result = if node.blob_exists_locally(&hash) {
1
} else {
0
};
std::mem::forget(node);
result
}
#[cfg(feature = "sync")]
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_blobEndpointIdJni(
mut env: JNIEnv,
_class: JClass,
handle: i64,
) -> jstring {
if handle == 0 {
return std::ptr::null_mut();
}
let node = unsafe { Arc::from_raw(handle as *const PeatNode) };
let result = match node.blob_endpoint_id() {
Some(id) => env.new_string(&id).ok().map(|s| s.into_raw()),
None => None,
};
std::mem::forget(node);
result.unwrap_or(std::ptr::null_mut())
}
#[no_mangle]
pub extern "system" fn Java_com_defenseunicorns_peat_PeatJni_nativeInit(
mut env: JNIEnv,
class: JClass,
) {
use jni::NativeMethod;
let methods: Vec<NativeMethod> = vec![
NativeMethod {
name: "peatVersion".into(),
sig: "()Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_peatVersion as *mut c_void,
},
NativeMethod {
name: "testJni".into(),
sig: "()Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_testJni as *mut c_void,
},
#[cfg(target_os = "android")]
NativeMethod {
name: "setAndroidContextJni".into(),
sig: "(Ljava/lang/Object;)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_setAndroidContextJni as *mut c_void,
},
#[cfg(target_os = "android")]
NativeMethod {
name: "verifyAndroidContextJni".into(),
sig: "()Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_verifyAndroidContextJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "createNodeJni".into(),
sig: "(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;)J".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_createNodeJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getGlobalNodeHandleJni".into(),
sig: "()J".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getGlobalNodeHandleJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "clearGlobalNodeHandleJni".into(),
sig: "()V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_clearGlobalNodeHandleJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "nodeIdJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_nodeIdJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "peerCountJni".into(),
sig: "(J)I".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_peerCountJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "connectedPeersJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_connectedPeersJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "requestSyncJni".into(),
sig: "(J)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_requestSyncJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "endpointSocketAddrJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_endpointSocketAddrJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getDocumentJni".into(),
sig: "(JLjava/lang/String;Ljava/lang/String;)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getDocumentJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "forceStoreErrorForTestingJni".into(),
sig: "(J)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_forceStoreErrorForTestingJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "startSyncJni".into(),
sig: "(J)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_startSyncJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "freeNodeJni".into(),
sig: "(J)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_freeNodeJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getCellsJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getCellsJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getTracksJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getTracksJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getNodesJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getNodesJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getCommandsJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getCommandsJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "publishNodeJni".into(),
sig: "(JLjava/lang/String;)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_publishNodeJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getMarkersJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getMarkersJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "publishMarkerJni".into(),
sig: "(JLjava/lang/String;)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_publishMarkerJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "publishDocumentJni".into(),
sig: "(JLjava/lang/String;Ljava/lang/String;)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_publishDocumentJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "publishDocumentWithOriginJni".into(),
sig: "(JLjava/lang/String;Ljava/lang/String;Ljava/lang/String;)Ljava/lang/String;"
.into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_publishDocumentWithOriginJni
as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth"))]
NativeMethod {
name: "ingestPositionJni".into(),
sig: "(JLjava/lang/String;)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_ingestPositionJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth"))]
NativeMethod {
name: "ingestInboundFrameJni".into(),
sig: "(JLjava/lang/String;[B)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_ingestInboundFrameJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", feature = "lite-bridge"))]
NativeMethod {
name: "ingestInboundLiteFrameJni".into(),
sig: "(JLjava/lang/String;[B)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_ingestInboundLiteFrameJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "connectPeerJni".into(),
sig: "(JLjava/lang/String;Ljava/lang/String;)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_connectPeerJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "createNodeWithConfigJni".into(),
sig: "(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;ZLjava/lang/String;)J"
.into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_createNodeWithConfigJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "enableBlobTransferJni".into(),
sig: "(JLjava/lang/String;)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_enableBlobTransferJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "blobAddPeerJni".into(),
sig: "(JLjava/lang/String;Ljava/lang/String;)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_blobAddPeerJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "blobPutJni".into(),
sig: "(J[BLjava/lang/String;)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_blobPutJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "blobGetJni".into(),
sig: "(JLjava/lang/String;)[B".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_blobGetJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "blobExistsLocallyJni".into(),
sig: "(JLjava/lang/String;)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_blobExistsLocallyJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "blobEndpointIdJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_blobEndpointIdJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "bleSetStartedJni".into(),
sig: "(JZ)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_bleSetStartedJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "bleAddPeerJni".into(),
sig: "(JLjava/lang/String;)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_bleAddPeerJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "bleRemovePeerJni".into(),
sig: "(JLjava/lang/String;)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_bleRemovePeerJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "bleIsAvailableJni".into(),
sig: "(J)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_bleIsAvailableJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "blePeerCountJni".into(),
sig: "(J)I".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_blePeerCountJni as *mut c_void,
},
];
if let Err(_e) = env.register_native_methods(&class, &methods) {
let _ = env.exception_describe();
let _ = env.exception_clear();
}
}
#[cfg(target_os = "android")]
fn init_android_tracing() {
use std::sync::OnceLock;
static INITIALIZED: OnceLock<()> = OnceLock::new();
INITIALIZED.get_or_init(|| {
use std::ffi::CString;
use std::fmt::Write as _;
use std::os::raw::c_char;
use tracing::field::{Field, Visit};
use tracing::{Event, Level, Subscriber};
use tracing_subscriber::layer::{Context, SubscriberExt};
use tracing_subscriber::util::SubscriberInitExt;
use tracing_subscriber::{EnvFilter, Layer};
extern "C" {
fn __android_log_write(prio: i32, tag: *const c_char, text: *const c_char) -> i32;
}
fn tag_ptr() -> *const c_char {
static TAG: OnceLock<CString> = OnceLock::new();
TAG.get_or_init(|| CString::new("PeatRust").expect("static tag"))
.as_ptr()
}
#[derive(Default)]
struct FieldStringifier(String);
impl Visit for FieldStringifier {
fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
if !self.0.is_empty() {
self.0.push(' ');
}
if field.name() == "message" {
let _ = write!(self.0, "{:?}", value);
} else {
let _ = write!(self.0, "{}={:?}", field.name(), value);
}
}
fn record_str(&mut self, field: &Field, value: &str) {
if !self.0.is_empty() {
self.0.push(' ');
}
if field.name() == "message" {
self.0.push_str(value);
} else {
let _ = write!(self.0, "{}={}", field.name(), value);
}
}
}
fn android_priority(level: &Level) -> i32 {
match *level {
Level::ERROR => 6,
Level::WARN => 5,
Level::INFO => 4,
Level::DEBUG => 3,
Level::TRACE => 2,
}
}
struct AndroidLayer;
impl<S: Subscriber> Layer<S> for AndroidLayer {
fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
let metadata = event.metadata();
let prio = android_priority(metadata.level());
let mut visitor = FieldStringifier::default();
event.record(&mut visitor);
let formatted = if visitor.0.is_empty() {
metadata.target().to_string()
} else {
format!("{}: {}", metadata.target(), visitor.0)
};
const LIMIT: usize = 3500;
let bytes = formatted.as_bytes();
let truncated: &[u8] = if bytes.len() > LIMIT {
let mut cut = LIMIT;
while cut > 0 && (bytes[cut] & 0b1100_0000) == 0b1000_0000 {
cut -= 1;
}
&bytes[..cut]
} else {
bytes
};
if let Ok(c_msg) = CString::new(truncated) {
unsafe {
__android_log_write(prio, tag_ptr(), c_msg.as_ptr());
}
}
}
}
let env_filter = EnvFilter::try_from_env("PEAT_TRACING_LEVEL")
.unwrap_or_else(|_| EnvFilter::new("info"));
let result = tracing_subscriber::registry()
.with(env_filter)
.with(AndroidLayer)
.try_init();
match result {
Ok(()) => android_log("init_android_tracing: subscriber installed"),
Err(e) => android_log(&format!(
"init_android_tracing: subscriber NOT installed (already set?): {}",
e
)),
}
});
}
#[cfg(target_os = "android")]
fn install_android_panic_hook() {
use std::sync::OnceLock;
static INSTALLED: OnceLock<()> = OnceLock::new();
INSTALLED.get_or_init(|| {
let default_hook = std::panic::take_hook();
std::panic::set_hook(Box::new(move |info| {
let payload = info
.payload()
.downcast_ref::<&str>()
.copied()
.or_else(|| info.payload().downcast_ref::<String>().map(String::as_str))
.unwrap_or("<non-string panic payload>");
let location = info
.location()
.map(|l| format!("{}:{}:{}", l.file(), l.line(), l.column()))
.unwrap_or_else(|| "<unknown location>".to_string());
let thread = std::thread::current();
let thread_name = thread.name().unwrap_or("<unnamed>");
android_log(&format!(
"PANIC in thread '{}' at {}: {}",
thread_name, location, payload
));
default_hook(info);
}));
android_log("install_android_panic_hook: panic hook installed");
});
}
#[no_mangle]
#[allow(non_snake_case)]
#[allow(clippy::not_unsafe_ptr_arg_deref)] pub extern "C" fn JNI_OnLoad(vm: *mut JavaVM, _reserved: *mut c_void) -> jint {
#[cfg(target_os = "android")]
android_log("JNI_OnLoad called for peat_ffi");
#[cfg(target_os = "android")]
init_android_tracing();
#[cfg(target_os = "android")]
install_android_panic_hook();
#[cfg(target_os = "android")]
unsafe {
ndk_context::initialize_android_context(vm as *mut c_void, std::ptr::null_mut());
android_log("JNI_OnLoad: ndk_context::initialize_android_context(vm, null) done");
}
let java_vm = unsafe {
match jni::JavaVM::from_raw(vm) {
Ok(jvm) => jvm,
Err(_) => {
#[cfg(target_os = "android")]
android_log("JNI_OnLoad: Failed to create JavaVM from raw pointer");
return jni::sys::JNI_ERR;
}
}
};
*JAVA_VM.lock().unwrap() = Some(java_vm);
let mut env = unsafe {
let mut env_ptr: *mut jni::sys::JNIEnv = std::ptr::null_mut();
let get_env_result = (**vm).GetEnv.unwrap()(
vm,
&mut env_ptr as *mut _ as *mut *mut c_void,
JNI_VERSION_1_6 as i32,
);
if get_env_result != jni::sys::JNI_OK as i32 {
#[cfg(target_os = "android")]
android_log("JNI_OnLoad: GetEnv failed");
return jni::sys::JNI_ERR;
}
match JNIEnv::from_raw(env_ptr) {
Ok(env) => env,
Err(_) => {
#[cfg(target_os = "android")]
android_log("JNI_OnLoad: JNIEnv::from_raw failed");
return jni::sys::JNI_ERR;
}
}
};
let peer_event_manager_class = "com/defenseunicorns/peat/PeerEventManager";
match env.find_class(peer_event_manager_class) {
Ok(class) => match env.new_global_ref(class) {
Ok(global_ref) => {
*PEER_EVENT_MANAGER_CLASS.lock().unwrap() = Some(global_ref);
#[cfg(target_os = "android")]
android_log("JNI_OnLoad: PeerEventManager class found and cached");
}
Err(_) => {
#[cfg(target_os = "android")]
android_log("JNI_OnLoad: Failed to create global ref for PeerEventManager");
}
},
Err(_) => {
let _ = env.exception_clear();
#[cfg(target_os = "android")]
android_log(
"JNI_OnLoad: PeerEventManager class not found (OK if loading before class init)",
);
}
}
#[cfg(target_os = "android")]
android_log("JNI_OnLoad: Got JNIEnv, looking for PeatJni class...");
let class_name = "com/defenseunicorns/peat/PeatJni";
match env.find_class(class_name) {
Ok(class) => {
#[cfg(target_os = "android")]
android_log("JNI_OnLoad: Found PeatJni class, registering natives...");
use jni::NativeMethod;
let methods: Vec<NativeMethod> = vec![
NativeMethod {
name: "nativeInit".into(),
sig: "()V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_nativeInit as *mut c_void,
},
NativeMethod {
name: "peatVersion".into(),
sig: "()Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_peatVersion as *mut c_void,
},
NativeMethod {
name: "testJni".into(),
sig: "()Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_testJni as *mut c_void,
},
#[cfg(target_os = "android")]
NativeMethod {
name: "setAndroidContextJni".into(),
sig: "(Ljava/lang/Object;)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_setAndroidContextJni
as *mut c_void,
},
#[cfg(target_os = "android")]
NativeMethod {
name: "verifyAndroidContextJni".into(),
sig: "()Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_verifyAndroidContextJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "createNodeJni".into(),
sig: "(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;)J".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_createNodeJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getGlobalNodeHandleJni".into(),
sig: "()J".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getGlobalNodeHandleJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "clearGlobalNodeHandleJni".into(),
sig: "()V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_clearGlobalNodeHandleJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "nodeIdJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_nodeIdJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "peerCountJni".into(),
sig: "(J)I".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_peerCountJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "connectedPeersJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_connectedPeersJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "requestSyncJni".into(),
sig: "(J)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_requestSyncJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "endpointSocketAddrJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_endpointSocketAddrJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getDocumentJni".into(),
sig: "(JLjava/lang/String;Ljava/lang/String;)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getDocumentJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "forceStoreErrorForTestingJni".into(),
sig: "(J)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_forceStoreErrorForTestingJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "startSyncJni".into(),
sig: "(J)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_startSyncJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "freeNodeJni".into(),
sig: "(J)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_freeNodeJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getCellsJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getCellsJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getTracksJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getTracksJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getNodesJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getNodesJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getCommandsJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getCommandsJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "getMarkersJni".into(),
sig: "(J)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_getMarkersJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "publishMarkerJni".into(),
sig: "(JLjava/lang/String;)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_publishMarkerJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "publishNodeJni".into(),
sig: "(JLjava/lang/String;)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_publishNodeJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "publishDocumentJni".into(),
sig: "(JLjava/lang/String;Ljava/lang/String;)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_publishDocumentJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "publishDocumentWithOriginJni".into(),
sig: "(JLjava/lang/String;Ljava/lang/String;Ljava/lang/String;)\
Ljava/lang/String;"
.into(),
fn_ptr:
Java_com_defenseunicorns_peat_PeatJni_publishDocumentWithOriginJni
as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth"))]
NativeMethod {
name: "ingestPositionJni".into(),
sig: "(JLjava/lang/String;)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_ingestPositionJni
as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth"))]
NativeMethod {
name: "ingestInboundFrameJni".into(),
sig: "(JLjava/lang/String;[B)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_ingestInboundFrameJni
as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", feature = "lite-bridge"))]
NativeMethod {
name: "ingestInboundLiteFrameJni".into(),
sig: "(JLjava/lang/String;[B)Ljava/lang/String;".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_ingestInboundLiteFrameJni
as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "connectPeerJni".into(),
sig: "(JLjava/lang/String;Ljava/lang/String;)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_connectPeerJni as *mut c_void,
},
#[cfg(feature = "sync")]
NativeMethod {
name: "createNodeWithConfigJni".into(),
sig: "(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;ZLjava/lang/String;)J"
.into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_createNodeWithConfigJni
as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "bleSetStartedJni".into(),
sig: "(JZ)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_bleSetStartedJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "bleAddPeerJni".into(),
sig: "(JLjava/lang/String;)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_bleAddPeerJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "bleRemovePeerJni".into(),
sig: "(JLjava/lang/String;)V".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_bleRemovePeerJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "bleIsAvailableJni".into(),
sig: "(J)Z".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_bleIsAvailableJni as *mut c_void,
},
#[cfg(all(feature = "sync", feature = "bluetooth", target_os = "android"))]
NativeMethod {
name: "blePeerCountJni".into(),
sig: "(J)I".into(),
fn_ptr: Java_com_defenseunicorns_peat_PeatJni_blePeerCountJni as *mut c_void,
},
];
match env.register_native_methods(&class, &methods) {
Ok(_) => {
#[cfg(target_os = "android")]
android_log("JNI_OnLoad: Native methods registered successfully!");
}
Err(_) => {
#[cfg(target_os = "android")]
android_log("JNI_OnLoad: Failed to register native methods");
let _ = env.exception_describe();
let _ = env.exception_clear();
}
}
}
Err(_) => {
#[cfg(target_os = "android")]
android_log(
"JNI_OnLoad: PeatJni class not found (this is OK if loading before class init)",
);
}
}
JNI_VERSION_1_6
}
#[cfg(target_os = "android")]
fn android_log(msg: &str) {
use std::ffi::CString;
use std::os::raw::c_char;
let tag = CString::new("PeatFFI").unwrap();
let msg = CString::new(msg).unwrap();
unsafe {
extern "C" {
fn __android_log_write(prio: i32, tag: *const c_char, text: *const c_char) -> i32;
}
__android_log_write(4, tag.as_ptr(), msg.as_ptr());
}
}
#[cfg(feature = "sync")]
fn notify_peer_connected(peer_id: &str) {
notify_peer_event("notifyPeerConnected", peer_id, None);
}
#[cfg(feature = "sync")]
fn notify_peer_disconnected(peer_id: &str, reason: &str) {
notify_peer_event("notifyPeerDisconnected", peer_id, Some(reason));
}
#[cfg(feature = "sync")]
fn notify_peer_event(method_name: &str, peer_id: &str, reason: Option<&str>) {
let java_vm_guard = JAVA_VM.lock().unwrap();
let java_vm = match java_vm_guard.as_ref() {
Some(vm) => vm,
None => {
#[cfg(target_os = "android")]
android_log("notify_peer_event: No JavaVM available");
return;
}
};
let mut class_guard = PEER_EVENT_MANAGER_CLASS.lock().unwrap();
if class_guard.is_none() {
#[cfg(target_os = "android")]
android_log("notify_peer_event: PeerEventManager class not cached, trying to find it...");
if let Ok(mut env) = java_vm.attach_current_thread() {
let peer_event_manager_class = "com/defenseunicorns/peat/PeerEventManager";
if let Ok(class) = env.find_class(peer_event_manager_class) {
if let Ok(global_ref) = env.new_global_ref(class) {
*class_guard = Some(global_ref);
#[cfg(target_os = "android")]
android_log("notify_peer_event: PeerEventManager class found and cached!");
}
} else {
let _ = env.exception_clear();
#[cfg(target_os = "android")]
android_log("notify_peer_event: PeerEventManager class not found");
}
}
}
let class_ref = match class_guard.as_ref() {
Some(c) => c,
None => {
#[cfg(target_os = "android")]
android_log("notify_peer_event: PeerEventManager class not available");
return;
}
};
let mut env = match java_vm.attach_current_thread() {
Ok(env) => env,
Err(e) => {
#[cfg(target_os = "android")]
android_log(&format!(
"notify_peer_event: Failed to attach thread: {:?}",
e
));
return;
}
};
let peer_id_jstring = match env.new_string(peer_id) {
Ok(s) => s,
Err(_) => {
#[cfg(target_os = "android")]
android_log("notify_peer_event: Failed to create peer_id string");
return;
}
};
let result = if let Some(reason) = reason {
let reason_jstring = match env.new_string(reason) {
Ok(s) => s,
Err(_) => {
#[cfg(target_os = "android")]
android_log("notify_peer_event: Failed to create reason string");
return;
}
};
env.call_static_method(
class_ref,
method_name,
"(Ljava/lang/String;Ljava/lang/String;)V",
&[
JValue::Object(&peer_id_jstring),
JValue::Object(&reason_jstring),
],
)
} else {
env.call_static_method(
class_ref,
method_name,
"(Ljava/lang/String;)V",
&[JValue::Object(&peer_id_jstring)],
)
};
if let Err(e) = result {
#[cfg(target_os = "android")]
android_log(&format!("notify_peer_event: Method call failed: {:?}", e));
let _ = env.exception_describe();
let _ = env.exception_clear();
} else {
#[cfg(target_os = "android")]
android_log(&format!(
"notify_peer_event: {} called for {}",
method_name, peer_id
));
}
}