//! Zenoh Transport - Distributed pub/sub and query for inter-node communication.
//!
//! Provides a Zenoh-based transport layer for cross-node MemorySignal events,
//! WAL entry replication, and distributed memory queries.
//!
//! This module is designed as an optional enhancement to the local EventBus.
//! Local-only operation (via EventBus) continues to work without Zenoh.

use crate::signal::MemorySignal;
use crate::wal::SagaEntry;
use mimirs_core::{AgentId, Memory, MimirError};
use std::sync::Arc;

/// ZenohTransport - Zenoh-based transport for cross-node communication.
pub struct ZenohTransport {
    /// The Zenoh session.
    session: Arc<zenoh::Session>,
}

impl ZenohTransport {
    /// Creates a new ZenohTransport with the given Zenoh configuration.
    ///
    /// This is an async constructor that opens a Zenoh session.
    pub async fn new(config: zenoh::Config) -> Result<Self, MimirError> {
        let session = zenoh::open(config)
            .await
            .map_err(|e| MimirError::Internal(format!("Failed to open Zenoh session: {}", e)))?;
        Ok(Self {
            session: Arc::new(session),
        })
    }

    /// Creates a new ZenohTransport from an existing Zenoh session.
    pub fn from_session(session: zenoh::Session) -> Self {
        Self {
            session: Arc::new(session),
        }
    }

    /// Returns a reference to the underlying Zenoh session.
    pub fn session(&self) -> &zenoh::Session {
        &self.session
    }

    /// Publishes a MemorySignal event to a Zenoh topic.
    ///
    /// Topic format: `mimirs/signal/{signal_type}`
    pub fn publish_signal(&self, signal: &MemorySignal) -> Result<(), MimirError> {
        let topic = format!("mimirs/signal/{}", signal.signal_type());
        let payload = bincode::serialize(signal)
            .map_err(|e| MimirError::Serialization(e.to_string()))?;

        use zenoh::Wait;
        self.session
            .put(&topic, payload)
            .wait()
            .map_err(|e| MimirError::Internal(format!("Zenoh publish failed: {}", e)))?;

        Ok(())
    }

    /// Subscribes to MemorySignal events on the cluster.
    ///
    /// Topic format: `mimirs/signal/>` (wildcard subscription)
    ///
    /// Returns a Subscriber with the default FifoChannel handler.
    pub async fn subscribe_signals(
        &self,
        _agent_id: AgentId,
    ) -> Result<zenoh::pubsub::Subscriber<zenoh::handlers::fifo::FifoChannelHandler<zenoh::sample::Sample>>, MimirError> {
        let subscriber = self
            .session
            .declare_subscriber("mimirs/signal/>")
            .await
            .map_err(|e| MimirError::Internal(format!("Zenoh subscribe failed: {}", e)))?;

        Ok(subscriber)
    }

    /// Queries for memories across the cluster.
    ///
    /// Sends a query to the Zenoh topic `mimirs/query/memory` and collects responses.
    pub fn query_memory(&self, query: &str) -> Result<Vec<Memory>, MimirError> {
        use zenoh::Wait;

        let reply_handler = self
            .session
            .get(format!("mimirs/query/memory/{}", query))
            .wait()
            .map_err(|e| MimirError::Internal(format!("Zenoh query failed: {}", e)))?;

        let mut memories = Vec::new();
        // Drain all available replies from the FifoChannel
        while let Ok(Some(reply)) = reply_handler.try_recv() {
            match reply.result() {
                Ok(sample) => {
                    let bytes = sample.payload().to_bytes();
                    if let Ok(memory) =
                        bincode::deserialize::<Memory>(bytes.as_ref())
                    {
                        memories.push(memory);
                    }
                }
                Err(_) => continue,
            }
        }

        Ok(memories)
    }

    /// Replicates a WAL entry to the cluster.
    ///
    /// Publishes the entry to `mimirs/wal/replicate` for other nodes to consume.
    pub fn replicate_entry(&self, entry: &SagaEntry) -> Result<(), MimirError> {
        let topic = "mimirs/wal/replicate";
        let payload = bincode::serialize(entry)
            .map_err(|e| MimirError::Serialization(e.to_string()))?;

        use zenoh::Wait;
        self.session
            .put(topic, payload)
            .wait()
            .map_err(|e| MimirError::Internal(format!("Zenoh replicate failed: {}", e)))?;

        Ok(())
    }
}

/// ClusterEventBus - EventBus that uses Zenoh for cross-node communication.
///
/// This wraps ZenohTransport and provides the same interface as EventBus,
/// but broadcasts signals to all nodes in the cluster via Zenoh pub/sub.
pub struct ClusterEventBus {
    transport: ZenohTransport,
}

impl ClusterEventBus {
    /// Creates a new ClusterEventBus with the given Zenoh configuration.
    pub async fn new(config: zenoh::Config) -> Result<Self, MimirError> {
        let transport = ZenohTransport::new(config).await?;
        Ok(Self { transport })
    }

    /// Creates a new ClusterEventBus from an existing Zenoh session.
    pub fn from_session(session: zenoh::Session) -> Result<Self, MimirError> {
        let transport = ZenohTransport::from_session(session);
        Ok(Self { transport })
    }

    /// Broadcasts a signal to all nodes in the cluster.
    pub fn broadcast(&self, signal: MemorySignal) -> Result<(), MimirError> {
        self.transport.publish_signal(&signal)
    }

    /// Sends a signal to a specific agent via Zenoh.
    pub fn send_to_agent(
        &self,
        agent_id: AgentId,
        signal: MemorySignal,
    ) -> Result<(), MimirError> {
        let topic = format!("mimirs/signal/agent/{}", agent_id);
        let payload = bincode::serialize(&signal)
            .map_err(|e| MimirError::Serialization(e.to_string()))?;

        use zenoh::Wait;
        self.transport
            .session
            .put(&topic, payload)
            .wait()
            .map_err(|e| MimirError::Internal(format!("Zenoh send failed: {}", e)))?;

        Ok(())
    }

    /// Subscribes to cluster-wide signals.
    pub async fn subscribe(
        &self,
        agent_id: AgentId,
    ) -> Result<zenoh::pubsub::Subscriber<zenoh::handlers::fifo::FifoChannelHandler<zenoh::sample::Sample>>, MimirError> {
        self.transport.subscribe_signals(agent_id).await
    }
}
