//! BLE [`Translator`] impl, peat-mesh-side (ADR-059 Amendment 4 Slice 4.a)
//!
//! Bridges [`peat_btle::translator::BleTranslator`]'s typed-struct surface
//! ([`BlePosition`], [`BlePeripheral`], [`BleEmergencyEvent`],
//! [`BleCannedMessage`]) into the generic [`Translator`] contract, so
//! `TransportManager`'s fan-out can route through BLE the same way it
//! routes through any other registered transport.
//!
//! ## Layering rule (ADR-059 Amendment 4)
//!
//! Wire-codec primitives — typed BLE structs, postcard scaffolding,
//! 0xB6 framing constants, the `*_to_*` JSON projection helpers — live
//! in peat-btle behind its `translator-codec` feature. The
//! `Translator`-trait *impl* on top of those primitives lives here, in
//! peat-mesh, behind this crate's `bluetooth` feature. Transport
//! crates do not depend on peat-mesh; this module is the one-way
//! adapter.
//!
//! Coexistence with peat-btle's pre-Amendment-4 `mesh-translator`-gated
//! impl is governed by Rust's coherence rule
//! ([E0119](https://doc.rust-lang.org/error_codes/E0119.html), no
//! overlapping impls): a consumer enabling **both**
//! `peat-mesh/bluetooth` and `peat-btle/mesh-translator` simultaneously
//! gets a hard compile error. That's the structural guard the
//! amendment relies on — there is no runtime ambiguity to resolve.
//! Consumers choose one feature; the back-edge `mesh-translator`
//! disappears entirely once Slice 4.b ships.
//!
//! ## Wire format
//!
//! Per ADR-059 Amendment 1, `encode_outbound` produces the fully-framed
//! bytes `[TRANSLATOR_FRAME_MARKER, collection_code, postcard_payload]`
//! and `decode_inbound` takes the payload-only bytes (peat-btle's
//! receive dispatch strips marker + code before invoking).
//!
//! [`Translator`]: crate::transport::Translator
//! [`BlePosition`]: peat_btle::translator::BlePosition
//! [`BlePeripheral`]: peat_btle::translator::BlePeripheral
//! [`BleEmergencyEvent`]: peat_btle::translator::BleEmergencyEvent
//! [`BleCannedMessage`]: peat_btle::translator::BleCannedMessage

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

use anyhow::Context as _;
use async_trait::async_trait;
use serde_json::{Map, Value};

use peat_btle::document::TRANSLATOR_FRAME_MARKER;
use peat_btle::translator::{
    BleCannedMessage, BleEmergencyEvent, BlePeripheral, BlePosition, BleTranslator,
    COLLECTION_CODE_ALERTS, COLLECTION_CODE_CANNED_MESSAGES, COLLECTION_CODE_PLATFORMS,
    COLLECTION_CODE_TRACKS,
};

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

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

    async fn encode_outbound(&self, doc: &Document, ctx: &TranslationContext) -> Option<Vec<u8>> {
        // Per ADR-059 §"Translator trait", the impl distinguishes at
        // telemetry between (a) declines — collection not carried by
        // this codec, normal traffic, no diagnostic — and (b) real
        // encode errors — codec carries the collection but couldn't
        // project the Document into the typed BLE struct (typed
        // projection returned `None`, e.g. malformed shape) or
        // postcard encode failed. Both paths return `None` per the
        // trait's `Option<Vec<u8>>` shape; case (b) emits a
        // `tracing::warn!` so sender-side wire-format drift is
        // visible to operators. Case (a) does not.
        let collection = ctx.collection.as_deref()?;
        let value = mesh_doc_to_value(doc);

        let payload = if collection == self.tracks_collection() {
            warn_on_none(
                "tracks",
                self.track_to_position(&value).and_then(postcard_encode),
            )?
        } else if collection == self.platforms_collection() {
            warn_on_none(
                "platforms",
                self.platform_to_peripheral(&value)
                    .and_then(postcard_encode),
            )?
        } else if collection == self.alerts_collection() {
            warn_on_none(
                "alerts",
                self.alert_to_emergency(&value).and_then(postcard_encode),
            )?
        } else if collection == self.canned_messages_collection() {
            warn_on_none(
                "canned_messages",
                self.doc_to_canned_message(&value).and_then(postcard_encode),
            )?
        } else {
            // Decline — collection not in this codec's table. No
            // telemetry: this is normal cross-transport fan-out
            // declining a doc that belongs to another codec.
            return None;
        };

        let code = collection_to_code(self, collection)?;
        let mut framed = Vec::with_capacity(2 + payload.len());
        framed.push(TRANSLATOR_FRAME_MARKER);
        framed.push(code);
        framed.extend_from_slice(&payload);
        Some(framed)
    }

    async fn decode_inbound(
        &self,
        bytes: &[u8],
        ctx: &TranslationContext,
    ) -> anyhow::Result<Option<Document>> {
        let collection = match ctx.collection.as_deref() {
            Some(c) => c,
            None => return Ok(None),
        };

        let callsign = ctx.local_callsign.as_deref();
        let mesh_id = ctx.cell_id.as_deref();

        let value = if collection == self.tracks_collection() {
            let pos: BlePosition =
                postcard::from_bytes(bytes).context("ble: decode BlePosition (tracks)")?;
            // Track docs key on `peripheral_id`; the wire frame is just
            // BlePosition, so the inbound bridge MUST stamp the
            // peripheral_id (hex u32) onto `ctx.local_wire_id` from the
            // BLE connection identity. A missing/unparsable id used to
            // collapse silently to `0`; per ADR-059 Amendment 1 we surface
            // it as Err so the bridge's parse-error metric flags the
            // misconfiguration.
            let peripheral_id = ctx
                .local_wire_id
                .as_deref()
                .and_then(|s| u32::from_str_radix(s.trim_start_matches("0x"), 16).ok())
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "ble: decode tracks requires ctx.local_wire_id (hex u32 \
                         peripheral_id) — inbound bridge must stamp it from the \
                         BLE connection identity before calling decode_inbound"
                    )
                })?;
            self.position_to_track_in_cell(&pos, peripheral_id, callsign, mesh_id)
        } else if collection == self.platforms_collection() {
            let per: BlePeripheral =
                postcard::from_bytes(bytes).context("ble: decode BlePeripheral (platforms)")?;
            self.peripheral_to_platform_in_cell(&per, mesh_id)
        } else if collection == self.alerts_collection() {
            let em: BleEmergencyEvent =
                postcard::from_bytes(bytes).context("ble: decode BleEmergencyEvent (alerts)")?;
            self.emergency_to_alert(&em, callsign)
        } else if collection == self.canned_messages_collection() {
            let cm: BleCannedMessage = postcard::from_bytes(bytes)
                .context("ble: decode BleCannedMessage (canned_messages)")?;
            self.canned_message_to_doc_in_cell(&cm, callsign, mesh_id)
        } else {
            return Ok(None);
        };

        Ok(Some(value_to_mesh_document(value, collection)?))
    }
}

// === Inline helpers ========================================================
//
// peat-btle's `translator-codec` exposes the typed-struct primitives, the
// `COLLECTION_CODE_*` constants, and the `*_to_*` JSON projections, but the
// `code_to_collection` / `collection_to_code` helpers and the
// MeshDocument↔Value bridge are gated behind peat-btle's `mesh-translator`
// feature — which this crate intentionally does not enable, since enabling
// it would re-introduce the back-edge cycle Amendment 4 is closing. The
// same shape is replicated here, taking `Document` (peat-mesh's type)
// directly rather than going through peat-btle's `MeshDocument` re-export.

/// Map a configured collection name to its translator-frame code. Returns
/// `None` for collections this `BleTranslator` doesn't carry — encode side
/// treats `None` as a decline (per the `Translator::encode_outbound`
/// contract).
fn collection_to_code(t: &BleTranslator, name: &str) -> Option<u8> {
    if name == t.tracks_collection() {
        Some(COLLECTION_CODE_TRACKS)
    } else if name == t.platforms_collection() {
        Some(COLLECTION_CODE_PLATFORMS)
    } else if name == t.alerts_collection() {
        Some(COLLECTION_CODE_ALERTS)
    } else if name == t.canned_messages_collection() {
        Some(COLLECTION_CODE_CANNED_MESSAGES)
    } else {
        None
    }
}

/// Wrap a postcard-encoded payload as `Some(bytes)` on success, `None` on
/// encode failure. Failures are vanishingly rare with serde-derived
/// structs (no manual encode logic to drift); the
/// `Translator::encode_outbound` contract returns `Option<Vec<u8>>` so
/// any failure surfaces as `None`. Surrounding telemetry comes from
/// [`warn_on_none`] which logs a `tracing::warn!` for the
/// known-collection-but-encode-failed case (real wire-format drift)
/// while leaving the unknown-collection decline path silent.
fn postcard_encode<T: serde::Serialize>(value: T) -> Option<Vec<u8>> {
    postcard::to_allocvec(&value).ok()
}

/// Tag a known-collection encode result with telemetry on failure. Used
/// only on the encode side, where a `None` from a typed projection or
/// postcard step (after a known-collection match) indicates a real
/// codec error — distinct from the silent decline that fires for
/// unknown collections. ADR-059 §"Translator trait" requires this
/// distinction at telemetry; without it, sender-side wire-format
/// drift is invisible to operators.
fn warn_on_none(collection: &'static str, payload: Option<Vec<u8>>) -> Option<Vec<u8>> {
    if payload.is_none() {
        // The typed-struct projection or postcard encode declined
        // even though the collection matched. Operators who haven't
        // seen this before should investigate sender-side
        // wire-format drift; e.g. a doc shape changed in a way the
        // `*_to_*` projection no longer recognizes.
        tracing::warn!(
            transport = "ble",
            collection = collection,
            "ble translator: encode failed for known collection — typed-struct \
             projection returned None or postcard encode failed; treating as \
             decline. (See ADR-059 §\"Translator trait\" — this is the \
             encode-error path, not a normal decline.)"
        );
    }
    payload
}

/// Convert a [`Document`] into a [`Value`] for the legacy `*_to_*`
/// methods. `Document.fields` is already `HashMap<String, Value>`; we
/// just wrap it as an Object and stamp the id if present.
///
/// **Id precedence.** When `doc.fields` already contains an `"id"`
/// entry (the typical shape of docs that round-tripped through
/// `value_to_mesh_document` previously), the existing field-side id
/// wins via `.or_insert_with(...)`. `doc.id` only fills in when the
/// fields don't already carry one. This is asymmetric with
/// `value_to_mesh_document`, which reads `value["id"]` for the
/// `Document.id` slot but leaves `id` in `fields` too — so a
/// round-tripped doc can carry id in both places. The `*_to_*`
/// methods all stamp `id` themselves, so on the encode side the
/// fields-side wins by construction.
fn mesh_doc_to_value(doc: &Document) -> Value {
    let mut map = Map::with_capacity(doc.fields.len() + 1);
    for (k, v) in &doc.fields {
        map.insert(k.clone(), v.clone());
    }
    if let Some(id) = &doc.id {
        map.entry("id".to_string())
            .or_insert_with(|| Value::String(id.clone()));
    }
    Value::Object(map)
}

/// Inverse of [`mesh_doc_to_value`] — rebuild a [`Document`] from the
/// `Value` shape the `*_to_*` methods produce.
///
/// `Document.updated_at` is set to `SystemTime::now()` (received-at
/// semantic). The wire structs (`BlePosition`, `BleEmergencyEvent`,
/// `BleCannedMessage`) carry their own timestamps which are preserved
/// inside `fields["last_update"]` / `fields["timestamp"]` for
/// downstream consumers that need source-time ordering. The
/// Document's `updated_at` deliberately uses local receive time so
/// that two peers receiving the same wire frame over different paths
/// each see their own consistent receive order; Automerge's own
/// vector clock provides causal ordering across peers and is
/// unaffected by this field.
///
/// Returns `Err` (not a silent empty Document) when `value` is not an
/// `Object` — this is the wire-format-drift signal the
/// `Translator::decode_inbound` `Result` channel exists for.
fn value_to_mesh_document(value: Value, collection: &str) -> anyhow::Result<Document> {
    let id = value
        .get("id")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string());
    let Value::Object(map) = value else {
        anyhow::bail!(
            "ble: *_to_* projection for collection {collection} returned \
             non-Object Value; this is wire-format drift, not a graceful \
             degrade case"
        );
    };
    let fields: HashMap<String, Value> = map.into_iter().collect();
    Ok(Document {
        id,
        fields,
        updated_at: SystemTime::now(),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use peat_btle::translator::TranslationConfig;
    use serde_json::json;

    #[tokio::test]
    async fn transport_id_is_ble() {
        let t = BleTranslator::new(TranslationConfig::default());
        assert_eq!(t.transport_id(), "ble");
    }

    #[tokio::test]
    async fn encode_outbound_declines_unknown_collection() {
        let t = BleTranslator::new(TranslationConfig::default());
        let doc = Document {
            id: Some("doc-1".into()),
            fields: HashMap::new(),
            updated_at: SystemTime::now(),
        };
        let ctx = TranslationContext::outbound().with_collection("not-a-ble-collection");
        let result = t.encode_outbound(&doc, &ctx).await;
        assert!(result.is_none(), "expected decline for unknown collection");
    }

    #[tokio::test]
    async fn encode_outbound_declines_when_no_collection_in_ctx() {
        let t = BleTranslator::new(TranslationConfig::default());
        let doc = Document {
            id: Some("doc-1".into()),
            fields: HashMap::new(),
            updated_at: SystemTime::now(),
        };
        let ctx = TranslationContext::outbound();
        let result = t.encode_outbound(&doc, &ctx).await;
        assert!(
            result.is_none(),
            "expected decline when ctx.collection is None"
        );
    }

    #[tokio::test]
    async fn decode_inbound_returns_ok_none_when_no_collection() {
        let t = BleTranslator::new(TranslationConfig::default());
        let ctx = TranslationContext::inbound("peer");
        let result = t.decode_inbound(&[], &ctx).await.unwrap();
        assert!(
            result.is_none(),
            "expected Ok(None) when ctx.collection is None — codec decline, not error"
        );
    }

    #[tokio::test]
    async fn decode_inbound_returns_ok_none_for_unknown_collection() {
        let t = BleTranslator::new(TranslationConfig::default());
        let ctx = TranslationContext::inbound("peer").with_collection("not-a-ble-collection");
        let result = t.decode_inbound(&[], &ctx).await.unwrap();
        assert!(
            result.is_none(),
            "expected Ok(None) when collection is unknown — codec decline"
        );
    }

    #[tokio::test]
    async fn platforms_round_trip() {
        let t = BleTranslator::new(TranslationConfig::default());
        // Build a platform doc shaped the way `peripheral_to_platform`
        // emits: id, name, type, status, battery_percent, activity,
        // last_update, ble_origin.
        let mut fields = HashMap::new();
        fields.insert("id".into(), json!("ble-deadbeef"));
        fields.insert("name".into(), json!("test-callsign"));
        fields.insert("type".into(), json!("wearable"));
        fields.insert("status".into(), json!("active"));
        fields.insert("battery_percent".into(), json!(80));
        fields.insert("activity".into(), json!("still"));
        fields.insert("last_update".into(), json!(1_700_000_000_u64));
        fields.insert("ble_origin".into(), json!(true));

        let doc = Document {
            id: Some("ble-deadbeef".into()),
            fields,
            updated_at: SystemTime::now(),
        };

        let ctx = TranslationContext::outbound().with_collection(t.platforms_collection());
        let bytes = t
            .encode_outbound(&doc, &ctx)
            .await
            .expect("platforms encode_outbound should produce framed bytes for a well-formed doc");

        // Verify the wire frame: marker byte + platforms code + postcard
        // payload.
        assert!(
            bytes.len() >= 2,
            "framed wire bytes too short: {} bytes",
            bytes.len()
        );
        assert_eq!(bytes[0], TRANSLATOR_FRAME_MARKER, "first byte must be 0xB6");
        assert_eq!(
            bytes[1], COLLECTION_CODE_PLATFORMS,
            "second byte must be the platforms collection code"
        );

        // Round-trip through decode (note: peat-btle's receive dispatch
        // strips marker+code before invoking decode_inbound, per ADR-059
        // Amendment 1 §"Encode/decode contract for the trait boundary").
        let payload = &bytes[2..];
        let in_ctx = TranslationContext::inbound("peer").with_collection(t.platforms_collection());
        let decoded = t
            .decode_inbound(payload, &in_ctx)
            .await
            .expect("decode_inbound should not error on well-formed bytes")
            .expect("platforms decode_inbound should produce Some(doc) on well-formed bytes");

        // The id stamped by `peripheral_to_platform_in_cell` is
        // `{prefix}{id_hex}` — verify it survives the round-trip.
        assert!(
            decoded.id.as_deref().unwrap_or("").starts_with("ble-"),
            "round-tripped doc id should keep the ble- prefix; got {:?}",
            decoded.id
        );
    }

    #[tokio::test]
    async fn tracks_round_trip() {
        // Build a track doc shaped the way `position_to_track` emits.
        let t = BleTranslator::new(TranslationConfig::default());
        let mut fields = HashMap::new();
        fields.insert("id".into(), json!("ble-deadbeef"));
        fields.insert("lat".into(), json!(40.7128));
        fields.insert("lon".into(), json!(-74.0060));
        fields.insert("hae".into(), json!(10.5));
        fields.insert("cep".into(), json!(2.5));

        let doc = Document {
            id: Some("ble-deadbeef".into()),
            fields,
            updated_at: SystemTime::now(),
        };

        let ctx = TranslationContext::outbound().with_collection(t.tracks_collection());
        let bytes = t
            .encode_outbound(&doc, &ctx)
            .await
            .expect("tracks encode_outbound should produce framed bytes");

        assert_eq!(bytes[0], TRANSLATOR_FRAME_MARKER);
        assert_eq!(bytes[1], COLLECTION_CODE_TRACKS);

        let payload = &bytes[2..];
        // Round-trip — peripheral_id rides on the BLE connection
        // identity (ctx.local_wire_id), not the postcard payload.
        let in_ctx = TranslationContext::inbound("peer-X")
            .with_collection(t.tracks_collection())
            .with_local_wire_id("DEADBEEF");
        let decoded = t
            .decode_inbound(payload, &in_ctx)
            .await
            .expect("tracks decode_inbound should succeed for well-formed input")
            .expect("tracks decode_inbound should produce Some(doc) for well-formed input");

        // The id stamped by `position_to_track_in_cell` derives from
        // peripheral_id (hex u32) — `ble-DEADBEEF` here.
        assert_eq!(decoded.id.as_deref(), Some("ble-DEADBEEF"));
    }

    #[tokio::test]
    async fn tracks_decode_errs_on_missing_local_wire_id() {
        // ADR-059 Amendment 1 explicitly requires Err (not silent
        // collapse to peripheral_id=0) when the inbound bridge fails
        // to stamp ctx.local_wire_id for tracks. This test locks
        // that contract.
        let t = BleTranslator::new(TranslationConfig::default());
        // Encode a valid tracks frame so we have well-formed bytes
        // to decode; the failure mode under test is context-side,
        // not byte-side.
        let mut fields = HashMap::new();
        fields.insert("id".into(), json!("ble-00000001"));
        fields.insert("lat".into(), json!(0.0));
        fields.insert("lon".into(), json!(0.0));
        let doc = Document {
            id: Some("ble-00000001".into()),
            fields,
            updated_at: SystemTime::now(),
        };
        let out_ctx = TranslationContext::outbound().with_collection(t.tracks_collection());
        let bytes = t.encode_outbound(&doc, &out_ctx).await.expect("encode");
        let payload = &bytes[2..];

        // Now decode WITHOUT local_wire_id — must Err.
        let bad_ctx = TranslationContext::inbound("peer-X").with_collection(t.tracks_collection());
        let result = t.decode_inbound(payload, &bad_ctx).await;
        assert!(
            result.is_err(),
            "tracks decode without ctx.local_wire_id MUST Err (silent collapse to \
             peripheral_id=0 is the ADR-059 Amendment 1 hazard this contract closes); \
             got Ok({:?})",
            result.ok()
        );

        // Same with unparsable hex.
        let unparsable_ctx = TranslationContext::inbound("peer-X")
            .with_collection(t.tracks_collection())
            .with_local_wire_id("not-hex");
        assert!(
            t.decode_inbound(payload, &unparsable_ctx).await.is_err(),
            "tracks decode with unparsable local_wire_id MUST Err"
        );
    }

    #[tokio::test]
    async fn alerts_round_trip() {
        let t = BleTranslator::new(TranslationConfig::default());
        // alert_to_emergency requires `type=="emergency"`, hex-string
        // source_node, hex-string ack keys (see
        // peat-btle/src/translator.rs alert_to_emergency for the shape).
        let mut fields = HashMap::new();
        fields.insert("id".into(), json!("ble-emergency-DEADBEEF-1700000000"));
        fields.insert("type".into(), json!("emergency"));
        fields.insert("source_node".into(), json!("DEADBEEF"));
        fields.insert("source".into(), json!("Alpha-1"));
        fields.insert("timestamp".into(), json!(1_700_000_000_u64));
        fields.insert("acks".into(), json!({}));

        let doc = Document {
            id: Some("ble-emergency-DEADBEEF-1700000000".into()),
            fields,
            updated_at: SystemTime::now(),
        };

        let ctx = TranslationContext::outbound().with_collection(t.alerts_collection());
        let bytes = t
            .encode_outbound(&doc, &ctx)
            .await
            .expect("alerts encode_outbound should produce framed bytes");
        assert_eq!(bytes[0], TRANSLATOR_FRAME_MARKER);
        assert_eq!(bytes[1], COLLECTION_CODE_ALERTS);

        let in_ctx = TranslationContext::inbound("peer-X").with_collection(t.alerts_collection());
        let decoded = t
            .decode_inbound(&bytes[2..], &in_ctx)
            .await
            .expect("alerts decode_inbound should succeed")
            .expect("alerts decode_inbound should produce Some(doc)");
        assert!(
            decoded
                .id
                .as_deref()
                .unwrap_or("")
                .starts_with("ble-emergency-"),
            "round-tripped alerts id should keep the ble-emergency- prefix; got {:?}",
            decoded.id
        );
    }

    #[tokio::test]
    async fn canned_messages_round_trip() {
        let t = BleTranslator::new(TranslationConfig::default());
        // doc_to_canned_message requires `type=="canned_message"`,
        // hex-string source_node, hex-string ack keys (see
        // peat-btle/src/translator.rs doc_to_canned_message).
        let mut fields = HashMap::new();
        fields.insert("id".into(), json!("ble-canned-DEADBEEF-1700000000"));
        fields.insert("type".into(), json!("canned_message"));
        fields.insert("source_node".into(), json!("DEADBEEF"));
        fields.insert("source".into(), json!("Alpha-1"));
        fields.insert("message_code".into(), json!(0x10_u8));
        fields.insert("message_name".into(), json!("Check In"));
        fields.insert("timestamp".into(), json!(1_700_000_000_u64));
        fields.insert("sequence".into(), json!(1_u32));
        fields.insert("acks".into(), json!({}));

        let doc = Document {
            id: Some("ble-canned-DEADBEEF-1700000000".into()),
            fields,
            updated_at: SystemTime::now(),
        };

        let ctx = TranslationContext::outbound().with_collection(t.canned_messages_collection());
        let bytes = t
            .encode_outbound(&doc, &ctx)
            .await
            .expect("canned_messages encode_outbound should produce framed bytes");
        assert_eq!(bytes[0], TRANSLATOR_FRAME_MARKER);
        assert_eq!(bytes[1], COLLECTION_CODE_CANNED_MESSAGES);

        let in_ctx =
            TranslationContext::inbound("peer-X").with_collection(t.canned_messages_collection());
        let decoded = t
            .decode_inbound(&bytes[2..], &in_ctx)
            .await
            .expect("canned_messages decode_inbound should succeed")
            .expect("canned_messages decode_inbound should produce Some(doc)");
        assert!(
            decoded
                .id
                .as_deref()
                .unwrap_or("")
                .starts_with("ble-canned-"),
            "round-tripped canned id should keep the ble-canned- prefix; got {:?}",
            decoded.id
        );
    }
}
