postvan 0.4.0

//! A lightweight message-routing hub bridging pub/sub semantics with actor-like concurrency.
//!
//! This crate provides two ways for components to communicate:
//!
//! - **Hub-mediated routing** through a central `Postoffice`.
//! - **Direct peer-to-peer delivery** through cached local contacts on a `Letterbox`.
//!
//! The design is intentionally simple. Each actor or client gets a `Letterbox`, which acts as its inbox/outbox.
//! A central `Postoffice` keeps track of registered letterboxes and subscriptions. Messages can be published
//! by type variant, sent directly by address, or delivered directly to known contacts.
//!

use anyhow::Result;
use papaya::{HashMap, HashSet};
use std::{fmt::Debug, hash::Hash};
use tokio::sync::mpsc::{self, Receiver, Sender};

/// Control messages sent from a `Letterbox` to the central `Postoffice`.
///
/// These are internal routing commands that
/// instruct the post office to publish, route directly, subscribe, or unsubscribe.
enum CommandMessage<A: Address, M: Message> {
    /// Broadcast a message to every letterbox subscribed to this message variant.
    Publish(M),
    /// Register interest in the given message discriminant.
    Subscribe(A, M::Topic),
    /// Remove a letterbox from the subscription list for the given discriminant.
    Unsubscribe(A, M::Topic),
    Unregister(A),
}

/// Marker trait for routable message types.
///
/// Messages must be clonable, debuggable, thread-safe, and `'static` so they can
/// be moved through channels and stored inside the routing hub.
pub trait Message: Clone + Debug + Send + Sync + 'static {
    type Topic: Topic;
    fn topics(&self) -> &'static [Self::Topic];
}

/// Marker trait for topic
pub trait Topic: Eq + Hash + Debug + Send + Sync + 'static {}

/// Marker trait for alias
pub trait Address: Clone + Debug + Hash + Eq + PartialEq + Send + Sync + 'static {}
impl<T> Address for T where T: Clone + Debug + Hash + Eq + PartialEq + Send + Sync + 'static {}

/// Local endpoint used by a participant in the messaging system.
///
/// A letterbox owns an inbox receiver, a sender that the `Postoffice` (or other `Letterbox`es) can use to
/// deliver messages, and optional metadata used for routing.
#[derive(Debug)]
pub struct Letterbox<A: Address, M: Message> {
    /// MPSC sender
    sender: mpsc::Sender<M>,
    /// MPSC receiver
    receiver: mpsc::Receiver<M>,
    address: Option<A>,
    /// MPSC Postoffice sender
    post_tx: Option<Sender<CommandMessage<A, M>>>,
    /// MPSC Letterbox senders
    contacts: HashMap<A, mpsc::Sender<M>>,
}

impl<A: Address, M: Message> Letterbox<A, M> {
    /// Create an unregistered letterbox.
    ///
    /// The letterbox cannot post, subscribe, or unsubscribe until it has been
    /// registered with a `Postoffice`.
    pub fn new() -> Self {
        let (sender, receiver) = mpsc::channel(32);
        Self {
            sender,
            receiver,
            address: None,
            post_tx: None,
            contacts: HashMap::new(),
        }
    }

    /// Publish a message through the central `Postoffice`.
    ///
    /// This sends a `Publish` command to the hub, which will route the message to
    /// all subscribers of the same variant.
    pub fn post(&mut self, message: M) -> Result<()> {
        self.post_tx
            .as_ref()
            .expect("Cannot post anything using an unregistered letterbox")
            .try_send(CommandMessage::Publish(message))?;
        Ok(())
    }

    /// Drain all currently available messages from this letterbox's inbox.
    pub fn recv_all(&mut self) -> Vec<M> {
        let mut messages = Vec::new();
        while let Ok(message) = self.receiver.try_recv() {
            messages.push(message);
        }
        messages
    }

    /// Recieve the oldest 'unread' message *asynchronously* from this letternox's inbox
    pub async fn recv(&mut self) -> Option<M> {
        return self.receiver.recv().await;
    }

    /// Recieve the oldest 'unread' message from this letterbox's inbox
    pub fn recv_now(&mut self) -> Option<M> {
        self.receiver.try_recv().ok()
    }

    /// Recieve the `limit` oldest messages from this letterbox's inbox
    pub fn recv_many(&mut self, limit: usize) -> Vec<M> {
        let mut out = Vec::with_capacity(limit);
        for _ in 0..limit {
            match self.receiver.try_recv() {
                Ok(msg) => out.push(msg),
                Err(_) => break,
            }
        }
        out
    }

    /// Subscribe to all future messages matching the discriminant of `message`.
    ///
    /// The value passed in is used only to identify the variant, not as a payload.
    pub fn subscribe(&mut self, topic: M::Topic) -> Result<()> {
        let (post_tx, address) = self
            .post_tx
            .as_ref()
            .zip(self.address.clone())
            .expect("Cannot subscribe using an unregistered letterbox");

        post_tx.try_send(CommandMessage::Subscribe(address, topic))?;
        Ok(())
    }

    /// Unsubscribe from messages matching the discriminant of `message`.
    ///
    /// The value passed in is used only to identify the variant, not as a payload.
    pub fn unsubscribe(&mut self, topic: M::Topic) -> Result<()> {
        let (post_tx, address) = self
            .post_tx
            .as_ref()
            .zip(self.address.clone())
            .expect("Cannot unsubscribe using an unregistered letterbox");

        post_tx.try_send(CommandMessage::Unsubscribe(address, topic))?;
        Ok(())
    }

    /// Unregister from the postoffice associated with this letterbox
    pub fn unregister(&self) -> Result<()> {
        self.post_tx
            .as_ref()
            .expect("Cannot unregister a letterbox if it hasn't been registered yet")
            .try_send(CommandMessage::Unregister(self.address.clone().unwrap()))?;
        Ok(())
    }

    /// Add a direct contact that can be used for local peer-to-peer delivery.
    ///
    /// The alias is a name chosen by the user that maps to the target's sender.
    /// This bypasses the `Postoffice` entirely.
    pub fn add_contact(&mut self, letterbox: &Letterbox<A, M>, address: A) {
        self.contacts
            .pin()
            .insert(address, letterbox.sender.clone());
    }

    /// Deliver a message directly to a locally known contact.
    ///
    /// This bypasses the central `Postoffice` and sends straight to the target
    /// letterbox's inbox through its channel sender.
    pub fn post_to(&self, address: A, message: M) -> Result<()> {
        let contacts = self.contacts.pin();
        let sender = contacts
            .get(&address)
            .ok_or_else(|| anyhow::anyhow!("Contact not found for alias: {:?}", address))?;
        sender.try_send(message)?;
        Ok(())
    }
}
impl<A: Address, M: Message> Default for Letterbox<A, M> {
    fn default() -> Self {
        Self::new()
    }
}

/// Central routing hub that manages registration and subscription state.
///
/// The post office receives internal control commands from registered letterboxes
/// and uses them to route messages to the appropriate inboxes.
#[derive(Debug)]
pub struct Postoffice<A: Address, M: Message> {
    subscriptions: HashMap<M::Topic, HashSet<A>>,
    registry: HashMap<A, Sender<M>>,
    post_tx: Sender<CommandMessage<A, M>>,
    post_rx: Receiver<CommandMessage<A, M>>,
}

impl<A: Address, M: Message> Postoffice<A, M> {
    /// Create a new, empty post office.
    pub fn new() -> Self {
        let (post_tx, post_rx) = mpsc::channel(128);
        Self {
            subscriptions: HashMap::new(),
            registry: HashMap::new(),
            post_tx,
            post_rx,
        }
    }

    /// Register a letterbox with the hub and assign it a unique address.
    ///
    /// Registration enables the letterbox to post messages and manage subscriptions.
    /// The letterbox's address and hub control sender are stored in the mailbox.
    pub fn register(&self, mailbox: &mut Letterbox<A, M>, address: &A) {
        self.registry
            .pin()
            .insert(address.clone(), mailbox.sender.clone());
        mailbox.address = Some(address.clone());
        mailbox.post_tx = Some(self.post_tx.clone());
    }

    /// Process all pending control messages currently waiting in the hub queue.
    ///
    /// This is the synchronous routing step for publish/direct/subscribe/unsubscribe
    /// commands.
    pub fn tick(&mut self) {
        while let Ok(cmd_message) = self.post_rx.try_recv() {
            self.handle_command(cmd_message);
        }
    }

    /// Asynchronously process control messages from the hub queue.
    ///
    /// The exact receive behavior depends on the enabled async feature.
    /// This method runs until the underlying receive stream ends.
    pub async fn tick_async(&mut self) {
        while let Some(cmd_message) = self.post_rx.recv().await {
            self.handle_command(cmd_message);
        }
    }

    /// Apply a single routing command.
    fn handle_command(&mut self, cmd_message: CommandMessage<A, M>) {
        match cmd_message {
            CommandMessage::Publish(message) => {
                for topic in message.topics() {
                    if let Some(addresses) = self.subscriptions.pin().get(topic) {
                        for address in addresses.pin().iter() {
                            if let Some(sender) = self.registry.pin().get(address) {
                                let _ = sender.try_send(message.clone());
                            }
                        }
                    }
                }
            }
            CommandMessage::Subscribe(address, topic) => {
                let subscriptions = self.subscriptions.pin();
                let subscribers = subscriptions.get_or_insert_with(topic, HashSet::new);
                subscribers.pin().insert(address);
            }
            CommandMessage::Unsubscribe(address, d) => {
                if let Some(subscribers) = self.subscriptions.pin().get(&d) {
                    subscribers.pin().remove(&address);
                }
            }
            CommandMessage::Unregister(address) => {
                self.registry.pin().remove(&address);
                self.subscriptions.pin().retain(|_, set| {
                    set.pin().remove(&address);
                    !set.is_empty()
                });
            }
        }
    }
}

impl<A: Address, M: Message> Default for Postoffice<A, M> {
    fn default() -> Self {
        Self::new()
    }
}