peat-btle 0.4.0

// Copyright (c) 2025-2026 (r)evolve - Revolve Team LLC
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! End-to-end integration: protocol-level state propagation across nodes.
//!
//! These tests are the contract for the holistic Prone-as-unavailable
//! behavior. Bench cycles are triage; this file is regression coverage:
//! a single sender publishes a platform-document state change through
//! the translator path, a separate receiver consumes the wire bytes, and
//! the test asserts the operationally-meaningful payload survives the
//! `encode_outbound → 0xB6 frame → decode dispatcher → DataReceivedResult`
//! pipeline.
//!
//! When this file goes red on a wire-adjacent change, the failure
//! reproduces the same misbehavior the user would see on the bench
//! (SCOUT-A698 visibly stuck in OPERATIONAL after tilting to Prone)
//! without requiring a tablet, a watch, or an M5Stack — that's the gap
//! the prior test setup left wide open.

#![cfg(feature = "translator-codec")]

use peat_btle::peat_mesh::{PeatMesh, PeatMeshConfig};
use peat_btle::translator::{
    BleHealthStatus, BlePeripheral, BlePeripheralType, BlePosition, BleTranslator,
};
use peat_btle::NodeId;

use peat_mesh::sync::Document as MeshDocument;
use peat_mesh::transport::{TranslationContext, Translator};

use std::collections::HashMap;
use std::time::SystemTime;

/// Build the wire-side bytes a publisher would put on the BLE air for a
/// given peripheral state. Mirrors what a translator-using publisher
/// (M5Stack post-Slice 2, peat-ffi today) would emit — the test takes
/// the publisher role explicitly so we exercise the same encode path
/// hosts will go through, not a side-door API.
async fn frame_platforms_doc(
    translator: &BleTranslator,
    peripheral: &BlePeripheral,
    mesh_id: Option<&str>,
) -> Vec<u8> {
    let value = translator.peripheral_to_platform_in_cell(peripheral, mesh_id);
    let fields_map = value
        .as_object()
        .expect("peripheral_to_platform_in_cell returns a JSON object")
        .clone();
    let id = fields_map
        .get("id")
        .and_then(|v| v.as_str())
        .map(String::from);
    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
    for (k, v) in fields_map {
        fields.insert(k, v);
    }
    let doc = MeshDocument {
        id,
        fields,
        updated_at: SystemTime::now(),
    };
    let ctx = TranslationContext::outbound().with_collection(translator.platforms_collection());
    translator
        .encode_outbound(&doc, &ctx)
        .await
        .expect("encode_outbound returns framed bytes for platforms collection")
}

/// Convenience: receive the framed wire bytes through the same anonymous
/// path peat-atak-plugin uses (`onBleDataReceivedAnonymous` →
/// `process_document_data_with_identifier` →
/// `try_handle_translator_marker`). Returns the JSON Value of the
/// decoded platform-document fields, *unwrapped* from the
/// `MeshDocument` envelope (`{id, fields, updated_at}`) so tests can
/// assert directly on the platform-shape keys (`name`, `alerts`, etc.)
/// the plugin will read.
fn decode_received_frame(receiver: &PeatMesh, framed: &[u8]) -> serde_json::Value {
    let result = receiver
        .on_ble_data_received_anonymous("ble-test-peer", framed, 1_777_000_000_500)
        .expect("receiver surfaces a DataReceivedResult for a 0xB6 platforms frame");
    let frame = result
        .decoded_translator_frame
        .expect("DataReceivedResult.decoded_translator_frame populated for 0xB6 platforms frame");
    assert_eq!(
        frame.collection, "platforms",
        "translator frame collection must round-trip as 'platforms'",
    );
    // doc_json is the serde-serialized MeshDocument: `{id, fields,
    // updated_at}`. Plugin consumers read the platform-shape keys out
    // of `fields`, so mirror that here rather than asserting on the
    // envelope shape.
    let envelope: serde_json::Value = serde_json::from_str(&frame.doc_json)
        .expect("decoded_translator_frame.doc_json parses as JSON");
    envelope
        .get("fields")
        .cloned()
        .unwrap_or_else(|| panic!("MeshDocument envelope missing `fields` map: {envelope}"))
}

fn fresh_receiver(node_id: u32, callsign: &str) -> PeatMesh {
    PeatMesh::new(PeatMeshConfig::new(
        NodeId::new(node_id),
        callsign,
        "WEARTAK",
    ))
}

fn scout_peripheral(alerts: u8, position: Option<BlePosition>) -> BlePeripheral {
    BlePeripheral {
        id: 0x6462_A698,
        parent_node: 0,
        peripheral_type: BlePeripheralType::SoldierSensor,
        callsign: "SCOUT-A698".to_string(),
        health: BleHealthStatus {
            battery_percent: 100,
            heart_rate: None,
            // Activity stays 0 (still/standing) regardless of state —
            // Prone is signalled via the alerts bitfield per the
            // sender-side decision (peat-btle/Cargo.toml + M5Stack).
            // Receivers should not infer Prone from `activity`.
            activity: 0,
            alerts,
        },
        timestamp: 1_777_000_000_000,
        position,
    }
}

/// Standing peripheral (no alerts) round-trips with `man_down` absent
/// from the alerts list. Locks the no-false-positive case so a future
/// edit can't silently inject ALERT_MAN_DOWN into the encode path.
#[tokio::test]
async fn standing_peripheral_round_trips_without_man_down_alert() {
    let translator = BleTranslator::with_defaults();
    let receiver = fresh_receiver(0x089F_A635, "TABLET");

    let standing = scout_peripheral(0, None);
    let framed = frame_platforms_doc(&translator, &standing, Some("WEARTAK")).await;
    let payload = decode_received_frame(&receiver, &framed);

    assert_eq!(
        payload.get("id").and_then(|v| v.as_str()),
        Some("ble-6462A698"),
    );
    assert_eq!(
        payload.get("name").and_then(|v| v.as_str()),
        Some("SCOUT-A698"),
    );
    let alerts = payload
        .get("alerts")
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().filter_map(|v| v.as_str()).collect::<Vec<&str>>())
        .unwrap_or_default();
    assert!(
        !alerts.contains(&"man_down"),
        "Standing peripheral must NOT carry man_down; got alerts={:?}",
        alerts,
    );
}

/// **Holistic Prone-as-unavailable contract.** The failure mode this
/// test guards against is the live regression we hit on the bench:
/// SCOUT-A698 visibly stuck OPERATIONAL on the tablet after tilting the
/// M5Stack to Prone. When the publisher (M5Stack post-Slice 2) sets
/// `ALERT_MAN_DOWN` in the alerts bitfield, the translator-frame
/// receiver must surface `man_down` in the decoded JSON's `alerts`
/// array — that's the wire contract Slice 3 (plugin) reads to flip
/// `PeatPlatform.Status` to UNAVAILABLE/OFFLINE.
#[tokio::test]
async fn prone_peripheral_propagates_man_down_alert_to_receiver() {
    let translator = BleTranslator::with_defaults();
    let receiver = fresh_receiver(0x089F_A635, "TABLET");

    let prone = scout_peripheral(BleHealthStatus::ALERT_MAN_DOWN, None);
    let framed = frame_platforms_doc(&translator, &prone, Some("WEARTAK")).await;
    let payload = decode_received_frame(&receiver, &framed);

    let alerts = payload
        .get("alerts")
        .and_then(|v| v.as_array())
        .expect("Prone peripheral encodes alerts array on the wire");
    let alert_strings: Vec<&str> = alerts.iter().filter_map(|v| v.as_str()).collect();
    assert!(
        alert_strings.contains(&"man_down"),
        "expected man_down to round-trip through translator; got {:?}",
        alert_strings,
    );
    assert_eq!(
        payload.get("id").and_then(|v| v.as_str()),
        Some("ble-6462A698"),
    );
}

/// Compound-alert case: a peer simultaneously Prone and on a low
/// battery emits both flags. The plugin's status-derivation logic
/// (Slice 3) needs both — Prone drives availability, low_battery drives
/// a separate badge — so the wire must keep them independent.
#[tokio::test]
async fn prone_with_low_battery_propagates_both_alerts() {
    let translator = BleTranslator::with_defaults();
    let receiver = fresh_receiver(0x089F_A635, "TABLET");

    let combined = scout_peripheral(
        BleHealthStatus::ALERT_MAN_DOWN | BleHealthStatus::ALERT_LOW_BATTERY,
        None,
    );
    let framed = frame_platforms_doc(&translator, &combined, Some("WEARTAK")).await;
    let payload = decode_received_frame(&receiver, &framed);

    let alert_strings: Vec<&str> = payload
        .get("alerts")
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().filter_map(|v| v.as_str()).collect())
        .unwrap_or_default();
    assert!(
        alert_strings.contains(&"man_down"),
        "compound alerts must include man_down; got {:?}",
        alert_strings,
    );
    assert!(
        alert_strings.contains(&"low_battery"),
        "compound alerts must include low_battery; got {:?}",
        alert_strings,
    );
}

/// **End-to-end publisher → receiver** through `PeatMesh::publish_translator_frame`.
/// This is the surface embedded firmware (M5Stack Core2) calls — no Tokio
/// runtime, no peat-mesh dep — to broadcast a translator-framed document
/// over its own transport. Validates that:
///
/// 1. The publisher's framed-and-encrypted wire bytes are
///    indistinguishable on the air from peat-mesh-using publishers.
/// 2. The receiver's standard anonymous receive path lands the
///    decoded JSON (with `man_down` alert intact) in
///    `DataReceivedResult.decoded_translator_frame`.
///
/// If this fixture goes red, M5Stack's Prone-as-unavailable broadcast
/// stops surfacing on the tablet — the full Slice 2 → Slice 3 path
/// is broken.
#[tokio::test]
async fn publish_translator_frame_round_trips_man_down_through_receiver() {
    use peat_btle::translator::BleTranslator;

    let publisher = PeatMesh::new(PeatMeshConfig::new(
        NodeId::new(0x6462_A698),
        "SCOUT-A698",
        "WEARTAK",
    ));
    let receiver = fresh_receiver(0x089F_A635, "TABLET");

    // Build the platform document the publisher hands to peat-btle.
    // Same shape M5Stack will assemble in Slice 2 — the translator's
    // `peripheral_to_platform_in_cell` is the canonical builder so
    // we use it here (otherwise tests drift from the on-device shape).
    let translator = BleTranslator::with_defaults();
    let prone = scout_peripheral(BleHealthStatus::ALERT_MAN_DOWN, None);
    let value = translator.peripheral_to_platform_in_cell(&prone, Some("WEARTAK"));
    let fields_map = value
        .as_object()
        .expect("peripheral_to_platform_in_cell returns Object")
        .clone();
    let id = fields_map
        .get("id")
        .and_then(|v| v.as_str())
        .map(String::from);
    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
    for (k, v) in fields_map {
        fields.insert(k, v);
    }
    let doc = MeshDocument {
        id,
        fields,
        updated_at: SystemTime::now(),
    };

    let wire_bytes = publisher
        .publish_translator_frame("platforms", &doc)
        .expect("publish_translator_frame returns wire bytes for the platforms collection");

    // Hand the publisher's bytes to the receiver's standard anonymous
    // receive path — the same entry point peat-atak-plugin calls from
    // Kotlin via `onBleDataReceivedAnonymous`.
    let payload = decode_received_frame(&receiver, &wire_bytes);

    let alert_strings: Vec<&str> = payload
        .get("alerts")
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().filter_map(|v| v.as_str()).collect())
        .unwrap_or_default();
    assert!(
        alert_strings.contains(&"man_down"),
        "publish_translator_frame must propagate man_down through receive; got {:?}",
        alert_strings,
    );
    assert_eq!(
        payload.get("name").and_then(|v| v.as_str()),
        Some("SCOUT-A698"),
    );
}

/// **Slim publish path used by embedded firmware.**
/// `PeatMesh::publish_platform_advertisement` takes a `BlePeripheral`
/// directly — no `peat-mesh` dep on the caller side — and produces
/// the same wire shape `publish_translator_frame` does. Locks the
/// embedded → receiver contract so a future change to the slim path's
/// framing reproduces the bench symptom (M5Stack Prone never
/// surfacing on the tablet).
#[tokio::test]
async fn publish_platform_advertisement_round_trips_man_down_through_receiver() {
    let publisher = PeatMesh::new(PeatMeshConfig::new(
        NodeId::new(0x6462_A698),
        "SCOUT-A698",
        "WEARTAK",
    ));
    let receiver = fresh_receiver(0x089F_A635, "TABLET");

    let prone = scout_peripheral(BleHealthStatus::ALERT_MAN_DOWN, None);
    let wire_bytes = publisher
        .publish_platform_advertisement(&prone)
        .expect("publish_platform_advertisement returns wire bytes for a typed BlePeripheral");

    let payload = decode_received_frame(&receiver, &wire_bytes);
    let alert_strings: Vec<&str> = payload
        .get("alerts")
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().filter_map(|v| v.as_str()).collect())
        .unwrap_or_default();
    assert!(
        alert_strings.contains(&"man_down"),
        "slim publish path must propagate man_down through receive; got {:?}",
        alert_strings,
    );
    assert_eq!(
        payload.get("name").and_then(|v| v.as_str()),
        Some("SCOUT-A698"),
    );
}

/// Slim and generic publish paths produce byte-identical wire payloads
/// for the same input — receivers can't tell them apart, which is the
/// contract that lets embedded firmware participate in the same
/// platforms collection peat-mesh-using hosts publish to.
#[tokio::test]
async fn slim_and_generic_publish_paths_emit_identical_wire_bytes() {
    use peat_btle::translator::BleTranslator;

    let publisher = PeatMesh::new(PeatMeshConfig::new(
        NodeId::new(0x6462_A698),
        "SCOUT-A698",
        "WEARTAK",
    ));
    // Encryption disabled (default config) so we compare plaintext
    // wire bytes — not nonces. With encryption on, the AEAD nonce
    // differs each call and the comparison would be meaningless.

    let prone = scout_peripheral(BleHealthStatus::ALERT_MAN_DOWN, None);

    // Slim path: typed input, no peat-mesh dep.
    let slim_bytes = publisher
        .publish_platform_advertisement(&prone)
        .expect("slim path returns bytes");

    // Generic path: build the equivalent peat-mesh Document.
    let translator = BleTranslator::with_defaults();
    let value = translator.peripheral_to_platform_in_cell(&prone, Some("WEARTAK"));
    let fields_map = value.as_object().expect("Object").clone();
    let id = fields_map
        .get("id")
        .and_then(|v| v.as_str())
        .map(String::from);
    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
    for (k, v) in fields_map {
        fields.insert(k, v);
    }
    let doc = MeshDocument {
        id,
        fields,
        updated_at: SystemTime::now(),
    };
    let generic_bytes = publisher
        .publish_translator_frame("platforms", &doc)
        .expect("generic path returns bytes");

    assert_eq!(
        slim_bytes, generic_bytes,
        "slim and generic publish paths must emit byte-identical \
         wire bytes for the same peripheral; receivers must not be \
         able to distinguish the two publishers",
    );
}

/// `publish_translator_frame` returns `None` for a collection the
/// translator doesn't carry — preserves the `Ok(None)` decline contract
/// that ADR-059 §"Backwards compatibility… 1" requires of orchestrators.
/// Without this, an embedded caller passing a typo'd collection would
/// silently ship malformed bytes that the receiver decode-fails on.
#[tokio::test]
async fn publish_translator_frame_declines_unknown_collection() {
    let publisher = PeatMesh::new(PeatMeshConfig::new(
        NodeId::new(0x6462_A698),
        "SCOUT-A698",
        "WEARTAK",
    ));
    let doc = MeshDocument {
        id: Some("ble-6462A698".to_string()),
        fields: HashMap::new(),
        updated_at: SystemTime::now(),
    };
    assert!(
        publisher
            .publish_translator_frame("not-a-real-collection", &doc)
            .is_none(),
        "publish_translator_frame must decline (return None) for unknown collections",
    );
}

/// Position-bearing peer round-trips lat/lon alongside the alerts
/// bitfield. WEAROS-7347 publishes through the same translator path
/// (post-Slice 2 if extended; today via the legacy peripheral path) —
/// keeping position present here protects the integrated case where a
/// peer with a GPS fix also asserts an alert.
#[tokio::test]
async fn position_and_alerts_coexist_on_wire() {
    let translator = BleTranslator::with_defaults();
    let receiver = fresh_receiver(0x089F_A635, "TABLET");

    let with_position = scout_peripheral(
        BleHealthStatus::ALERT_MAN_DOWN,
        Some(BlePosition {
            latitude: 33.71571,
            longitude: -84.41128,
            altitude: Some(285.5),
            accuracy: None,
        }),
    );
    let framed = frame_platforms_doc(&translator, &with_position, Some("WEARTAK")).await;
    let payload = decode_received_frame(&receiver, &framed);

    let lat = payload
        .get("lat")
        .and_then(|v| v.as_f64())
        .expect("position encoded peripheral surfaces lat in decoded JSON");
    let lon = payload
        .get("lon")
        .and_then(|v| v.as_f64())
        .expect("position encoded peripheral surfaces lon in decoded JSON");
    assert!((lat - 33.71571_f64).abs() < 1e-3, "lat mismatch: got {lat}");
    assert!(
        (lon - (-84.41128_f64)).abs() < 1e-3,
        "lon mismatch: got {lon}"
    );
    let alert_strings: Vec<&str> = payload
        .get("alerts")
        .and_then(|v| v.as_array())
        .map(|arr| arr.iter().filter_map(|v| v.as_str()).collect())
        .unwrap_or_default();
    assert!(
        alert_strings.contains(&"man_down"),
        "alerts must remain alongside a position payload; got {:?}",
        alert_strings,
    );
}