// Copyright (c) 2025-2026 Defense Unicorns, Inc.
// SPDX-License-Identifier: Apache-2.0

//! Universal-Document [`Translator`] impl over the peat-lite envelope.
//!
//! Bridges peat-mesh's [`Document`] type onto the
//! [`peat_lite::MessageType::Document`] envelope (carried by peat-btle today,
//! peat-lora tomorrow) so the existing
//! [`TransportManager`](crate::transport::manager::TransportManager) fan-out
//! observes doc-store changes and routes them through the universal
//! transport without the application orchestrating each emit by hand.
//!
//! Layering rule (ADR-059 Amendment 4): the wire codec
//! ([`crate::transport::document_codec`]) lives in peat-mesh, not in any one
//! physical-transport crate. This module is the trait-level glue between
//! that codec and the orchestrator. **It does not own the radio** — the
//! consumer (peat-ffi today) wires an
//! [`OutboundSink`](crate::transport::OutboundSink) that hands the encoded
//! bytes to the appropriate physical transport
//! (`peat_btle::PeatMesh::publish_peat_lite_document` for BLE).
//!
//! ## transport_id naming
//!
//! Per the [`Translator::transport_id`] contract, the ID must be a
//! compile-time `&'static str` literal — the orchestrator uses it for
//! origin-skip (echo-loop suppression on inbound) and for
//! `allowed_transports` (per-doc opt-in). Same codec across physical
//! transports, but **distinct IDs per physical transport** so origin-skip
//! can tell BLE-originated bytes from LoRa-originated bytes:
//!
//! | Physical transport | Suggested transport_id | Constant            |
//! |--------------------|------------------------|---------------------|
//! | peat-btle (BLE)    | `"ble-lite"`           | [`BLE_LITE_BRIDGE`] |
//! | future peat-lora   | `"lora-lite"`          | (define when added) |
//!
//! The struct stores its ID rather than hard-coding, so a future LoRa
//! consumer constructs `LiteBridgeTranslator::new(LORA_LITE_BRIDGE)`
//! against the same codec without forking the impl.
//!
//! ## Outbound flow
//!
//! 1. App: `node.publish_with_origin("markers", doc, Some("self"))`.
//! 2. Doc-store fires [`ChangeEvent::Updated`] with `origin = Some("self")`.
//! 3. [`TransportManager`] fan-out checks each registered translator's
//!    `transport_id`; non-matching origins receive a `FanoutItem`.
//! 4. The per-transport drain task calls
//!    [`Translator::encode_outbound`](crate::transport::Translator::encode_outbound)
//!    which routes here.
//! 5. We call [`document_codec::encode_document`] with `flags = 0` and
//!    return the envelope bytes (no header / no encryption — the consumer's
//!    `OutboundSink` adds those when it hands bytes to peat-btle's
//!    `publish_peat_lite_document`).
//!
//! ## Inbound flow
//!
//! Symmetric to outbound: peat-btle's GATT receive surfaces a
//! `DataReceivedResult.peat_lite_document` whose fields the consumer
//! can either feed straight into `Node::publish_with_origin` (the
//! peat-atak-plugin path, since the Kotlin side already pulls the
//! envelope apart field-by-field) or hand the raw envelope payload to
//! [`Translator::decode_inbound`](crate::transport::Translator::decode_inbound) —
//! equivalent code paths because the trait method just calls
//! [`document_codec::decode_document`] internally.
//!
//! ### Slice-1.a trait-shape limitation
//!
//! `Translator::decode_inbound` returns `Document`, but the codec
//! produces `(collection, Document)` — the trait shape *drops the
//! authoritative collection name on the inbound side*. Today's
//! consumers (peat-atak-plugin / peat-ffi) sidestep this by reading
//! `peat_lite_document.collection` directly off the receive struct
//! before feeding the body to ingest, so the lossy trait method isn't
//! on the hot path. If a future code path drives doc-store ingest
//! through `Translator::decode_inbound` directly, the trait will need
//! a richer return shape (e.g. `decode_inbound_with_collection`
//! returning `(String, Document)`); tracked as part of ADR-059
//! Slice 2's `Translator::encode_delete` work where the trait surface
//! is already revisited.
//!
//! ## Tombstones / deletions
//!
//! Today the fan-out orchestrator silently skips
//! [`ChangeEvent::Removed`](crate::sync::types::ChangeEvent::Removed) (Slice
//! 1.a scope cut, see `transport::fanout::FanoutState::fanout_event`); Slice
//! 2 will add a `Translator::encode_delete` hook. Until then,
//! tombstone-flag emission on this codec is unreachable from the fan-out
//! path. Soft-deletes via the `_deleted: true` field convention work
//! today — they ride as ordinary `Updated` events and the body bytes carry
//! the sentinel through to the receiver, which the Slice 1.a
//! `document_codec::decode_document` already routes onto the
//! `_deleted: true` field for [ADR-034 / Issue
//! #369](https://github.com/defenseunicorns/peat/issues/369) compatibility.
//!
//! [`Document`]: crate::sync::types::Document
//! [`Translator`]: crate::transport::Translator
//! [`Translator::transport_id`]: crate::transport::Translator::transport_id
//! [`document_codec::encode_document`]: crate::transport::document_codec::encode_document
//! [`document_codec::decode_document`]: crate::transport::document_codec::decode_document
//! [`ChangeEvent::Updated`]: crate::sync::types::ChangeEvent::Updated
//! [`TransportManager`]: crate::transport::manager::TransportManager

use async_trait::async_trait;

use crate::sync::types::Document;
use crate::transport::document_codec;
use crate::transport::translator::{TranslationContext, Translator};

/// `transport_id` for the BLE physical transport carrying the
/// universal-Document envelope. Pass to
/// [`LiteBridgeTranslator::new`] from peat-ffi when registering against
/// peat-btle.
pub const BLE_LITE_BRIDGE: &str = "ble-lite";

/// Universal-Document Translator. Encodes and decodes peat-mesh
/// `Document`s into / out of peat-lite envelope bytes via
/// [`crate::transport::document_codec`]. Physical-transport-agnostic;
/// the consumer's [`OutboundSink`](crate::transport::OutboundSink)
/// hands the bytes to whichever radio carries the envelope (peat-btle
/// today, peat-lora later).
///
/// Construct with [`LiteBridgeTranslator::new`], passing the
/// `&'static str` `transport_id` for the physical transport this
/// instance fronts. See [`BLE_LITE_BRIDGE`].
#[derive(Debug, Clone)]
pub struct LiteBridgeTranslator {
    transport_id: &'static str,
}

impl LiteBridgeTranslator {
    /// Build a translator whose [`Translator::transport_id`] returns
    /// the given `transport_id`. The argument is `&'static str` so the
    /// orchestrator's stable-string requirement is enforced at the
    /// type level — pass a literal or a `pub const`, never a runtime
    /// `&str`.
    ///
    /// [`Translator::transport_id`]: crate::transport::Translator::transport_id
    pub fn new(transport_id: &'static str) -> Self {
        Self { transport_id }
    }

    /// Convenience constructor for the BLE physical transport. Equivalent
    /// to `LiteBridgeTranslator::new(BLE_LITE_BRIDGE)`. Keeps consumers
    /// (peat-ffi) from having to import the const explicitly when they
    /// only register the BLE-side instance.
    pub fn for_ble() -> Self {
        Self::new(BLE_LITE_BRIDGE)
    }
}

#[async_trait]
impl Translator for LiteBridgeTranslator {
    fn transport_id(&self) -> &'static str {
        self.transport_id
    }

    async fn encode_outbound(&self, doc: &Document, ctx: &TranslationContext) -> Option<Vec<u8>> {
        // Per ADR-059 §"Translator trait": `None` is a *decline*, not an
        // error. Decline the encode if the orchestrator didn't stamp a
        // collection on the context — we have nowhere to put it on the
        // wire (the envelope's `collection` field is mandatory). This
        // is a routing-shape issue, not a real codec failure, so it's
        // a silent decline matching every other Translator impl.
        let collection = ctx.collection.as_deref()?;

        // Slice 1.a: tombstone bit unused (Removed events skipped at
        // fan-out). Slice 2's `Translator::encode_delete` hook will
        // pass the flag through; for now flags=0.
        match document_codec::encode_document(0, collection, doc) {
            Ok(bytes) => Some(bytes),
            Err(e) => {
                // Real encode error (over-size body, JSON serialise
                // failure on Document.fields). Per ADR-059 §"Translator
                // trait" telemetry contract: distinguish silent
                // declines from real codec errors via tracing::warn so
                // sender-side wire-format drift is visible to
                // operators. Same pattern as `BleTranslator`'s
                // `warn_on_none`.
                tracing::warn!(
                    transport = %self.transport_id,
                    collection = %collection,
                    error = ?e,
                    "lite-bridge: encode_outbound error — wire-format drift candidate"
                );
                None
            }
        }
    }

    async fn decode_inbound(
        &self,
        bytes: &[u8],
        ctx: &TranslationContext,
    ) -> anyhow::Result<Option<Document>> {
        // The envelope itself carries the collection name (unlike
        // BleTranslator's typed-frame path which depends on a 1-byte
        // collection code requiring a peer-side dispatch table). The
        // orchestrator's `ctx.collection` may not be set on inbound
        // calls for this codec, so we don't read it. `ctx.peer`
        // however is stamped by the inbound bridge for correlation —
        // include it on the error path so operators tracking
        // wire-format drift can attribute decode failures to a
        // specific peer. Symmetric to the outbound `tracing::warn!`.
        let (_collection_from_envelope, doc) =
            document_codec::decode_document(bytes).map_err(|e| {
                tracing::warn!(
                    transport = %self.transport_id,
                    peer = ?ctx.peer,
                    error = ?e,
                    "lite-bridge: decode_inbound error — wire-format drift candidate"
                );
                anyhow::anyhow!(
                    "lite-bridge: decode_document failed (peer={:?}): {e:?}",
                    ctx.peer
                )
            })?;
        Ok(Some(doc))
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::time::SystemTime;

    use serde_json::{json, Value};

    use super::*;

    fn doc_with_fields(id: Option<&str>, fields: HashMap<String, Value>) -> Document {
        Document {
            id: id.map(|s| s.to_string()),
            fields,
            updated_at: SystemTime::UNIX_EPOCH
                + std::time::Duration::from_millis(1_700_000_000_000),
        }
    }

    #[test]
    fn transport_id_is_stable_static_str() {
        // The orchestrator's stable-string contract is enforced at the
        // type level: storing a `&'static str` and returning it through
        // the trait's `&'static str` accessor compiles only because
        // both are `'static`-lifetimed. If a refactor ever drops the
        // 'static bound on the field, this assignment fails to compile
        // — the test guards the signature.
        let t = LiteBridgeTranslator::new(BLE_LITE_BRIDGE);
        let id: &'static str = t.transport_id();
        assert_eq!(id, "ble-lite");
        assert_eq!(BLE_LITE_BRIDGE, "ble-lite");
    }

    #[test]
    fn for_ble_constructor_sets_ble_lite_id() {
        let t = LiteBridgeTranslator::for_ble();
        assert_eq!(t.transport_id(), BLE_LITE_BRIDGE);
    }

    #[test]
    fn future_lora_consumer_can_register_with_distinct_transport_id() {
        // Locks the design intent: the same Translator impl serves
        // any peat-lite-envelope-carrying physical transport, just
        // with a different `transport_id`. If a future refactor moves
        // the id to a hard-coded const inside the impl, this fails.
        const LORA_LITE_BRIDGE: &str = "lora-lite";
        let lora = LiteBridgeTranslator::new(LORA_LITE_BRIDGE);
        let ble = LiteBridgeTranslator::for_ble();
        assert_ne!(
            lora.transport_id(),
            ble.transport_id(),
            "BLE and LoRa instances MUST register with distinct transport_ids \
             so origin-skip can disambiguate echo-loops between them"
        );
    }

    // === encode_outbound ===

    #[tokio::test]
    async fn encode_outbound_declines_when_collection_absent() {
        // Per the trait contract, missing collection → silent decline,
        // not an error. The orchestrator always stamps `ctx.collection`
        // from the originating ChangeEvent, but defensive callers
        // (tests, future direct invocations) may not — we shouldn't
        // panic or warn.
        let t = LiteBridgeTranslator::for_ble();
        let doc = doc_with_fields(Some("m1"), HashMap::new());
        let ctx = TranslationContext::outbound();
        assert!(
            t.encode_outbound(&doc, &ctx).await.is_none(),
            "missing ctx.collection MUST decline silently"
        );
    }

    #[tokio::test]
    async fn encode_outbound_produces_well_formed_envelope() {
        let t = LiteBridgeTranslator::for_ble();
        let mut fields = HashMap::new();
        fields.insert("lat".to_string(), json!(33.71576));
        fields.insert("lon".to_string(), json!(-84.41152));
        fields.insert("type".to_string(), json!("a-f-G-U-C"));
        let doc = doc_with_fields(Some("marker-uuid-001"), fields.clone());

        let ctx = TranslationContext::outbound().with_collection("markers");
        let bytes = t.encode_outbound(&doc, &ctx).await.expect("encode");

        // Round-trip via the codec to verify the bytes are a valid
        // envelope. We don't assert exact byte layout (that's
        // document_codec's job); we assert "the envelope decodes back
        // to the same Document".
        let (collection, decoded) = document_codec::decode_document(&bytes).expect("decode");
        assert_eq!(collection, "markers");
        assert_eq!(decoded.id, doc.id);
        assert_eq!(decoded.fields, fields);
    }

    #[tokio::test]
    async fn encode_outbound_round_trips_through_decode_inbound() {
        // End-to-end: encode_outbound bytes feed straight into
        // decode_inbound and produce the original Document. Locks the
        // codec symmetry the orchestrator-driven flow depends on
        // (the receive side of the translator is still exercised
        // today by other consumers — peat-ffi will register the same
        // Translator on both sides).
        let t = LiteBridgeTranslator::for_ble();
        let mut fields = HashMap::new();
        fields.insert(
            "complex".to_string(),
            json!({
                "tags": ["alpha", "bravo"],
                "nested": { "deep": [1, 2, 3] }
            }),
        );
        let doc = doc_with_fields(Some("complex-1"), fields.clone());

        let ctx = TranslationContext::outbound().with_collection("custom");
        let bytes = t.encode_outbound(&doc, &ctx).await.expect("encode");

        // decode_inbound is collection-agnostic; envelope carries it.
        let inbound_ctx = TranslationContext::inbound("test-peer");
        let decoded = t
            .decode_inbound(&bytes, &inbound_ctx)
            .await
            .expect("decode")
            .expect("non-decline");
        assert_eq!(decoded.id, doc.id);
        assert_eq!(decoded.fields, fields);
    }

    #[tokio::test]
    async fn encode_outbound_round_trips_doc_with_no_id() {
        // Document.id = None → wire doc_id = "" (empty) per
        // document_codec convention. Round-trip must preserve the
        // None-ness; otherwise the doc_store's "publisher-delegated
        // id" path collapses on cross-transport.
        let t = LiteBridgeTranslator::for_ble();
        let mut fields = HashMap::new();
        fields.insert("k".to_string(), json!("v"));
        let doc = doc_with_fields(None, fields.clone());

        let ctx = TranslationContext::outbound().with_collection("c");
        let bytes = t.encode_outbound(&doc, &ctx).await.expect("encode");

        let inbound_ctx = TranslationContext::inbound("p");
        let decoded = t
            .decode_inbound(&bytes, &inbound_ctx)
            .await
            .expect("decode")
            .expect("non-decline");
        assert!(
            decoded.id.is_none(),
            "id=None on input MUST round-trip to id=None on output"
        );
        assert_eq!(decoded.fields, fields);
    }

    #[tokio::test]
    async fn encode_outbound_carries_soft_delete_sentinel_through_body() {
        // peat-mesh's soft-delete convention (ADR-034 / Issue #369) is
        // a `_deleted: true` field on the Document. Until Slice 2 lands
        // wire-bit tombstone propagation through the fan-out, the
        // sentinel is the deletion signal that survives the bridge —
        // it rides in the envelope body bytes as ordinary JSON. This
        // test pins that path so a future encoder change doesn't
        // strip the sentinel and silently break deletes.
        let t = LiteBridgeTranslator::for_ble();
        let mut fields = HashMap::new();
        fields.insert("_deleted".to_string(), Value::Bool(true));
        let doc = doc_with_fields(Some("to-delete"), fields);

        let ctx = TranslationContext::outbound().with_collection("markers");
        let bytes = t.encode_outbound(&doc, &ctx).await.expect("encode");

        let inbound_ctx = TranslationContext::inbound("p");
        let decoded = t
            .decode_inbound(&bytes, &inbound_ctx)
            .await
            .expect("decode")
            .expect("non-decline");
        assert_eq!(
            decoded.fields.get("_deleted"),
            Some(&Value::Bool(true)),
            "soft-delete sentinel MUST survive the bridge until Slice 2 \
             wires up wire-bit tombstone propagation"
        );
    }

    // === decode_inbound ===

    #[tokio::test]
    async fn decode_inbound_surfaces_envelope_collection_through_doc_round_trip() {
        // The Translator trait returns Document, not (collection,
        // Document). The orchestrator's inbound bridge separately
        // tracks the collection (typically via ctx.collection or
        // by calling decode_inbound per-collection-bucket). For this
        // codec the collection lives in the envelope itself; we
        // verify the field round-trips.
        let t = LiteBridgeTranslator::for_ble();
        let mut fields = HashMap::new();
        fields.insert("x".to_string(), json!(1));
        let doc = doc_with_fields(Some("d1"), fields);

        let ctx = TranslationContext::outbound().with_collection("my-collection");
        let bytes = t.encode_outbound(&doc, &ctx).await.expect("encode");

        // The collection is on the wire; future direct callers can
        // also call document_codec::decode_document for the
        // (collection, Document) tuple.
        let (envelope_collection, _) =
            document_codec::decode_document(&bytes).expect("envelope decode");
        assert_eq!(envelope_collection, "my-collection");
    }

    #[tokio::test]
    async fn decode_inbound_propagates_codec_error_as_anyhow_err() {
        let t = LiteBridgeTranslator::for_ble();
        // Malformed envelope: not even peat-lite MAGIC.
        let bytes = b"not a peat-lite frame";
        let ctx = TranslationContext::inbound("peer");
        let result = t.decode_inbound(bytes, &ctx).await;
        assert!(
            result.is_err(),
            "malformed bytes MUST surface as Err so the orchestrator's \
             parse-error metric can fire"
        );
    }

    // === Fan-out integration ===
    //
    // The unit tests above cover the trait surface in isolation. This
    // section exercises the full registered-with-TransportManager
    // path the production wiring will take: publish a doc, observe
    // the OutboundSink receive the encoded envelope bytes, verify
    // round-trip + origin-skip.

    use std::sync::{Arc, Mutex as StdMutex};

    use async_trait::async_trait;
    use tokio::time::{timeout, Duration};

    use crate::sync::in_memory::InMemoryBackend;
    use crate::sync::traits::DataSyncBackend;
    use crate::transport::manager::{TransportManager, TransportManagerConfig};
    use crate::transport::{OutboundSink, TranslatorRegistrationConfig};

    /// Capturing sink: records every (bytes, ctx) tuple it sees so
    /// tests can assert on the wire output. Drop-in replacement for
    /// `DiscardSink` in fanout's own test module — duplicated here
    /// because cross-module test-helper exports aren't worth the
    /// cfg-gating churn.
    struct CapturingSink {
        captured: Arc<StdMutex<Vec<Vec<u8>>>>,
    }

    #[async_trait]
    impl OutboundSink for CapturingSink {
        async fn send_outbound(
            &self,
            bytes: Vec<u8>,
            _ctx: &TranslationContext,
        ) -> anyhow::Result<()> {
            self.captured.lock().unwrap().push(bytes);
            Ok(())
        }
    }

    fn make_test_doc(id: &str, body_field: &str, body_value: &str) -> Document {
        let mut fields = HashMap::new();
        fields.insert(body_field.to_string(), json!(body_value));
        Document {
            id: Some(id.to_string()),
            fields,
            updated_at: SystemTime::UNIX_EPOCH
                + std::time::Duration::from_millis(1_700_000_000_000),
        }
    }

    /// End-to-end: publishing a Document via Node fan-out drives
    /// LiteBridgeTranslator::encode_outbound, which emits well-formed
    /// peat-lite envelope bytes to the registered OutboundSink. This
    /// is the load-bearing test for ADR-059 Slice 1.b's "scope #3"
    /// architecture — peat-mesh observes doc-store changes and routes
    /// through the universal Document transport without per-collection
    /// orchestration in the application.
    #[tokio::test]
    async fn fanout_publish_drives_lite_bridge_encode_to_outbound_sink() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Arc::new(crate::node::Node::new(Arc::clone(&backend)));
        let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));

        let captured = Arc::new(StdMutex::new(Vec::<Vec<u8>>::new()));
        let sink = Arc::new(CapturingSink {
            captured: Arc::clone(&captured),
        });
        let translator = Arc::new(LiteBridgeTranslator::for_ble());

        mgr.register_translator(translator, sink, TranslatorRegistrationConfig::ble())
            .await
            .expect("register translator");

        let _handle = mgr
            .start_fanout(Arc::clone(&node), vec!["markers".to_string()])
            .expect("start_fanout");

        // Publish with origin="self" — distinct from the BLE_LITE_BRIDGE
        // transport_id, so fan-out should NOT skip the BLE drain.
        let doc = make_test_doc("marker-uuid-001", "type", "a-f-G-U-C");
        node.publish_with_origin("markers", doc.clone(), Some("self".to_string()))
            .await
            .expect("publish");

        // Wait for observer → fan-out → drain → encode → sink chain to
        // settle. 200 ms is generous for an in-process pipeline.
        timeout(Duration::from_millis(200), async {
            loop {
                if !captured.lock().unwrap().is_empty() {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("OutboundSink should have received envelope bytes within 200ms");

        let captured_bytes = captured.lock().unwrap().clone();
        assert_eq!(
            captured_bytes.len(),
            1,
            "exactly one envelope should have been emitted"
        );

        // Round-trip the captured bytes back through document_codec to
        // verify the wire format is well-formed and reconstructs the
        // original Document.
        let (collection, decoded) =
            document_codec::decode_document(&captured_bytes[0]).expect("decode envelope");
        assert_eq!(collection, "markers");
        assert_eq!(decoded.id.as_deref(), Some("marker-uuid-001"));
        assert_eq!(decoded.fields, doc.fields);
    }

    /// Origin-skip / echo-loop suppression: a doc published with
    /// `origin = Some(BLE_LITE_BRIDGE)` (i.e. the receive side just
    /// bridged something in from BLE) MUST NOT re-emit through this
    /// transport. Without this guarantee, every BLE-ingested doc
    /// loops back to BLE — the failure mode ADR-059's origin field
    /// exists to prevent.
    #[tokio::test]
    async fn fanout_publish_with_ble_lite_origin_skips_emission() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Arc::new(crate::node::Node::new(Arc::clone(&backend)));
        let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));

        let captured = Arc::new(StdMutex::new(Vec::<Vec<u8>>::new()));
        let sink = Arc::new(CapturingSink {
            captured: Arc::clone(&captured),
        });
        let translator = Arc::new(LiteBridgeTranslator::for_ble());

        mgr.register_translator(translator, sink, TranslatorRegistrationConfig::ble())
            .await
            .expect("register translator");

        let _handle = mgr
            .start_fanout(Arc::clone(&node), vec!["markers".to_string()])
            .expect("start_fanout");

        // Publish two docs in order:
        //   1. skip-origin doc (origin=BLE_LITE_BRIDGE) — MUST be dropped.
        //   2. barrier doc (origin="self") — MUST be emitted.
        //
        // The observer task processes events FIFO from a single
        // mpsc receiver, so when (2) lands at the sink we know (1)
        // was already processed. That's a happens-before barrier: it
        // converts "X did not happen yet" into "X did not happen
        // ever" without exposing FanoutState internals or relying on
        // wall-clock generosity. Originally addressed a QA review
        // [SUGGESTION] that timing-based negative assertions can pass
        // for the wrong reason on contended runners.
        let skip_doc = make_test_doc("marker-skip", "type", "a-f-G");
        node.publish_with_origin("markers", skip_doc, Some(BLE_LITE_BRIDGE.to_string()))
            .await
            .expect("publish skip-origin");

        let barrier_doc = make_test_doc("marker-barrier", "type", "a-f-X");
        node.publish_with_origin("markers", barrier_doc, Some("self".to_string()))
            .await
            .expect("publish barrier");

        // Block until barrier doc lands at the sink — at that point
        // the skip-origin doc has demonstrably been processed (and
        // correctly dropped) by the same FIFO observer.
        timeout(Duration::from_millis(200), async {
            loop {
                if !captured.lock().unwrap().is_empty() {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("barrier doc should have reached the sink within 200ms");

        let captured_bytes = captured.lock().unwrap().clone();
        assert_eq!(
            captured_bytes.len(),
            1,
            "exactly one doc (the barrier) should have reached the sink; \
             the skip-origin doc MUST have been suppressed by origin-skip"
        );

        // Confirm the one captured doc is the barrier, not the
        // skip-origin one (defends against a hypothetical bug where
        // origin-skip fires backwards).
        let (_, barrier_back) =
            document_codec::decode_document(&captured_bytes[0]).expect("decode barrier");
        assert_eq!(
            barrier_back.id.as_deref(),
            Some("marker-barrier"),
            "the captured doc MUST be the barrier (origin=\"self\"), \
             not the skip-origin doc — origin-skip is misrouting"
        );
    }

    #[tokio::test]
    async fn decode_inbound_ignores_ctx_collection_envelope_authoritative() {
        // The envelope is the source of truth for collection, not
        // ctx.collection. This locks the design choice: a caller that
        // miscomputes ctx.collection on inbound shouldn't change what
        // we emit. Test that decode_inbound succeeds even with an
        // outright wrong ctx.collection — and that the resulting
        // Document is the same as if ctx.collection were unset.
        let t = LiteBridgeTranslator::for_ble();
        let mut fields = HashMap::new();
        fields.insert("z".to_string(), json!(true));
        let doc = doc_with_fields(Some("d1"), fields.clone());

        let ctx_out = TranslationContext::outbound().with_collection("real");
        let bytes = t.encode_outbound(&doc, &ctx_out).await.expect("encode");

        let ctx_misaligned = TranslationContext::inbound("peer").with_collection("LIES");
        let decoded = t
            .decode_inbound(&bytes, &ctx_misaligned)
            .await
            .expect("decode")
            .expect("non-decline");
        assert_eq!(decoded.id, doc.id);
        assert_eq!(decoded.fields, fields);
    }
}
