// 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.

//! ADR-059 Amendment 1 §"Backwards compatibility… 2" regression fixture.
//!
//! Locks the post-#70 silent-drop guarantee for the reserved translator-
//! marker range (0xB7..=0xBF). Without an explicit drop branch, a
//! 0xB7-prefixed payload whose bytes 8..11 form a value <= 256 would
//! pass `GCounter::decode`'s `MAX_COUNTER_ENTRIES` bound, slide through
//! `PeatDocument::decode`, and pollute the receiver's GCounter via
//! `merge_document`. The amendment requires every receive entry point
//! to reject the reserved range *before* fall-through; this fixture
//! exercises that contract by feeding crafted 0xB7..=0xBF frames into
//! all three entry points and asserting:
//!
//!   1. `merge_document` was never called (no peripheral / counter
//!      change observed via `total_count` / `peripheral_count`).
//!   2. The entry-point return value is `None`.
//!
//! Pairs with the §"Backwards compatibility… 2" concrete instance
//! (translator-encoded BlePosition at `(0,0)+None+None` would cause
//! `num_entries = 0` to slide through). Failing this fixture means a
//! future receive-path edit dropped the reserved-marker rejection;
//! restore it before merging.

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

use peat_btle::peat_mesh::{PeatMesh, PeatMeshConfig};
use peat_btle::NodeId;

/// Build a forged translator frame: marker + collection_code + 10 zero
/// bytes. The 10-byte payload lets `data[8..11]` land on zeros, which
/// is exactly the GCounter-pollution input the reserved-marker check
/// exists to neutralize.
fn forged_frame(marker: u8) -> Vec<u8> {
    let mut buf = Vec::with_capacity(12);
    buf.push(marker);
    buf.push(0x01); // collection_code (irrelevant for reserved markers)
    buf.extend(std::iter::repeat(0u8).take(10));
    buf
}

/// Fresh mesh with a fixed node_id so total_count() reads cleanly.
fn make_mesh(node_id: u32) -> PeatMesh {
    let cfg = PeatMeshConfig::new(NodeId::new(node_id), "TEST-1", "test-mesh");
    PeatMesh::new(cfg)
}

#[test]
fn anonymous_receive_drops_all_reserved_markers() {
    for marker in 0xB7u8..=0xBF {
        let mesh = make_mesh(0xCAFE_0001);
        let count_before = mesh.total_count();

        let result = mesh.on_ble_data_received_anonymous(
            "peer-test",
            &forged_frame(marker),
            1_700_000_000_000,
        );

        assert!(
            result.is_none(),
            "anonymous receive must return None for reserved marker 0x{marker:02X}, got Some(..)"
        );
        assert_eq!(
            mesh.total_count(),
            count_before,
            "reserved-marker drop must not advance the GCounter (marker=0x{marker:02X})"
        );
    }
}

#[test]
fn from_node_receive_drops_all_reserved_markers() {
    for marker in 0xB7u8..=0xBF {
        let mesh = make_mesh(0xCAFE_0002);
        let count_before = mesh.total_count();

        let result = mesh.on_ble_data_received_from_node(
            NodeId::new(0xCAFE_0099),
            &forged_frame(marker),
            1_700_000_000_000,
        );

        assert!(
            result.is_none(),
            "from-node receive must return None for reserved marker 0x{marker:02X}, got Some(..)"
        );
        assert_eq!(
            mesh.total_count(),
            count_before,
            "reserved-marker drop must not advance the GCounter (marker=0x{marker:02X})"
        );
    }
}

// Coverage note: `on_ble_data_received(identifier, ...)` (the
// known-peer entry point) calls `process_document_data_with_identifier`,
// while `on_ble_data_received_from_node` calls `process_document_data`.
// Both helpers invoke the **same** `try_handle_translator_marker` helper
// before the legacy/delta dispatch branches; testing one exercises the
// shared marker-rejection logic for the other. A direct test of
// `on_ble_data_received` requires a registered peer-identifier mapping
// for which there is no public API today; the higher-level integration
// path is covered once peat-ffi (Slice 1.b.4) wires the
// `DecodedDocumentCallback` and end-to-end fixtures land.

/// Forged 0xB6 translator frame with a postcard-mismatched payload also
/// silent-drops (decode_inbound returns Err for malformed bytes; the
/// receive dispatch logs and drops without falling through). Locks the
/// 0xB6-but-malformed branch as a sibling of the reserved-marker
/// rejection — same return contract, different metric path.
#[test]
fn anonymous_receive_drops_malformed_translator_frame() {
    let mesh = make_mesh(0xCAFE_0004);
    let count_before = mesh.total_count();

    // 0xB6 marker + tracks code + bytes that postcard can't parse as
    // BlePosition. The receive dispatch reaches decode_inbound_sync;
    // postcard's decoder returns Err; the helper logs+drops; result is
    // None and the GCounter is untouched.
    let mut frame = vec![0xB6, 0x01];
    frame.extend(b"this is not a postcard-encoded BlePosition");

    let result = mesh.on_ble_data_received_anonymous("peer-malformed", &frame, 1_700_000_000_000);

    assert!(
        result.is_none(),
        "malformed translator frame must return None"
    );
    assert_eq!(
        mesh.total_count(),
        count_before,
        "malformed-frame drop must not advance the GCounter"
    );
}

/// ADR-059 Amendment 1 §"Decoded-document callback" — release-skew
/// window observability test. When a 0xB6 frame decodes successfully
/// but no `DecodedDocumentCallback` is installed (the deliberate
/// 1.b.3-ships-before-1.b.4 window), the receive dispatch MUST emit a
/// `PeatEvent::TranslatorNoCallback { collection, peer }` event so
/// operators staging the rollout see the gap in real time. Without
/// this event the no-callback path would be silent (`log::debug!` is
/// disabled at default log levels), and the staged-rollout safety
/// story the slice depends on falls apart.
///
/// QA on PR #31 round 1 specifically called this out as a [WARNING];
/// this fixture locks the `PeatEvent` emission as a regression so a
/// future edit can't accidentally drop it.
#[test]
fn no_callback_path_emits_translator_no_callback_event() {
    use peat_btle::observer::{CollectingObserver, PeatEvent};
    use peat_btle::translator::{BlePeripheral, BlePeripheralType, BleTranslator};
    use std::sync::Arc;

    let mesh = make_mesh(0xCAFE_0005);
    let observer = Arc::new(CollectingObserver::new());
    mesh.add_observer(observer.clone());

    // Build a valid 0xB6 platforms frame by encoding a BlePeripheral
    // through the translator the receive path will use to decode it.
    // Platforms collection chosen because BlePosition (tracks) requires
    // ctx.local_wire_id which the anonymous-receive path can't provide.
    let translator = BleTranslator::with_defaults();
    let peripheral = BlePeripheral {
        id: 0xCAFE_AAAA,
        parent_node: 0,
        peripheral_type: BlePeripheralType::SoldierSensor,
        callsign: "OBS-1".into(),
        health: peat_btle::translator::BleHealthStatus {
            battery_percent: 75,
            heart_rate: None,
            activity: 0,
            alerts: 0,
        },
        timestamp: 1_700_000_000_000,
        position: None,
    };
    let payload = postcard::to_allocvec(&peripheral).expect("encode BlePeripheral");
    let mut frame = Vec::with_capacity(2 + payload.len());
    frame.push(peat_btle::document::TRANSLATOR_FRAME_MARKER);
    frame.push(peat_btle::translator::COLLECTION_CODE_PLATFORMS);
    frame.extend_from_slice(&payload);

    // Sanity: the translator decodes the same bytes back, so the
    // receive dispatch will reach the no-callback branch (not Err,
    // not None decline).
    {
        use peat_mesh::transport::TranslationContext;
        let ctx = TranslationContext::inbound("peer-test")
            .with_collection(translator.platforms_collection());
        let decoded = translator
            .decode_inbound_sync(&payload, &ctx)
            .expect("decode_inbound_sync result is Ok")
            .expect("Ok(Some) decode for platforms");
        assert!(
            decoded.fields.contains_key("name") || decoded.fields.contains_key("id"),
            "decoded platforms doc has expected fields: {:?}",
            decoded.fields.keys().collect::<Vec<_>>()
        );
    }

    // Receive the frame anonymously — no DecodedDocumentCallback was
    // installed, so the no-callback branch fires.
    let _ = mesh.on_ble_data_received_anonymous("peer-observer-test", &frame, 1_700_000_000_000);

    // Assert the operator-observable event fired with the correct
    // collection name and peer identifier.
    let events = observer.events();
    let found = events.iter().any(|e| {
        matches!(
            e,
            PeatEvent::TranslatorNoCallback { collection, peer }
            if collection == translator.platforms_collection()
                && peer.as_deref() == Some("peer-observer-test")
        )
    });
    assert!(
        found,
        "expected PeatEvent::TranslatorNoCallback {{ collection=\"platforms\", peer=Some(\"peer-observer-test\") }} \
         in observer events; got {events:?}"
    );
}

/// ADR-059 Amendment 2 — UniFFI callback fires on 0xB6 receive, and the
/// `PeatEvent::TranslatorNoCallback` event is **suppressed** when at
/// least one callback (Rust-trait or UniFFI-JSON) is installed.
///
/// Locks both the JSON-string serialization contract (Document → JSON
/// before crossing the FFI boundary) and the both-callbacks-fire-
/// independently guarantee the amendment specifies. A future edit that
/// makes the JSON callback suppress the Rust callback (or vice versa)
/// fails this test.
#[cfg(feature = "uniffi")]
#[test]
fn json_callback_fires_and_suppresses_no_callback_event() {
    use peat_btle::observer::{CollectingObserver, PeatEvent};
    use peat_btle::translator::{BlePeripheral, BlePeripheralType, BleTranslator};
    use peat_btle::DecodedDocumentJsonCallback;
    use std::sync::{Arc, Mutex};

    /// Records each `on_document` call so the test can assert against
    /// the captured `(collection, doc_json, peer)` tuples.
    struct RecordingJsonCallback {
        calls: Mutex<Vec<(String, String, Option<String>)>>,
    }

    impl DecodedDocumentJsonCallback for RecordingJsonCallback {
        fn on_document(&self, collection: String, doc_json: String, peer: Option<String>) {
            self.calls
                .lock()
                .unwrap()
                .push((collection, doc_json, peer));
        }
    }

    let mesh = make_mesh(0xCAFE_0006);
    let observer = Arc::new(CollectingObserver::new());
    mesh.add_observer(observer.clone());

    let recorder = Arc::new(RecordingJsonCallback {
        calls: Mutex::new(Vec::new()),
    });
    // The setter takes Box; UniFFI's callback_interface generates code
    // that builds that Box from a foreign Kotlin/Swift impl. In Rust
    // tests we need a Box that holds the same Arc the test inspects —
    // a small adapter routes Box::on_document calls back to the Arc'd
    // recorder so the test still sees the captured events.
    struct AdapterCallback(Arc<RecordingJsonCallback>);
    impl DecodedDocumentJsonCallback for AdapterCallback {
        fn on_document(&self, collection: String, doc_json: String, peer: Option<String>) {
            self.0.on_document(collection, doc_json, peer);
        }
    }
    mesh.set_decoded_document_json_callback(Box::new(AdapterCallback(recorder.clone())));

    // Build a valid 0xB6 platforms frame the same way the no-callback
    // sibling test does.
    let translator = BleTranslator::with_defaults();
    let peripheral = BlePeripheral {
        id: 0xCAFE_BBBB,
        parent_node: 0,
        peripheral_type: BlePeripheralType::SoldierSensor,
        callsign: "JSON-1".into(),
        health: peat_btle::translator::BleHealthStatus {
            battery_percent: 92,
            heart_rate: None,
            activity: 0,
            alerts: 0,
        },
        timestamp: 1_700_000_000_000,
        position: None,
    };
    let payload = postcard::to_allocvec(&peripheral).expect("encode BlePeripheral");
    let mut frame = Vec::with_capacity(2 + payload.len());
    frame.push(peat_btle::document::TRANSLATOR_FRAME_MARKER);
    frame.push(peat_btle::translator::COLLECTION_CODE_PLATFORMS);
    frame.extend_from_slice(&payload);

    let _ = mesh.on_ble_data_received_anonymous("peer-json-test", &frame, 1_700_000_000_000);

    // (a) The JSON callback fired exactly once with the platforms collection.
    let calls = recorder.calls.lock().unwrap().clone();
    assert_eq!(
        calls.len(),
        1,
        "expected one JSON callback invocation, got {calls:?}"
    );
    assert_eq!(
        calls[0].0,
        translator.platforms_collection(),
        "JSON callback collection name"
    );
    assert_eq!(
        calls[0].2.as_deref(),
        Some("peer-json-test"),
        "JSON callback peer identifier"
    );

    // (b) The doc_json string is well-formed JSON that round-trips
    // through serde_json — the host can parse it without further
    // negotiation. Avoid asserting exact field shape (translator
    // internals may evolve); just that it parses.
    let parsed: serde_json::Value =
        serde_json::from_str(&calls[0].1).expect("doc_json is well-formed JSON");
    assert!(
        parsed.is_object(),
        "doc_json should serialize Document as JSON object; got {parsed:?}"
    );

    // (c) `PeatEvent::TranslatorNoCallback` did NOT fire — the JSON
    // callback being installed counts as a callback for suppression
    // purposes.
    let events = observer.events();
    let no_callback_event_fired = events
        .iter()
        .any(|e| matches!(e, PeatEvent::TranslatorNoCallback { .. }));
    assert!(
        !no_callback_event_fired,
        "TranslatorNoCallback must NOT fire when a JSON callback is installed; got {events:?}"
    );
}

/// ADR-059 Amendment 2 — when **both** the Rust-trait
/// [`DecodedDocumentCallback`] and the UniFFI-JSON
/// [`DecodedDocumentJsonCallback`] are installed on the same mesh,
/// a single 0xB6 receive must invoke **each** callback exactly once.
/// Neither suppresses the other.
///
/// QA on PR #36 round 1 specifically called this out as a [WARNING]:
/// the previous test only installed the JSON callback, so a future
/// refactor that collapsed the dual-fire branch to either-or
/// (`if let Some(json) ... else if let Some(rust) ...`) would still
/// pass every other test in the file. This fixture pins the dual-fire
/// branch directly: both recorders see exactly one invocation, and
/// `PeatEvent::TranslatorNoCallback` is suppressed.
#[cfg(feature = "uniffi")]
#[test]
fn both_callbacks_fire_independently_on_one_frame() {
    use peat_btle::observer::{CollectingObserver, PeatEvent};
    use peat_btle::translator::{BlePeripheral, BlePeripheralType, BleTranslator};
    use peat_btle::{DecodedDocumentCallback, DecodedDocumentJsonCallback, MeshDocument};
    use std::sync::{Arc, Mutex};

    /// Records each Rust-trait callback fire (collection, peer).
    struct RecordingRustCallback {
        calls: Mutex<Vec<(String, Option<String>)>>,
    }
    impl DecodedDocumentCallback for RecordingRustCallback {
        fn on_document(&self, collection: &str, _doc: MeshDocument, peer: Option<&str>) {
            self.calls
                .lock()
                .unwrap()
                .push((collection.to_string(), peer.map(str::to_string)));
        }
    }

    /// Records each JSON callback fire.
    struct RecordingJsonCallback {
        calls: Mutex<Vec<(String, Option<String>)>>,
    }
    impl DecodedDocumentJsonCallback for RecordingJsonCallback {
        fn on_document(&self, collection: String, _doc_json: String, peer: Option<String>) {
            self.calls.lock().unwrap().push((collection, peer));
        }
    }

    let mesh = make_mesh(0xCAFE_0007);
    let observer = Arc::new(CollectingObserver::new());
    mesh.add_observer(observer.clone());

    // Install BOTH callbacks on the same mesh.
    let rust_recorder = Arc::new(RecordingRustCallback {
        calls: Mutex::new(Vec::new()),
    });
    mesh.set_decoded_document_callback(rust_recorder.clone());

    let json_recorder = Arc::new(RecordingJsonCallback {
        calls: Mutex::new(Vec::new()),
    });
    // Adapter routes the Box-required UniFFI signature back to the
    // Arc the test inspects (same shape as the JSON-only test above).
    struct AdapterCallback(Arc<RecordingJsonCallback>);
    impl DecodedDocumentJsonCallback for AdapterCallback {
        fn on_document(&self, collection: String, doc_json: String, peer: Option<String>) {
            self.0.on_document(collection, doc_json, peer);
        }
    }
    mesh.set_decoded_document_json_callback(Box::new(AdapterCallback(json_recorder.clone())));

    // Send one 0xB6 platforms frame.
    let translator = BleTranslator::with_defaults();
    let peripheral = BlePeripheral {
        id: 0xCAFE_CCCC,
        parent_node: 0,
        peripheral_type: BlePeripheralType::SoldierSensor,
        callsign: "DUAL-1".into(),
        health: peat_btle::translator::BleHealthStatus {
            battery_percent: 50,
            heart_rate: None,
            activity: 0,
            alerts: 0,
        },
        timestamp: 1_700_000_000_000,
        position: None,
    };
    let payload = postcard::to_allocvec(&peripheral).expect("encode BlePeripheral");
    let mut frame = Vec::with_capacity(2 + payload.len());
    frame.push(peat_btle::document::TRANSLATOR_FRAME_MARKER);
    frame.push(peat_btle::translator::COLLECTION_CODE_PLATFORMS);
    frame.extend_from_slice(&payload);

    let _ = mesh.on_ble_data_received_anonymous("peer-dual-test", &frame, 1_700_000_000_000);

    // (a) Rust-trait callback fired exactly once with the right
    // collection + peer.
    let rust_calls = rust_recorder.calls.lock().unwrap().clone();
    assert_eq!(
        rust_calls.len(),
        1,
        "Rust-trait callback must fire exactly once when both callbacks are installed; \
         got {rust_calls:?}"
    );
    assert_eq!(rust_calls[0].0, translator.platforms_collection());
    assert_eq!(rust_calls[0].1.as_deref(), Some("peer-dual-test"));

    // (b) JSON callback also fired exactly once with the same
    // collection + peer. Each callback is independent — neither
    // suppresses the other.
    let json_calls = json_recorder.calls.lock().unwrap().clone();
    assert_eq!(
        json_calls.len(),
        1,
        "JSON callback must fire exactly once when both callbacks are installed; \
         got {json_calls:?}"
    );
    assert_eq!(json_calls[0].0, translator.platforms_collection());
    assert_eq!(json_calls[0].1.as_deref(), Some("peer-dual-test"));

    // (c) `PeatEvent::TranslatorNoCallback` did NOT fire — having
    // either callback installed (let alone both) suppresses it.
    let events = observer.events();
    let no_callback_event_fired = events
        .iter()
        .any(|e| matches!(e, PeatEvent::TranslatorNoCallback { .. }));
    assert!(
        !no_callback_event_fired,
        "TranslatorNoCallback must NOT fire when any callback is installed; got {events:?}"
    );
}
