// message_bus.rs --- T M3.2 typed in-process message bus with `MessagePack`
// serialization, a schema registry, and structured schema-mismatch errors.

//! Message bus core (T M3.2).
//!
//! Spec contract: a single typed protocol used uniformly for every
//! producer the editor talks to ([spec §5.2]). T M3.2 implements only
//! the in-process transport --- two crossbeam channels wired into a
//! pair of [`BusEnd`]s --- but the wire format (`MessagePack`) is the
//! same one out-of-process workers will eventually use, so a future
//! migration of one topic from in-process to subprocess is a transport
//! swap, not a serialisation rewrite.
//!
//! # Topology
//!
//! [`MessageBus::pair`] returns two [`BusEnd`]s. Each end owns a
//! [`Sender`] aimed at the *other* end and a [`Receiver`] reading
//! messages aimed at *itself*. Cloning a [`BusEnd`] yields another
//! handle that shares the same channels --- typical use is for the
//! main thread to keep one end and hand a clone of the other end to
//! every worker via the closure captured by [`crate::worker::WorkerPool::dispatch`].
//!
//! # Schema registry
//!
//! Topics are `&'static str` keys. Each topic is bound (at runtime,
//! once) to a Rust type via [`SchemaRegistry::register`]. Sending
//! validates the payload type against the registry; decoding
//! validates the requested type against the registry. Both produce
//! [`BusError::SchemaMismatch`] with `&'static str` topic and type
//! names so the error payload itself does not allocate.
//!
//! # Hot-path allocation
//!
//! The send path performs exactly one allocation, the
//! `MessagePack`-encoded payload buffer. Topic strings are
//! `&'static str`, registry lookups use the borrowed key, and the
//! envelope construction moves the payload buffer rather than copying
//! it. Internal `crossbeam_channel` node allocation is implementation
//! detail of the channel and is acceptable per the spec acceptance
//! ("beyond payload buffer").

use std::any::{TypeId, type_name};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use std::time::Duration;

use crossbeam::channel::{self, Receiver, Sender};
use serde::Serialize;
use serde::de::DeserializeOwned;

/// Stable identifier for an envelope, monotonically increasing per
/// [`MessageBus`] from `0`.
pub type MessageId = u64;

// ---------------------------------------------------------------------------
// Errors
// ---------------------------------------------------------------------------

/// Structured failures from the bus. All variants use `&'static str`
/// for topic and type names so the error itself does not allocate ---
/// codec errors are wrapped owned because the underlying
/// `rmp_serde` types are owned.
#[derive(Debug)]
pub enum BusError {
    /// A topic was used for send/decode but not previously
    /// [`SchemaRegistry::register`]ed.
    UnregisteredTopic {
        /// The topic that was not registered.
        topic: &'static str,
    },
    /// The Rust type used for send/decode does not match the type the
    /// topic was registered with.
    SchemaMismatch {
        /// The topic involved in the mismatch.
        topic: &'static str,
        /// The fully-qualified type name supplied by the caller.
        expected: &'static str,
        /// The fully-qualified type name the registry was bound to.
        registered: &'static str,
    },
    /// A topic was registered twice with two different types.
    /// Re-registering with the *same* type is a no-op and not an error.
    DuplicateRegistration {
        /// The contested topic.
        topic: &'static str,
        /// The type name from the prior registration.
        existing: &'static str,
        /// The type name attempted in the second registration.
        attempted: &'static str,
    },
    /// `MessagePack` encoding failed.
    Encode(rmp_serde::encode::Error),
    /// `MessagePack` decoding failed.
    Decode(rmp_serde::decode::Error),
    /// The peer end of the channel has been dropped; no further
    /// sends or receives will succeed.
    Disconnected,
    /// Non-blocking receive found nothing.
    Empty,
}

impl std::fmt::Display for BusError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::UnregisteredTopic { topic } => {
                write!(f, "unregistered topic {topic:?}")
            }
            Self::SchemaMismatch {
                topic,
                expected,
                registered,
            } => {
                write!(
                    f,
                    "schema mismatch on topic {topic:?}: caller used {expected}, \
                     registry has {registered}"
                )
            }
            Self::DuplicateRegistration {
                topic,
                existing,
                attempted,
            } => {
                write!(
                    f,
                    "topic {topic:?} already registered as {existing}; cannot \
                     re-register as {attempted}"
                )
            }
            Self::Encode(e) => write!(f, "MessagePack encode failed: {e}"),
            Self::Decode(e) => write!(f, "MessagePack decode failed: {e}"),
            Self::Disconnected => write!(f, "bus peer is gone"),
            Self::Empty => write!(f, "no message available"),
        }
    }
}

impl std::error::Error for BusError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Encode(e) => Some(e),
            Self::Decode(e) => Some(e),
            _ => None,
        }
    }
}

// ---------------------------------------------------------------------------
// SchemaRegistry
// ---------------------------------------------------------------------------

#[derive(Clone, Copy, Debug)]
struct TopicEntry {
    type_id: TypeId,
    type_name: &'static str,
}

/// Maps topic names to the Rust types their payloads serialize to.
///
/// Senders register each topic up front (typically at editor startup
/// or at package load); send and decode operations validate against
/// the registry. A registry is cheap to share via [`Arc`] --- all
/// runtime methods take `&self` and use a single internal `RwLock`.
#[derive(Debug, Default)]
pub struct SchemaRegistry {
    topics: RwLock<HashMap<&'static str, TopicEntry>>,
}

impl SchemaRegistry {
    /// Build an empty registry.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Bind `topic` to type `T`. Re-registering with the same `T`
    /// is a no-op; re-registering with a different type returns
    /// [`BusError::DuplicateRegistration`].
    pub fn register<T: 'static>(&self, topic: &'static str) -> Result<(), BusError> {
        let new = TopicEntry {
            type_id: TypeId::of::<T>(),
            type_name: type_name::<T>(),
        };
        let mut guard = self.topics.write().expect("schema registry poisoned");
        if let Some(existing) = guard.get(topic) {
            if existing.type_id == new.type_id {
                return Ok(());
            }
            return Err(BusError::DuplicateRegistration {
                topic,
                existing: existing.type_name,
                attempted: new.type_name,
            });
        }
        guard.insert(topic, new);
        Ok(())
    }

    /// Validate that `topic` is registered and bound to type `T`.
    /// Used internally on the send and decode hot paths.
    fn check<T: 'static>(&self, topic: &'static str) -> Result<(), BusError> {
        let guard = self.topics.read().expect("schema registry poisoned");
        let Some(entry) = guard.get(topic) else {
            return Err(BusError::UnregisteredTopic { topic });
        };
        if entry.type_id != TypeId::of::<T>() {
            return Err(BusError::SchemaMismatch {
                topic,
                expected: type_name::<T>(),
                registered: entry.type_name,
            });
        }
        Ok(())
    }

    /// Number of topics currently registered. Useful in tests.
    #[must_use]
    pub fn len(&self) -> usize {
        self.topics.read().expect("schema registry poisoned").len()
    }

    /// Whether the registry is empty.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

// ---------------------------------------------------------------------------
// Envelope
// ---------------------------------------------------------------------------

/// A typed message in transit. The bus moves envelopes between
/// [`BusEnd`]s; the payload is `MessagePack`-encoded bytes whose Rust
/// type is determined by `topic` via the [`SchemaRegistry`].
#[derive(Clone, Debug)]
pub struct Envelope {
    /// Per-bus monotonic id.
    pub id: MessageId,
    /// Schema key naming the payload type.
    pub topic: &'static str,
    /// `MessagePack`-encoded payload bytes.
    pub payload: Vec<u8>,
}

// ---------------------------------------------------------------------------
// BusEnd
// ---------------------------------------------------------------------------

/// One end of a [`MessageBus`]. Send goes to the *other* end; recv
/// reads messages aimed at *this* end. Cloning yields another handle
/// that shares the same underlying channels and id counter.
///
/// `BusEnd` is `Send + Sync`; clones can move to other threads and
/// each one can send concurrently. Recv calls from multiple clones
/// race --- exactly one clone will see any given envelope.
#[derive(Clone)]
pub struct BusEnd {
    out: Sender<Envelope>,
    inbox: Receiver<Envelope>,
    registry: Arc<SchemaRegistry>,
    next_id: Arc<AtomicU64>,
}

impl std::fmt::Debug for BusEnd {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BusEnd")
            .field("inbox_len", &self.inbox.len())
            .field("registered_topics", &self.registry.len())
            .finish_non_exhaustive()
    }
}

impl BusEnd {
    /// Encode `payload` as `MessagePack` and send to the peer end.
    /// Validates `topic` against the schema registry first; an
    /// unregistered or mismatched topic returns a structured
    /// [`BusError`] without touching the channel.
    pub fn send<T>(&self, topic: &'static str, payload: &T) -> Result<MessageId, BusError>
    where
        T: Serialize + 'static,
    {
        self.registry.check::<T>(topic)?;
        let bytes = rmp_serde::to_vec(payload).map_err(BusError::Encode)?;
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let env = Envelope {
            id,
            topic,
            payload: bytes,
        };
        self.out.send(env).map_err(|_| BusError::Disconnected)?;
        Ok(id)
    }

    /// Block until an envelope arrives at this end. Returns
    /// [`BusError::Disconnected`] if the peer has dropped its sender
    /// without queuing further messages.
    pub fn recv(&self) -> Result<Envelope, BusError> {
        self.inbox.recv().map_err(|_| BusError::Disconnected)
    }

    /// Non-blocking receive: returns [`BusError::Empty`] if no
    /// envelope is queued, or [`BusError::Disconnected`] if the peer
    /// is gone and the queue is drained.
    pub fn try_recv(&self) -> Result<Envelope, BusError> {
        match self.inbox.try_recv() {
            Ok(env) => Ok(env),
            Err(channel::TryRecvError::Empty) => Err(BusError::Empty),
            Err(channel::TryRecvError::Disconnected) => Err(BusError::Disconnected),
        }
    }

    /// Block up to `timeout` waiting for the next envelope.
    pub fn recv_timeout(&self, timeout: Duration) -> Result<Envelope, BusError> {
        match self.inbox.recv_timeout(timeout) {
            Ok(env) => Ok(env),
            Err(channel::RecvTimeoutError::Timeout) => Err(BusError::Empty),
            Err(channel::RecvTimeoutError::Disconnected) => Err(BusError::Disconnected),
        }
    }

    /// Decode an envelope's payload as type `T`. Validates the
    /// envelope's topic against the schema registry: if the
    /// registered type for `env.topic` is not `T`, returns
    /// [`BusError::SchemaMismatch`] without invoking the codec.
    pub fn decode<T>(&self, env: &Envelope) -> Result<T, BusError>
    where
        T: DeserializeOwned + 'static,
    {
        self.registry.check::<T>(env.topic)?;
        rmp_serde::from_slice(&env.payload).map_err(BusError::Decode)
    }

    /// Borrow the schema registry. Useful for callers that hold the
    /// pair but want to register topics through the same `Arc` the
    /// bus is using.
    #[must_use]
    pub fn registry(&self) -> &Arc<SchemaRegistry> {
        &self.registry
    }
}

// ---------------------------------------------------------------------------
// MessageBus
// ---------------------------------------------------------------------------

/// Factory for paired [`BusEnd`]s. The bus type itself holds no
/// state; it exists as a namespace for [`Self::pair`].
pub struct MessageBus;

impl MessageBus {
    /// Build a pair of bidirectional [`BusEnd`]s sharing `registry`
    /// and a single per-bus id counter. Either end can send and
    /// receive; messages are routed point-to-point only (each end
    /// has one peer).
    #[must_use]
    pub fn pair(registry: Arc<SchemaRegistry>) -> (BusEnd, BusEnd) {
        let (a_to_b_tx, a_to_b_rx) = channel::unbounded::<Envelope>();
        let (b_to_a_tx, b_to_a_rx) = channel::unbounded::<Envelope>();
        let next_id = Arc::new(AtomicU64::new(0));
        let a = BusEnd {
            out: a_to_b_tx,
            inbox: b_to_a_rx,
            registry: Arc::clone(&registry),
            next_id: Arc::clone(&next_id),
        };
        let b = BusEnd {
            out: b_to_a_tx,
            inbox: a_to_b_rx,
            registry,
            next_id,
        };
        (a, b)
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::worker::WorkerPool;
    use serde::Deserialize;
    use std::time::Instant;

    #[derive(Debug, PartialEq, Serialize, Deserialize)]
    struct Hit {
        line: u32,
        col: u32,
        snippet: String,
    }

    fn registry_with<T: 'static>(topic: &'static str) -> Arc<SchemaRegistry> {
        let r = Arc::new(SchemaRegistry::new());
        r.register::<T>(topic).expect("register");
        r
    }

    /// Schema registration is idempotent for the same type; a
    /// conflicting registration is a structured error.
    #[test]
    fn schema_register_idempotent_for_same_type() {
        let reg = SchemaRegistry::new();
        reg.register::<u32>("ping").unwrap();
        reg.register::<u32>("ping").unwrap();
        assert_eq!(reg.len(), 1);
    }

    #[test]
    fn schema_register_conflicting_type_is_structured_error() {
        let reg = SchemaRegistry::new();
        reg.register::<u32>("ping").unwrap();
        match reg.register::<String>("ping") {
            Err(BusError::DuplicateRegistration {
                topic,
                existing,
                attempted,
            }) => {
                assert_eq!(topic, "ping");
                assert!(existing.contains("u32"), "existing was {existing}");
                assert!(attempted.contains("String"), "attempted was {attempted}");
            }
            other => panic!("expected DuplicateRegistration, got {other:?}"),
        }
    }

    /// Acceptance bullet 1: messages can be sent between the two
    /// ends. Smoke test on a paired bus, no workers involved yet.
    #[test]
    fn round_trip_preserves_value_in_each_direction() {
        let reg = registry_with::<Hit>("search.hit");
        let (a, b) = MessageBus::pair(reg);
        let original = Hit {
            line: 7,
            col: 3,
            snippet: "fn main".into(),
        };
        a.send("search.hit", &original).unwrap();
        let env = b.recv().unwrap();
        let got: Hit = b.decode(&env).unwrap();
        assert_eq!(got, original);

        let echo = Hit {
            line: 8,
            col: 4,
            snippet: "}".into(),
        };
        b.send("search.hit", &echo).unwrap();
        let env = a.recv().unwrap();
        let got: Hit = a.decode(&env).unwrap();
        assert_eq!(got, echo);
    }

    /// Per-bus id counter is monotonic across both directions.
    #[test]
    fn message_ids_are_monotonic_across_directions() {
        let reg = registry_with::<u32>("n");
        let (a, b) = MessageBus::pair(reg);
        let id1 = a.send("n", &1u32).unwrap();
        let id2 = b.send("n", &2u32).unwrap();
        let id3 = a.send("n", &3u32).unwrap();
        assert!(id1 < id2 && id2 < id3);
    }

    /// Acceptance bullet 2 (send side): unregistered topic is a
    /// structured error and never reaches the channel.
    #[test]
    fn send_unregistered_topic_returns_structured_error() {
        let reg = Arc::new(SchemaRegistry::new());
        let (a, _b) = MessageBus::pair(reg);
        match a.send("nope", &1u32) {
            Err(BusError::UnregisteredTopic { topic }) => assert_eq!(topic, "nope"),
            other => panic!("expected UnregisteredTopic, got {other:?}"),
        }
    }

    /// Acceptance bullet 2 (send side): mismatched type is a
    /// structured error and never reaches the channel.
    #[test]
    fn send_with_wrong_type_returns_schema_mismatch() {
        let reg = registry_with::<u32>("n");
        let (a, _b) = MessageBus::pair(reg);
        match a.send("n", &"hi".to_string()) {
            Err(BusError::SchemaMismatch {
                topic,
                expected,
                registered,
            }) => {
                assert_eq!(topic, "n");
                assert!(expected.contains("String"), "expected = {expected}");
                assert!(registered.contains("u32"), "registered = {registered}");
            }
            other => panic!("expected SchemaMismatch, got {other:?}"),
        }
    }

    /// Acceptance bullet 2 (decode side): pulling an envelope and
    /// decoding it at the wrong type fails before the codec runs.
    #[test]
    fn decode_with_wrong_type_returns_schema_mismatch() {
        let reg = registry_with::<u32>("n");
        let (a, b) = MessageBus::pair(reg);
        a.send("n", &42u32).unwrap();
        let env = b.recv().unwrap();
        let result: Result<String, _> = b.decode(&env);
        match result {
            Err(BusError::SchemaMismatch { topic, .. }) => assert_eq!(topic, "n"),
            other => panic!("expected SchemaMismatch on decode, got {other:?}"),
        }
    }

    /// Acceptance bullet 1, integrated path: a worker pool job sends
    /// a result through the bus; the main thread receives and
    /// decodes it. This is the canonical M3.2 use case --- T M3.3
    /// will wrap this in a Lua coroutine API.
    #[test]
    fn worker_dispatched_job_sends_through_bus() {
        let reg = registry_with::<u64>("worker.result");
        let (main_end, worker_end) = MessageBus::pair(reg);
        let pool = WorkerPool::new(2);

        let _ = pool.dispatch(move |_| {
            // A real job would fetch a snapshot, compute, and send.
            let answer: u64 = (1..=10).sum();
            worker_end.send("worker.result", &answer).unwrap();
        });

        let env = main_end
            .recv_timeout(Duration::from_secs(2))
            .expect("worker reply");
        let got: u64 = main_end.decode(&env).unwrap();
        assert_eq!(got, 55);
    }

    /// Multiple workers each holding a clone of the `BusEnd` should
    /// all be able to send concurrently; the main end collects every
    /// envelope.
    #[test]
    fn many_workers_share_one_bus_end_clone() {
        const N: usize = 50;
        let reg = registry_with::<u32>("worker.tick");
        let (main_end, worker_end) = MessageBus::pair(reg);
        let pool = WorkerPool::new(4);

        for i in 0..N {
            let bus = worker_end.clone();
            let _ = pool.dispatch(move |_| {
                #[allow(clippy::cast_possible_truncation)]
                bus.send("worker.tick", &(i as u32)).unwrap();
            });
        }

        let mut collected = Vec::with_capacity(N);
        for _ in 0..N {
            let env = main_end
                .recv_timeout(Duration::from_secs(2))
                .expect("worker tick");
            let v: u32 = main_end.decode(&env).unwrap();
            collected.push(v);
        }
        collected.sort_unstable();
        let expected: Vec<u32> = (0..N).map(|i| u32::try_from(i).unwrap()).collect();
        assert_eq!(collected, expected);
    }

    /// `try_recv` distinguishes empty from disconnected.
    #[test]
    fn try_recv_reports_empty_then_disconnected() {
        let reg = registry_with::<u32>("n");
        let (a, b) = MessageBus::pair(reg);
        assert!(matches!(a.try_recv(), Err(BusError::Empty)));
        drop(b);
        // After the peer drops, try_recv reports Disconnected (queue
        // was empty when the peer left).
        assert!(matches!(a.try_recv(), Err(BusError::Disconnected)));
    }

    /// Acceptance bullet 3: round-trip latency under a microsecond
    /// budget. The spec target is 50 microseconds in release; we
    /// assert a generous upper bound here so debug-mode CI stays
    /// green, and rely on the loop average rather than any single
    /// iteration to absorb scheduler noise. The point is to fail
    /// loudly if anyone introduces a per-message allocation storm
    /// (a String topic, a clone of the registry `HashMap`, a Box
    /// per send, etc).
    #[test]
    fn round_trip_latency_is_within_budget() {
        const ITERS: usize = 5_000;
        let reg = registry_with::<u32>("ping");
        let (a, b) = MessageBus::pair(reg);

        // Warm-up to amortise channel bootstrap.
        for _ in 0..256 {
            a.send("ping", &1u32).unwrap();
            let _ = b.recv().unwrap();
        }

        let start = Instant::now();
        for i in 0..ITERS {
            #[allow(clippy::cast_possible_truncation)]
            a.send("ping", &(i as u32)).unwrap();
            let env = b.recv().unwrap();
            let _: u32 = b.decode(&env).unwrap();
        }
        let elapsed = start.elapsed();
        #[allow(clippy::cast_possible_truncation)]
        let per_op = elapsed / ITERS as u32;

        // Spec target: 50 microseconds in release. Debug builds run
        // serde without optimisations, so we assert a 500-microsecond
        // ceiling that still catches order-of-magnitude regressions.
        let ceiling = if cfg!(debug_assertions) {
            Duration::from_micros(500)
        } else {
            Duration::from_micros(50)
        };
        assert!(
            per_op < ceiling,
            "round-trip averaged {per_op:?} (target {ceiling:?})"
        );
    }

    /// Acceptance bullet 4 (sentinel): the send hot path produces
    /// exactly one `Vec<u8>` allocation --- the payload buffer ---
    /// per call. We test this indirectly by verifying that the
    /// envelope's payload buffer is the *only* heap-bound piece of
    /// state on the wire (topic is `&'static str`, ids are `u64`).
    /// Compile-time evidence: changing `topic` to `String` would
    /// fail this test by adding String allocations per send.
    #[test]
    fn envelope_topic_is_borrowed_static_str() {
        let reg = registry_with::<u32>("ping");
        let (a, b) = MessageBus::pair(reg);
        a.send("ping", &1u32).unwrap();
        let env = b.recv().unwrap();
        // Static-str equality is by pointer, not by content, when
        // both literals are deduplicated by the linker. Confirm the
        // envelope's topic *is* the same static slot as the literal.
        assert!(std::ptr::eq(env.topic, "ping"));
        // Payload buffer length matches a single u32 in `MessagePack`
        // (positive fixint or u8 form, ≤ 2 bytes).
        assert!(
            env.payload.len() <= 2,
            "payload was {} bytes, expected ≤2 for a small u32",
            env.payload.len()
        );
    }

    /// Disconnect detection: after the peer drops, send returns
    /// Disconnected without panicking.
    #[test]
    fn send_after_peer_drop_returns_disconnected() {
        let reg = registry_with::<u32>("n");
        let (a, b) = MessageBus::pair(reg);
        drop(b);
        assert!(matches!(a.send("n", &1u32), Err(BusError::Disconnected)));
    }
}