//! Cross-transport observer-driven fan-out (ADR-059 Slice 1.a).
//!
//! The orchestration half of ADR-059: `TransportManager` subscribes to
//! [`Node::observe`](crate::node::Node::observe) for every collection the
//! caller declares, and on each [`ChangeEvent`] dispatches the document
//! into per-transport bounded mpsc channels. A long-lived per-transport
//! drain task encodes via [`Translator::encode_outbound`] and routes the
//! resulting bytes through an [`OutboundSink`].
//!
//! See `peat/docs/adr/059-cross-transport-document-bridging.md` for the
//! architectural rationale (origin-on-event, iroh-as-substrate,
//! single-node loop prevention, per-transport channel sizing).
//!
//! # Slice 1.a scope cuts (intentional)
//!
//! - **`Removed` events are silently skipped.** Delete propagation
//!   ships in Slice 2 per ADR-059 §"Delete propagation". A
//!   `transport.delete_dropped` metric is the Slice-1-mitigation
//!   Slice 2 will wire up.
//! - **`allowed_transports` filter is not applied.** The doc field
//!   itself ships in Slice 2; Slice 1.a fans every translator the
//!   non-`Removed` events, subject only to the origin filter.
//! - **Channel coalescing for bandwidth-constrained transports
//!   (LoRa/SBD)** is not implemented. The codecs aren't in tree yet;
//!   coalescing without a real consumer would be speculative. Slice 2
//!   adds it alongside the LoRa/SBD codecs.
//! - **`TransportManager::start_fanout` does not enforce single-Node
//!   binding.** A second call spawns additional observer tasks rather
//!   than rejecting. The orchestrator currently has no per-Node
//!   identity to compare against; tracked for Slice 2.

use std::collections::HashMap;
use std::sync::Arc;

use async_trait::async_trait;
use tokio::sync::{mpsc, RwLock};
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};

use crate::node::Node;
use crate::sync::types::{ChangeEvent, Document, Query};
use crate::transport::manager::TransportManager;
use crate::transport::translator::{TranslationContext, Translator};

/// Per-translator registration config (ADR-059 §"Per-transport channel sizing").
///
/// Channel buffer depth is part of QoS calibration, not an
/// implementation detail — different transport families have different
/// throughput, MTU, and burst tolerance characteristics. A single
/// default would either over-buffer cheap transports (delaying QoS
/// preemption past ADR-019 intent) or under-buffer fast ones (spurious
/// drops on bursts the transport could absorb).
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct TranslatorRegistrationConfig {
    /// Bounded mpsc channel depth for this transport's outbound queue.
    pub outbound_buffer_depth: usize,
}

impl TranslatorRegistrationConfig {
    /// Default for BLE: 32. Per-link MTU ~150 B, ~5–10 frames/sec
    /// sustained. 32 absorbs ~3 s of burst before QoS kicks in;
    /// larger depths would let stale low-priority traffic delay
    /// emergency frames.
    pub fn ble() -> Self {
        Self {
            outbound_buffer_depth: 32,
        }
    }

    /// Default for TAK: 256. TCP-backed, high throughput, low
    /// per-message cost. Depth protects against TAK-server hiccups
    /// without losing traffic.
    pub fn tak() -> Self {
        Self {
            outbound_buffer_depth: 256,
        }
    }

    /// Default for LoRa: 16. Duty-cycle limited (per ADR-052); a
    /// deeper queue is meaningless because the radio can't drain it.
    /// QoS preemption needs to fire early to keep the queue's
    /// contents fresh.
    pub fn lora() -> Self {
        Self {
            outbound_buffer_depth: 16,
        }
    }

    /// Default for SBD: 8. Satellite windows are minutes apart (per
    /// ADR-051); only the highest-priority items survive each window.
    /// Shallow depth forces operators/QoS to pick winners up front
    /// rather than queue indefinitely.
    pub fn sbd() -> Self {
        Self {
            outbound_buffer_depth: 8,
        }
    }
}

impl Default for TranslatorRegistrationConfig {
    fn default() -> Self {
        // No transport-family hint; pick a middle value that's
        // unlikely to be right for any real transport. Operators
        // should use the family-specific constructors above.
        Self {
            outbound_buffer_depth: 64,
        }
    }
}

/// Item enqueued for outbound fan-out on a single transport's channel.
///
/// `doc` is shared via `Arc` rather than cloned per-translator —
/// fanout_doc enqueues the same `Arc<Document>` into N channels, and
/// each per-transport drain task derefs it for `encode_outbound`. This
/// scales O(N transports) reference-bumps per event instead of N full
/// `Document` clones, which matters as either document size or event
/// rate grows.
///
/// Fields are public to give per-transport drain tasks zero-overhead
/// access; the type is `#[non_exhaustive]` so adding fields later is
/// non-breaking.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct FanoutItem {
    pub doc: Arc<Document>,
    pub ctx: TranslationContext,
}

/// Outbound transport-side recipient of encoded bytes.
///
/// For native Rust transports, wrap the underlying transport's send
/// path. For JNI-bound transports (Android plugin's BLE manager), the
/// sink is a thin dispatcher that calls a registered Kotlin callback
/// via the generic `OutboundFrameCallback` JNI surface (Slice 1.b).
///
/// The sink is invoked from the per-transport drain task; one drain
/// task per registered translator means there is no concurrent invocation
/// on a single sink instance, but if the same sink is registered for
/// multiple translators (unusual), it must tolerate concurrent calls.
#[async_trait]
pub trait OutboundSink: Send + Sync {
    async fn send_outbound(&self, bytes: Vec<u8>, ctx: &TranslationContext) -> anyhow::Result<()>;
}

/// QoS-aware drop policy invoked when a per-transport channel is full
/// (per ADR-059 §"Negative consequences" and §"Per-transport channel
/// sizing").
///
/// Slice 1.a ships [`TailDropPolicy`] as the default — drop the new
/// item, log, bump a metric. Slice 2+ will add priority-aware eviction
/// per ADR-019 (evict the lowest-priority queued item before the new
/// arrival) once `allowed_transports` and per-event priority hints
/// land.
pub trait QosDropPolicy: Send + Sync {
    fn handle_full(&self, transport_id: &str, item: FanoutItem);
}

/// Default Slice 1.a drop policy: tail-drop with telemetry.
#[derive(Debug, Default)]
pub struct TailDropPolicy;

impl QosDropPolicy for TailDropPolicy {
    fn handle_full(&self, transport_id: &str, _item: FanoutItem) {
        warn!(
            transport = %transport_id,
            "outbound queue full; tail-drop. Slice 2+ will replace with \
             QoS-aware eviction per ADR-019."
        );
        // metrics::counter!("transport.fanout_dropped_full",
        //                   "transport" => transport_id.to_string()).increment(1);
    }
}

/// Internal state for fan-out orchestration. Owned by `TransportManager`.
pub(crate) struct FanoutState {
    pub translators: RwLock<HashMap<&'static str, Arc<dyn Translator>>>,
    pub outbound_queues: RwLock<HashMap<&'static str, mpsc::Sender<FanoutItem>>>,
    pub qos_drop_policy: Arc<dyn QosDropPolicy>,
    /// Cancellation signal for spawned observer + drain tasks. Cancel
    /// once on `TransportManager` drop or when a `FanoutHandle` is
    /// dropped.
    pub cancel: CancellationToken,
}

impl FanoutState {
    pub fn new() -> Self {
        Self {
            translators: RwLock::new(HashMap::new()),
            outbound_queues: RwLock::new(HashMap::new()),
            qos_drop_policy: Arc::new(TailDropPolicy),
            cancel: CancellationToken::new(),
        }
    }
}

impl Default for FanoutState {
    fn default() -> Self {
        Self::new()
    }
}

/// Handle returned from [`TransportManager::start_fanout`]. Drop to
/// cancel the observer task(s) for the bound Node + collections.
///
/// **Scope of cancellation:** the handle holds a *child* cancellation
/// token derived from `FanoutState::cancel`. Dropping the handle (or
/// calling [`FanoutHandle::cancel`]) cancels the child only —
/// **drain tasks for registered translators are unaffected**, because
/// drain tasks were spawned with the parent token directly. Drain
/// tasks continue running until their input channel is closed
/// (`unregister_translator` removes the sender) or the parent token
/// fires (`TransportManager` is dropped).
///
/// This split exists because dropping a `FanoutHandle` is a
/// natural-feeling action when changing which Node is observed, but
/// "stop fan-out for this Node" should not also tear down the
/// drain-task infrastructure. Each `start_fanout` call gets its own
/// child token, so handles for different Nodes can be cancelled
/// independently.
pub struct FanoutHandle {
    cancel: CancellationToken,
}

impl FanoutHandle {
    /// Explicit cancel; equivalent to dropping.
    pub fn cancel(self) {
        self.cancel.cancel();
    }
}

impl Drop for FanoutHandle {
    fn drop(&mut self) {
        // Cancel the child token only — drain tasks registered against
        // the parent token continue running. See type docs for the
        // rationale.
        self.cancel.cancel();
    }
}

// =========================================================================
// FanoutState methods (the actual orchestration logic)
// =========================================================================
//
// These methods live on FanoutState rather than TransportManager so
// observer tasks can hold Arc<FanoutState> without keeping the
// surrounding manager alive — TransportManager exposes thin proxies
// below for the public API.

impl FanoutState {
    /// Register a translator + sink. Spawns the drain task. Rejects on
    /// transport_id collision per ADR-059 §"Transport ID uniqueness".
    pub async fn register_translator(
        &self,
        translator: Arc<dyn Translator>,
        sink: Arc<dyn OutboundSink>,
        config: TranslatorRegistrationConfig,
    ) -> anyhow::Result<()> {
        let transport_id = translator.transport_id();

        let mut translators = self.translators.write().await;
        let mut queues = self.outbound_queues.write().await;

        if translators.contains_key(transport_id) {
            anyhow::bail!(
                "translator with transport_id {:?} is already registered \
                 (ADR-059 §Transport ID uniqueness invariant)",
                transport_id
            );
        }

        let (tx, rx) = mpsc::channel::<FanoutItem>(config.outbound_buffer_depth);
        translators.insert(transport_id, Arc::clone(&translator));
        queues.insert(transport_id, tx);

        let cancel = self.cancel.clone();
        tokio::spawn(drain_task(transport_id, translator, sink, rx, cancel));

        info!(
            transport = %transport_id,
            depth = config.outbound_buffer_depth,
            "registered translator"
        );
        Ok(())
    }

    pub async fn unregister_translator(&self, transport_id: &str) -> anyhow::Result<()> {
        let mut translators = self.translators.write().await;
        let mut queues = self.outbound_queues.write().await;
        translators.remove(transport_id);
        // Dropping the sender closes the channel; the drain task's
        // `rx.recv().await` returns None and it exits its loop.
        queues.remove(transport_id);
        info!(transport = %transport_id, "unregistered translator");
        Ok(())
    }

    pub async fn has_translator(&self, transport_id: &str) -> bool {
        self.translators.read().await.contains_key(transport_id)
    }

    /// Apply origin filter; try_send to each eligible per-transport
    /// channel. On `Full`, the QoS drop policy fires.
    pub async fn fanout_event(&self, event: &ChangeEvent) {
        match event {
            ChangeEvent::Updated {
                collection,
                document,
                origin,
                ..
            } => {
                let doc = Arc::new(document.clone());
                self.fanout_doc_steady(doc, origin.as_deref(), collection)
                    .await;
            }
            ChangeEvent::Initial {
                collection,
                documents,
            } => {
                // Initial snapshot is observer-startup state. Use
                // **blocking sends** rather than try_send: shallow
                // transport buffers (LoRa=16, SBD=8) would otherwise
                // tail-drop most of the snapshot at startup, defeating
                // ADR-059's expectation that initial state converges
                // across transports. We pay observer-startup latency
                // in exchange for completeness — Initial is one-shot,
                // not on the steady-state hot path.
                for d in documents {
                    let doc = Arc::new(d.clone());
                    self.fanout_doc_blocking(doc, None, collection).await;
                }
            }
            // Slice 1.a scope cut: Removed events are silently skipped.
            // Slice 2 §Delete propagation re-enables them with
            // Translator::encode_delete + per-codec wire support.
            ChangeEvent::Removed { .. } => {
                debug!("ChangeEvent::Removed skipped (Slice 1.a; Slice 2 enables)");
            }
        }
    }

    /// Steady-state fan-out: try_send + QoS drop on `Full`.
    async fn fanout_doc_steady(&self, doc: Arc<Document>, origin: Option<&str>, collection: &str) {
        let queues = self.outbound_queues.read().await;
        for (&transport_id, sender) in queues.iter() {
            if origin == Some(transport_id) {
                continue;
            }
            // Slice 2: allowed_transports filter goes here.
            let ctx = TranslationContext::outbound().with_collection(collection);
            let item = FanoutItem {
                doc: Arc::clone(&doc),
                ctx,
            };
            match sender.try_send(item) {
                Ok(()) => {}
                Err(mpsc::error::TrySendError::Full(item)) => {
                    self.qos_drop_policy.handle_full(transport_id, item);
                }
                Err(mpsc::error::TrySendError::Closed(_)) => {
                    // Per ADR-059 round-8: should remove transport_id
                    // from outbound_queues to avoid repeatedly hitting
                    // Closed. Read lock can't upgrade to write
                    // mid-iteration; Slice 2 adds a reaper task.
                    warn!(
                        transport = %transport_id,
                        "fanout channel closed unexpectedly"
                    );
                }
            }
        }
    }

    /// Initial-snapshot fan-out: blocking `send().await` for
    /// completeness. Snapshots a Vec of (transport_id, sender) under
    /// the read lock then **releases the lock before awaiting** —
    /// holding a read lock across `.await` would deadlock against
    /// `unregister_translator`'s write lock if the sender's channel
    /// is full and a writer is queued behind us.
    ///
    /// **Startup-completeness depends on drain-task health.** If a
    /// drain task panicked (the Slice 1.a gap documented on
    /// `drain_task`), its sender remains in `outbound_queues` with
    /// no consumer; `send().await` blocks until either the manager
    /// is dropped or its parent cancellation fires. The Slice 2
    /// reaper (same fix that handles `TrySendError::Closed` in
    /// steady-state) removes such senders eagerly, eliminating the
    /// wedge. Until then, codec authors writing non-panicking
    /// translators is the operational precondition.
    async fn fanout_doc_blocking(
        &self,
        doc: Arc<Document>,
        origin: Option<&str>,
        collection: &str,
    ) {
        // Snapshot of (transport_id, sender) pairs taken under the
        // read lock. Type aliased to keep clippy::type_complexity
        // quiet without obscuring the intent.
        type SenderSnapshot = Vec<(&'static str, mpsc::Sender<FanoutItem>)>;
        let senders: SenderSnapshot = {
            let queues = self.outbound_queues.read().await;
            queues.iter().map(|(k, v)| (*k, v.clone())).collect()
        };
        for (transport_id, sender) in senders {
            if origin == Some(transport_id) {
                continue;
            }
            let ctx = TranslationContext::outbound().with_collection(collection);
            let item = FanoutItem {
                doc: Arc::clone(&doc),
                ctx,
            };
            if let Err(e) = sender.send(item).await {
                warn!(
                    transport = %transport_id,
                    error = %e,
                    "Initial-snapshot send failed (channel closed)"
                );
            }
        }
    }
}

// =========================================================================
// TransportManager-side proxies + start_fanout
// =========================================================================

impl TransportManager {
    /// Register a translator and its outbound sink. See
    /// [`FanoutState::register_translator`] for semantics.
    pub async fn register_translator(
        &self,
        translator: Arc<dyn Translator>,
        sink: Arc<dyn OutboundSink>,
        config: TranslatorRegistrationConfig,
    ) -> anyhow::Result<()> {
        self.fanout
            .register_translator(translator, sink, config)
            .await
    }

    /// Unregister a translator.
    pub async fn unregister_translator(&self, transport_id: &str) -> anyhow::Result<()> {
        self.fanout.unregister_translator(transport_id).await
    }

    /// Whether a translator is currently registered for `transport_id`.
    pub async fn has_translator(&self, transport_id: &str) -> bool {
        self.fanout.has_translator(transport_id).await
    }

    /// Start observer-driven fan-out from a Node into all registered
    /// translators, for the given collections.
    ///
    /// One observer task per collection. Each task holds an
    /// `Arc<FanoutState>` (not `Arc<TransportManager>`) so the
    /// surrounding manager remains droppable — when the user drops
    /// their last `Arc<TransportManager>`, [`Drop`] fires, the
    /// **parent** cancellation token cancels, and both observer and
    /// drain tasks exit on the next select tick.
    ///
    /// Each call to `start_fanout` derives a **child** cancellation
    /// token from the parent and uses it for the spawned observer
    /// tasks. The returned [`FanoutHandle`] holds this child. Dropping
    /// the handle cancels only the child — observer tasks for this
    /// Node stop, but drain tasks (registered against the parent
    /// token) keep running and continue draining subsequent
    /// observer-task fan-outs from any other call to `start_fanout`.
    ///
    /// Slice 1.a fans out `Updated` and `Initial` only; `Removed` is
    /// silently skipped pending Slice 2's delete-propagation contract.
    pub fn start_fanout(
        &self,
        node: Arc<Node>,
        collections: Vec<String>,
    ) -> anyhow::Result<FanoutHandle> {
        // Child token: cancellation propagates from parent → child
        // (TransportManager drop cancels everything), but cancelling
        // the child does NOT propagate up to the parent (FanoutHandle
        // drop cancels only the observers for this fanout call).
        let child = self.fanout.cancel.child_token();

        for collection in collections {
            let stream = node.observe(&collection, &Query::All)?;
            let state = Arc::clone(&self.fanout);
            let token = child.clone();
            let coll_name = collection.clone();
            tokio::spawn(observer_task(state, stream, coll_name, token));
        }

        Ok(FanoutHandle { cancel: child })
    }
}

// =========================================================================
// Drop for TransportManager
// =========================================================================

impl Drop for TransportManager {
    /// Cancel all spawned fanout tasks (observers + drains) on
    /// manager drop. Safe to be no-op when fanout was never started —
    /// `CancellationToken::cancel` is idempotent.
    ///
    /// This is the cleanup half of the `Arc<FanoutState>`-in-tasks
    /// design: tasks hold the state (not the manager), so the
    /// manager can drop normally; this `Drop` ensures the
    /// cancellation token actually fires when it does, preventing a
    /// task leak when the user dropped their last
    /// `Arc<TransportManager>` reference without first dropping the
    /// `FanoutHandle`.
    fn drop(&mut self) {
        self.fanout.cancel.cancel();
    }
}

// =========================================================================
// Spawned-task bodies (free functions to keep the impl block readable)
// =========================================================================

/// Per-transport drain task body.
///
/// **Known Slice 1.a gap (Slice 2 reaper):** if `encode_outbound`
/// panics, this task dies but the corresponding sender remains in
/// `FanoutState::outbound_queues`. Subsequent `fanout_doc_steady`
/// calls would keep enqueuing into a channel nobody is pulling from
/// — items pile up to the buffer cap, then tail-drop indefinitely
/// (ADR-059 round-3 review). Fixing this requires a reaper task that
/// observes drain-task exit and removes the stale sender, which is
/// the same Slice 2 mechanism that handles `TrySendError::Closed`
/// in `fanout_doc_steady`. Translator authors should write
/// non-panicking codecs.
async fn drain_task(
    transport_id: &'static str,
    translator: Arc<dyn Translator>,
    sink: Arc<dyn OutboundSink>,
    mut rx: mpsc::Receiver<FanoutItem>,
    cancel: CancellationToken,
) {
    loop {
        tokio::select! {
            _ = cancel.cancelled() => {
                debug!(transport = %transport_id, "drain task cancelled");
                break;
            }
            maybe_item = rx.recv() => {
                let Some(item) = maybe_item else {
                    debug!(transport = %transport_id, "drain task channel closed");
                    break;
                };
                match translator.encode_outbound(item.doc.as_ref(), &item.ctx).await {
                    Some(bytes) => {
                        if let Err(e) = sink.send_outbound(bytes, &item.ctx).await {
                            warn!(
                                transport = %transport_id,
                                error = %e,
                                "outbound sink failed"
                            );
                        }
                    }
                    None => {
                        // Codec declined — normal traffic, no diagnostic
                        // (real codec errors are the codec's responsibility
                        // to log + metric per the trait contract).
                        debug!(transport = %transport_id, "encode_outbound declined");
                    }
                }
            }
        }
    }
}

async fn observer_task(
    state: Arc<FanoutState>,
    mut stream: crate::sync::types::ChangeStream,
    collection: String,
    cancel: CancellationToken,
) {
    loop {
        tokio::select! {
            _ = cancel.cancelled() => {
                debug!(%collection, "observer task cancelled");
                break;
            }
            maybe_event = stream.receiver.recv() => {
                let Some(event) = maybe_event else {
                    debug!(%collection, "observer stream closed");
                    break;
                };
                state.fanout_event(&event).await;
            }
        }
    }
}

// =========================================================================
// Tests
// =========================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sync::in_memory::InMemoryBackend;
    use crate::sync::traits::DataSyncBackend;
    use crate::transport::manager::TransportManagerConfig;
    use crate::transport::translator::TranslationContext;
    use std::collections::HashMap as StdHashMap;
    use std::sync::Mutex as StdMutex;
    use tokio::time::{timeout, Duration};

    /// Test translator that records every encode_outbound call so
    /// fan-out behavior is observable from the test side.
    struct RecordingTranslator {
        id: &'static str,
        encoded: Arc<StdMutex<Vec<Document>>>,
    }

    impl RecordingTranslator {
        fn new(id: &'static str) -> (Arc<Self>, Arc<StdMutex<Vec<Document>>>) {
            let encoded = Arc::new(StdMutex::new(Vec::new()));
            let t = Arc::new(Self {
                id,
                encoded: Arc::clone(&encoded),
            });
            (t, encoded)
        }
    }

    #[async_trait]
    impl Translator for RecordingTranslator {
        fn transport_id(&self) -> &'static str {
            self.id
        }
        async fn encode_outbound(
            &self,
            doc: &Document,
            _ctx: &TranslationContext,
        ) -> Option<Vec<u8>> {
            self.encoded.lock().unwrap().push(doc.clone());
            Some(b"encoded".to_vec())
        }
        async fn decode_inbound(
            &self,
            _bytes: &[u8],
            _ctx: &TranslationContext,
        ) -> anyhow::Result<Option<Document>> {
            Ok(None)
        }
    }

    /// No-op sink; test asserts on the recording translator's captured
    /// docs, not on bytes-out-the-wire.
    struct DiscardSink;
    #[async_trait]
    impl OutboundSink for DiscardSink {
        async fn send_outbound(
            &self,
            _bytes: Vec<u8>,
            _ctx: &TranslationContext,
        ) -> anyhow::Result<()> {
            Ok(())
        }
    }

    fn doc(id: &str, body: &str) -> Document {
        let mut fields = StdHashMap::new();
        fields.insert(
            "body".to_string(),
            serde_json::Value::String(body.to_string()),
        );
        let mut d = Document::new(fields);
        d.id = Some(id.to_string());
        d
    }

    /// ADR-059 §Transport ID uniqueness invariant: registering two
    /// translators with the same transport_id is rejected at
    /// registration. Without this, origin tagging and
    /// `allowed_transports` matching become ambiguous.
    #[tokio::test]
    async fn register_translator_rejects_duplicate_transport_id() {
        let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));
        let (t1, _) = RecordingTranslator::new("ble");
        let (t2, _) = RecordingTranslator::new("ble");

        mgr.register_translator(
            t1,
            Arc::new(DiscardSink),
            TranslatorRegistrationConfig::ble(),
        )
        .await
        .expect("first ble translator");

        let err = mgr
            .register_translator(
                t2,
                Arc::new(DiscardSink),
                TranslatorRegistrationConfig::ble(),
            )
            .await
            .expect_err("second ble translator must be rejected");
        assert!(
            err.to_string().contains("already registered"),
            "expected uniqueness error, got: {err}"
        );
    }

    /// ADR-059 single-node echo-loop prevention: when
    /// publish_with_origin stamps origin=Some("ble"), the BLE drain
    /// task does NOT see the document, but other registered transports
    /// do. This is the load-bearing invariant for the entire
    /// cross-transport story; without it, every BLE-ingested doc loops
    /// back to BLE and every iroh-ingested doc loops back to iroh.
    #[tokio::test]
    async fn publish_with_origin_skips_originating_transport() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Arc::new(Node::new(Arc::clone(&backend)));
        let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));

        let (ble_tx, ble_seen) = RecordingTranslator::new("ble");
        let (tak_tx, tak_seen) = RecordingTranslator::new("tak");

        mgr.register_translator(
            ble_tx,
            Arc::new(DiscardSink),
            TranslatorRegistrationConfig::ble(),
        )
        .await
        .unwrap();
        mgr.register_translator(
            tak_tx,
            Arc::new(DiscardSink),
            TranslatorRegistrationConfig::tak(),
        )
        .await
        .unwrap();

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

        // Publish a doc as if ingested via BLE. The orchestrator must
        // skip the BLE drain task and dispatch only to TAK.
        node.publish_with_origin("chats", doc("d1", "from-ble"), Some("ble".to_string()))
            .await
            .unwrap();

        // Wait briefly for the observer→fanout→drain pipeline to
        // settle. The pipeline is fully async but in-process; 200 ms
        // is more than enough on any non-pathological host.
        timeout(Duration::from_millis(200), async {
            loop {
                if !tak_seen.lock().unwrap().is_empty() {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("TAK drain task should have observed the doc within 200ms");

        let ble_docs = ble_seen.lock().unwrap().clone();
        let tak_docs = tak_seen.lock().unwrap().clone();
        assert_eq!(
            ble_docs.len(),
            0,
            "BLE drain task must NOT see a doc with origin=Some(\"ble\") \
             (echo-loop prevention failed)"
        );
        assert_eq!(
            tak_docs.len(),
            1,
            "TAK drain task must see the BLE-origin doc (cross-transport bridging failed)"
        );
        assert_eq!(tak_docs[0].id.as_deref(), Some("d1"));
    }

    /// ADR-059 origin=None case (locally-published or sync-delivered
    /// doc): every registered translator sees the doc. No transport is
    /// special when there's no origin to skip.
    #[tokio::test]
    async fn publish_without_origin_fans_out_to_all_translators() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Arc::new(Node::new(Arc::clone(&backend)));
        let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));

        let (ble_tx, ble_seen) = RecordingTranslator::new("ble");
        let (tak_tx, tak_seen) = RecordingTranslator::new("tak");

        mgr.register_translator(
            ble_tx,
            Arc::new(DiscardSink),
            TranslatorRegistrationConfig::ble(),
        )
        .await
        .unwrap();
        mgr.register_translator(
            tak_tx,
            Arc::new(DiscardSink),
            TranslatorRegistrationConfig::tak(),
        )
        .await
        .unwrap();

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

        // No origin — locally-published or sync-delivered.
        node.publish("chats", doc("d2", "broadcast")).await.unwrap();

        timeout(Duration::from_millis(200), async {
            loop {
                if !ble_seen.lock().unwrap().is_empty() && !tak_seen.lock().unwrap().is_empty() {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("both drain tasks should have observed the doc within 200ms");

        assert_eq!(ble_seen.lock().unwrap().len(), 1);
        assert_eq!(tak_seen.lock().unwrap().len(), 1);
    }

    /// Lifecycle invariant: observer tasks must NOT extend the
    /// manager's lifetime. If they did (by holding
    /// `Arc<TransportManager>`), a user dropping their last `Arc`
    /// reference wouldn't actually drop the manager — the spawned
    /// tasks would keep it alive forever, the cancellation token
    /// would never fire from `Drop`, and the tasks would leak until
    /// process exit.
    ///
    /// This test wraps the manager in an `Arc`, takes a `Weak`
    /// observer, drops every strong reference (including the
    /// `FanoutHandle`), yields, and asserts the `Weak` can't
    /// upgrade — i.e. the manager actually dropped.
    ///
    /// Regression-locks the `Arc<FanoutState>`-in-tasks design fix
    /// made during PR self-review.
    #[tokio::test]
    async fn manager_drops_when_user_releases_last_arc() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Arc::new(Node::new(Arc::clone(&backend)));

        let weak = {
            let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));
            // Register a translator so a drain task is also alive
            // (drain tasks hold Arc<dyn Translator> + Arc<dyn Sink>,
            // not Arc<TransportManager> — verifies the same property
            // for drain tasks too).
            let (t, _) = RecordingTranslator::new("ble");
            mgr.register_translator(
                t,
                Arc::new(DiscardSink),
                TranslatorRegistrationConfig::ble(),
            )
            .await
            .unwrap();
            // Spawn observer tasks for a real Node.
            let _handle = mgr
                .start_fanout(Arc::clone(&node), vec!["chats".to_string()])
                .unwrap();
            // Drop _handle right here? No — we want to verify that
            // manager-drop alone is sufficient. Drop the handle by
            // letting it go out of scope at the end of this block,
            // *along with* mgr. If observer tasks held
            // Arc<TransportManager>, the strong count would still be
            // >0 after this block.
            Arc::downgrade(&mgr)
            // `mgr` and `_handle` drop here together. Both Drop impls
            // fire cancel; observer tasks see cancellation; they exit
            // and release their Arc<FanoutState>. The TransportManager
            // strong count goes to 0; the manager drops.
        };

        // Give the runtime a few ticks to process the cancellation
        // and let observer tasks exit. On a single-thread runtime,
        // a couple of yield_now calls is plenty.
        for _ in 0..16 {
            tokio::task::yield_now().await;
        }

        assert!(
            weak.upgrade().is_none(),
            "TransportManager leaked — a spawned task is holding \
             Arc<TransportManager>, which would defeat ADR-059's \
             observer cancellation contract"
        );
    }

    /// Recording sink — captures every byte payload + the
    /// transport_id (carried implicitly via the sink instance, since
    /// each registration pairs one sink with one translator).
    struct RecordingSink {
        bytes: Arc<StdMutex<Vec<Vec<u8>>>>,
    }

    type SinkBytesLog = Arc<StdMutex<Vec<Vec<u8>>>>;

    impl RecordingSink {
        fn new() -> (Arc<Self>, SinkBytesLog) {
            let bytes = Arc::new(StdMutex::new(Vec::new()));
            (
                Arc::new(Self {
                    bytes: Arc::clone(&bytes),
                }),
                bytes,
            )
        }
    }

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

    /// Recording QoS drop policy — captures every drop call so
    /// tail-drop behavior is observable from the test side.
    struct RecordingDropPolicy {
        drops: Arc<StdMutex<Vec<&'static str>>>,
    }

    impl RecordingDropPolicy {
        fn new() -> (Arc<Self>, Arc<StdMutex<Vec<&'static str>>>) {
            let drops = Arc::new(StdMutex::new(Vec::new()));
            (
                Arc::new(Self {
                    drops: Arc::clone(&drops),
                }),
                drops,
            )
        }
    }

    impl QosDropPolicy for RecordingDropPolicy {
        fn handle_full(&self, transport_id: &str, _item: FanoutItem) {
            // The interner the test uses is a static-string set so we
            // can fish out the &'static str from the id we registered.
            // For the canonical "ble"/"tak" cases this round-trips
            // fine. The `Box::leak` fallback is **test-only and
            // intentional**: process-lived test runs make the leak
            // bounded and unreachable in production. Do not lift this
            // pattern into non-test code — `&'static str` requirements
            // in the orchestrator should be satisfied by const
            // literals on the codec side, not by leaks at runtime.
            let id_static: &'static str = match transport_id {
                "ble" => "ble",
                "tak" => "tak",
                "lora" => "lora",
                "sbd" => "sbd",
                "decline" => "decline",
                "blocking" => "blocking",
                other => Box::leak(other.to_string().into_boxed_str()),
            };
            self.drops.lock().unwrap().push(id_static);
        }
    }

    /// Translator that always declines (returns `None`) on
    /// encode_outbound. Locks the "codec declined" branch in
    /// `drain_task` — bytes never reach the sink.
    struct DecliningTranslator;

    #[async_trait]
    impl Translator for DecliningTranslator {
        fn transport_id(&self) -> &'static str {
            "decline"
        }
        async fn encode_outbound(
            &self,
            _doc: &Document,
            _ctx: &TranslationContext,
        ) -> Option<Vec<u8>> {
            None
        }
        async fn decode_inbound(
            &self,
            _bytes: &[u8],
            _ctx: &TranslationContext,
        ) -> anyhow::Result<Option<Document>> {
            Ok(None)
        }
    }

    /// ADR-059 §"TransportManager fan-out orchestration": the encoded
    /// bytes from a successful `encode_outbound` MUST reach the
    /// registered `OutboundSink`. Without this assertion, the
    /// drain-task → sink hand-off has no test coverage and a future
    /// regression (e.g. the wrong arg passed to send_outbound) could
    /// land silently.
    #[tokio::test]
    async fn encode_bytes_reach_outbound_sink() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Arc::new(Node::new(Arc::clone(&backend)));
        let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));

        let (translator, _seen) = RecordingTranslator::new("ble");
        let (sink, sink_bytes) = RecordingSink::new();

        mgr.register_translator(translator, sink, TranslatorRegistrationConfig::ble())
            .await
            .unwrap();
        let _handle = mgr
            .start_fanout(Arc::clone(&node), vec!["chats".to_string()])
            .unwrap();

        node.publish("chats", doc("d1", "hello")).await.unwrap();

        timeout(Duration::from_millis(200), async {
            loop {
                if !sink_bytes.lock().unwrap().is_empty() {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("sink should have received bytes within 200ms");

        let captured = sink_bytes.lock().unwrap().clone();
        assert_eq!(captured.len(), 1, "exactly one frame to the sink");
        assert_eq!(
            captured[0],
            b"encoded".to_vec(),
            "sink must receive the encoded bytes from encode_outbound"
        );
    }

    /// `encode_outbound` returning `None` is "decline" — the codec
    /// is signalling that this document doesn't belong on this
    /// transport (size limits, format affinity per ADR-038, etc.).
    /// `drain_task` must NOT call the sink in that case. Locks the
    /// `None` branch of `drain_task`'s match.
    #[tokio::test]
    async fn encode_decline_does_not_call_sink() {
        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Arc::new(Node::new(Arc::clone(&backend)));
        let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));

        let (sink, sink_bytes) = RecordingSink::new();
        mgr.register_translator(
            Arc::new(DecliningTranslator),
            sink,
            TranslatorRegistrationConfig::ble(),
        )
        .await
        .unwrap();
        let _handle = mgr
            .start_fanout(Arc::clone(&node), vec!["chats".to_string()])
            .unwrap();

        node.publish("chats", doc("d1", "decline-me"))
            .await
            .unwrap();

        // Give the pipeline plenty of time to (not) deliver. If the
        // drain task incorrectly forwarded the None to the sink as an
        // empty Vec, this test would catch it.
        tokio::time::sleep(Duration::from_millis(100)).await;

        assert!(
            sink_bytes.lock().unwrap().is_empty(),
            "sink must NOT be called when encode_outbound returns None"
        );
    }

    /// ADR-059 §"Per-transport channel sizing" + §"QoS-aware drop":
    /// when a per-transport channel is full, `try_send` returns
    /// `Full`, the orchestrator hands the item to the QoS policy,
    /// and the policy gets exactly one `handle_full` call per drop.
    /// This locks the channel-overflow → policy-invoked contract.
    ///
    /// Constructed via `FanoutState` directly (skipping the
    /// surrounding `TransportManager`) so the policy can be swapped
    /// without exposing the policy field on the public API.
    #[tokio::test]
    async fn full_channel_triggers_qos_drop_policy() {
        // Build a FanoutState with a recording policy.
        let (policy, drops) = RecordingDropPolicy::new();
        let mut state = FanoutState::new();
        state.qos_drop_policy = policy;
        let state = Arc::new(state);

        // Slow sink wedges every send for a long time, so the depth-1
        // channel fills up after the first item and every subsequent
        // enqueue hits `Full`.
        struct SlowSink;
        #[async_trait]
        impl OutboundSink for SlowSink {
            async fn send_outbound(
                &self,
                _bytes: Vec<u8>,
                _ctx: &TranslationContext,
            ) -> anyhow::Result<()> {
                tokio::time::sleep(Duration::from_secs(60)).await;
                Ok(())
            }
        }

        let (t, _) = RecordingTranslator::new("ble");
        state
            .register_translator(
                t,
                Arc::new(SlowSink),
                TranslatorRegistrationConfig {
                    outbound_buffer_depth: 1,
                },
            )
            .await
            .unwrap();

        // Drive fanout_doc_steady directly — bypasses the
        // observer/Node path so the test doesn't depend on
        // observer-task scheduling.
        for i in 0..5 {
            let d = Arc::new(doc(&format!("d{i}"), "burst"));
            state.fanout_doc_steady(d, None, "chats").await;
        }

        // Give the drain task a tick to run; meanwhile the slow sink
        // is wedged so subsequent fanout_doc_steady calls saw Full.
        timeout(Duration::from_millis(500), async {
            loop {
                if !drops.lock().unwrap().is_empty() {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("QoS drop policy should have fired at least once");

        let observed = drops.lock().unwrap().clone();
        assert!(
            !observed.is_empty(),
            "QoS drop policy must be invoked when channel is full"
        );
        assert!(
            observed.iter().all(|t| *t == "ble"),
            "every drop must carry the right transport_id"
        );

        // Cancel so the wedged sleep doesn't outlive the test.
        state.cancel.cancel();
    }

    /// ADR-059 Slice 1.b: TranslationContext.collection is populated
    /// from ChangeEvent.collection during fan-out so codecs can
    /// dispatch by doc type. Locks the plumbing from
    /// fanout_event → fanout_doc_steady → FanoutItem.ctx.collection.
    #[tokio::test]
    async fn fanout_propagates_collection_to_translation_context() {
        // Test translator that records ctx.collection from each
        // encode_outbound call, so we can assert the orchestrator
        // populated it.
        struct CtxCapturingTranslator {
            seen: Arc<StdMutex<Vec<Option<String>>>>,
        }
        #[async_trait]
        impl Translator for CtxCapturingTranslator {
            fn transport_id(&self) -> &'static str {
                "ble"
            }
            async fn encode_outbound(
                &self,
                _doc: &Document,
                ctx: &TranslationContext,
            ) -> Option<Vec<u8>> {
                self.seen.lock().unwrap().push(ctx.collection.clone());
                Some(vec![])
            }
            async fn decode_inbound(
                &self,
                _bytes: &[u8],
                _ctx: &TranslationContext,
            ) -> anyhow::Result<Option<Document>> {
                Ok(None)
            }
        }

        let backend: Arc<dyn DataSyncBackend> = Arc::new(InMemoryBackend::new_initialized());
        let node = Arc::new(Node::new(Arc::clone(&backend)));
        let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));

        let seen = Arc::new(StdMutex::new(Vec::new()));
        let translator = Arc::new(CtxCapturingTranslator {
            seen: Arc::clone(&seen),
        });

        mgr.register_translator(
            translator,
            Arc::new(DiscardSink),
            TranslatorRegistrationConfig::ble(),
        )
        .await
        .unwrap();
        let _handle = mgr
            .start_fanout(Arc::clone(&node), vec!["tracks".to_string()])
            .unwrap();

        node.publish("tracks", doc("d1", "lat-lon")).await.unwrap();

        timeout(Duration::from_millis(200), async {
            loop {
                if !seen.lock().unwrap().is_empty() {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("encode_outbound should have been called within 200ms");

        let observed = seen.lock().unwrap().clone();
        assert_eq!(
            observed,
            vec![Some("tracks".to_string())],
            "ctx.collection must be populated from ChangeEvent.collection \
             so codecs can dispatch by doc type"
        );
    }

    /// Symmetric to `fanout_propagates_collection_to_translation_context`
    /// but for the `ChangeEvent::Initial` path. No in-tree backend
    /// produces `Initial` today (`InMemoryBackend::observe` skips it
    /// by design), so a synthetic event is constructed and passed
    /// directly to `fanout_event` — the only way to exercise the
    /// `fanout_doc_blocking` branch under unit test. Locks the
    /// Initial → blocking-send → ctx.collection plumbing so a future
    /// backend impl emitting `Initial` doesn't silently regress this.
    #[tokio::test]
    async fn initial_event_propagates_collection_to_translation_context() {
        struct CtxCapturingTranslator {
            seen: Arc<StdMutex<Vec<Option<String>>>>,
        }
        #[async_trait]
        impl Translator for CtxCapturingTranslator {
            fn transport_id(&self) -> &'static str {
                "ble"
            }
            async fn encode_outbound(
                &self,
                _doc: &Document,
                ctx: &TranslationContext,
            ) -> Option<Vec<u8>> {
                self.seen.lock().unwrap().push(ctx.collection.clone());
                Some(vec![])
            }
            async fn decode_inbound(
                &self,
                _bytes: &[u8],
                _ctx: &TranslationContext,
            ) -> anyhow::Result<Option<Document>> {
                Ok(None)
            }
        }

        let state = Arc::new(FanoutState::new());
        let seen = Arc::new(StdMutex::new(Vec::new()));
        state
            .register_translator(
                Arc::new(CtxCapturingTranslator {
                    seen: Arc::clone(&seen),
                }),
                Arc::new(DiscardSink),
                TranslatorRegistrationConfig::ble(),
            )
            .await
            .unwrap();

        // Synthesize an Initial event with two docs and pass it
        // directly to fanout_event — exercises the blocking-send path
        // without needing a backend that emits Initial.
        let initial = ChangeEvent::Initial {
            collection: "tracks".to_string(),
            documents: vec![doc("d1", "alpha"), doc("d2", "bravo")],
        };
        state.fanout_event(&initial).await;

        // fanout_doc_blocking uses send().await, so by the time
        // fanout_event returns, items are queued. Wait for the drain
        // task to encode them.
        timeout(Duration::from_millis(200), async {
            loop {
                if seen.lock().unwrap().len() >= 2 {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("both Initial docs should reach the translator within 200ms");

        let observed = seen.lock().unwrap().clone();
        assert_eq!(observed.len(), 2);
        assert!(
            observed.iter().all(|c| c.as_deref() == Some("tracks")),
            "ctx.collection on the Initial path must match \
             ChangeEvent::Initial.collection (got: {observed:?})"
        );

        // Cancel so drain task exits cleanly when the test ends.
        state.cancel.cancel();
    }

    /// has_translator + unregister_translator round-trip.
    #[tokio::test]
    async fn unregister_translator_removes_from_registry() {
        let mgr = Arc::new(TransportManager::new(TransportManagerConfig::default()));
        let (t, _) = RecordingTranslator::new("ble");

        mgr.register_translator(
            t,
            Arc::new(DiscardSink),
            TranslatorRegistrationConfig::ble(),
        )
        .await
        .unwrap();
        assert!(mgr.has_translator("ble").await);

        mgr.unregister_translator("ble").await.unwrap();
        assert!(!mgr.has_translator("ble").await);
    }
}
