/******************************************************************************
   Author: Joaquín Béjar García
   Email: jb@taunais.com
   Date: 2/10/25
******************************************************************************/

//! Multi-book management with centralized trade event routing.
//!
//! This module provides book management through a trait-based design, with implementations
//! for both standard library (`BookManagerStd`) and Tokio (`BookManagerTokio`) channels.

use crate::orderbook::OrderBook;
use crate::orderbook::error::ManagerError;
use crate::orderbook::mass_cancel::MassCancelResult;
use crate::orderbook::trade::{TradeEvent, TradeListener, TradeResult};
use pricelevel::{Hash32, OrderType, Side, TimestampMs};
use std::collections::HashMap;
use std::sync::Arc;
use tracing::{error, info};

/// Trait for managing multiple order books with centralized trade event routing.
///
/// This trait defines the interface for book managers, allowing different
/// implementations using various channel types (std::mpsc, tokio::mpsc, etc.).
pub trait BookManager<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    /// Add a new order book for a symbol with an automatically configured trade listener.
    ///
    /// # Errors
    ///
    /// Returns [`ManagerError::BookAlreadyExists`] if a book already exists for
    /// `symbol` — `add_book` refuses to overwrite it, which would silently drop
    /// the existing book's resting orders and order locations.
    fn add_book(&mut self, symbol: &str) -> Result<(), ManagerError>;

    /// Get a reference to an order book by symbol.
    fn get_book(&self, symbol: &str) -> Option<&OrderBook<T>>;

    /// Get a mutable reference to an order book by symbol.
    fn get_book_mut(&mut self, symbol: &str) -> Option<&mut OrderBook<T>>;

    /// Get the list of all symbols with order books in this manager.
    fn symbols(&self) -> Vec<String>;

    /// Remove an order book for a specific symbol.
    fn remove_book(&mut self, symbol: &str) -> Option<OrderBook<T>>;

    /// Check if a book exists for a specific symbol.
    fn has_book(&self, symbol: &str) -> bool;

    /// Get the number of order books in this manager.
    fn book_count(&self) -> usize;
}

/// BookManager implementation using standard library mpsc channels.
///
/// # Trade-event channel is unbounded by design
///
/// Trade events are pushed onto a `std::sync::mpsc` channel, which is
/// **unbounded**. This is deliberate: the matching path must never block to
/// deliver an audit event, so the producer cannot apply backpressure (a bounded
/// channel would force the synchronous matching path to block or to silently
/// drop trade events — both unacceptable).
///
/// **Start the processor before submitting orders.** The receiver sits in an
/// `Option` until [`start_trade_processor`](Self::start_trade_processor) is
/// called. If the consumer is never started — or lags persistently — trade
/// events buffer **without bound** and grow memory. Call `start_trade_processor`
/// before routing order flow, and keep the consumer draining at least as fast as
/// trades are produced.
pub struct BookManagerStd<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    /// Collection of order books indexed by symbol
    books: HashMap<String, OrderBook<T>>,
    /// Sender for trade events
    trade_sender: std::sync::mpsc::Sender<TradeEvent>,
    /// Receiver for trade events (taken when processor starts)
    trade_receiver: Option<std::sync::mpsc::Receiver<TradeEvent>>,
}

impl<T> BookManagerStd<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    /// Create a new BookManagerStd with a standard library mpsc channel.
    pub fn new() -> Self {
        let (sender, receiver) = std::sync::mpsc::channel();

        Self {
            books: HashMap::new(),
            trade_sender: sender,
            trade_receiver: Some(receiver),
        }
    }

    /// Start the trade event processor in a separate thread.
    ///
    /// **Call this before submitting orders.** The trade-event channel is
    /// unbounded (see the [type-level docs](BookManagerStd)); until the processor
    /// is running, every trade event buffers in the channel without bound. Once
    /// started, the spawned thread drains the receiver for the manager's
    /// lifetime.
    ///
    /// # Errors
    ///
    /// Returns [`ManagerError::ProcessorAlreadyStarted`] if the processor has
    /// already been started (the receiver was already taken).
    pub fn start_trade_processor(&mut self) -> Result<std::thread::JoinHandle<()>, ManagerError> {
        let receiver = self
            .trade_receiver
            .take()
            .ok_or(ManagerError::ProcessorAlreadyStarted)?;

        Ok(std::thread::spawn(move || {
            info!("Trade processor started");

            while let Ok(trade_event) = receiver.recv() {
                Self::process_trade_event(trade_event);
            }

            info!("Trade processor stopped");
        }))
    }

    /// Process a single trade event.
    fn process_trade_event(event: TradeEvent) {
        info!(
            "Processing trade for {}: {} trades, executed quantity: {}",
            event.symbol,
            event.trade_result.match_result.trades().as_vec().len(),
            event
                .trade_result
                .match_result
                .executed_quantity()
                .map_or(0, |q| q.as_u64())
        );

        for trade in event.trade_result.match_result.trades().as_vec() {
            info!(
                "  Trade: {} units at price {} (ID: {})",
                trade.quantity(),
                trade.price(),
                trade.trade_id()
            );
        }
    }
}

impl<T> BookManagerStd<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    /// Cancel all orders across all managed books.
    ///
    /// Returns a map from symbol to the [`MassCancelResult`] for that book.
    /// Books with no orders produce a result with `cancelled_count == 0`.
    ///
    /// # Examples
    ///
    /// ```
    /// use orderbook_rs::orderbook::manager::{BookManager, BookManagerStd};
    /// use pricelevel::{Id, Side, TimeInForce};
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut mgr: BookManagerStd<()> = BookManagerStd::new();
    /// mgr.add_book("BTC/USD")?;
    /// mgr.add_book("ETH/USD")?;
    ///
    /// if let Some(book) = mgr.get_book("BTC/USD") {
    ///     book.add_limit_order(Id::new_uuid(), 100, 10, Side::Buy, TimeInForce::Gtc, None).ok();
    /// }
    ///
    /// let results = mgr.cancel_all_across_books();
    /// assert!(results.contains_key("BTC/USD"));
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    pub fn cancel_all_across_books(&self) -> HashMap<String, MassCancelResult> {
        self.books
            .iter()
            .map(|(symbol, book)| (symbol.clone(), book.cancel_all_orders()))
            .collect()
    }

    /// Cancel all orders for a specific user across all managed books.
    ///
    /// Returns a map from symbol to the [`MassCancelResult`] for that book.
    ///
    /// # Arguments
    ///
    /// * `user_id` — the user whose orders should be cancelled
    #[must_use]
    pub fn cancel_by_user_across_books(
        &self,
        user_id: Hash32,
    ) -> HashMap<String, MassCancelResult> {
        self.books
            .iter()
            .map(|(symbol, book)| (symbol.clone(), book.cancel_orders_by_user(user_id)))
            .collect()
    }

    /// Cancel all orders on a specific side across all managed books.
    ///
    /// Returns a map from symbol to the [`MassCancelResult`] for that book.
    ///
    /// # Arguments
    ///
    /// * `side` — the side to cancel ([`Side::Buy`] or [`Side::Sell`])
    #[must_use]
    pub fn cancel_by_side_across_books(&self, side: Side) -> HashMap<String, MassCancelResult> {
        self.books
            .iter()
            .map(|(symbol, book)| (symbol.clone(), book.cancel_orders_by_side(side)))
            .collect()
    }

    /// Evict expired resting orders from a single managed book.
    ///
    /// Pass-through to [`OrderBook::evict_expired_orders`]. `now_ms` is
    /// caller-supplied Unix milliseconds (see that method for the boundary and
    /// determinism contract). Returns `None` when `symbol` is not managed, or
    /// `Some(evicted)` with the evicted orders in the book's documented
    /// deterministic order (empty when nothing expired).
    #[must_use]
    pub fn evict_expired_orders(
        &self,
        symbol: &str,
        now_ms: TimestampMs,
    ) -> Option<Vec<Arc<OrderType<T>>>> {
        self.books
            .get(symbol)
            .map(|book| book.evict_expired_orders(now_ms))
    }

    /// Evict expired resting orders across all managed books at `now_ms`.
    ///
    /// Returns a map from symbol to that book's evicted orders (in the book's
    /// documented deterministic order). Books with nothing expired map to an
    /// empty vector. `now_ms` is caller-supplied Unix milliseconds.
    #[must_use]
    pub fn evict_expired_across_books(
        &self,
        now_ms: TimestampMs,
    ) -> HashMap<String, Vec<Arc<OrderType<T>>>> {
        self.books
            .iter()
            .map(|(symbol, book)| (symbol.clone(), book.evict_expired_orders(now_ms)))
            .collect()
    }
}

impl<T> BookManager<T> for BookManagerStd<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    fn add_book(&mut self, symbol: &str) -> Result<(), ManagerError> {
        if self.books.contains_key(symbol) {
            return Err(ManagerError::BookAlreadyExists {
                symbol: symbol.to_string(),
            });
        }

        let sender = self.trade_sender.clone();
        let symbol_clone = symbol.to_string();

        let trade_listener: TradeListener = Arc::new(move |trade_result: &TradeResult| {
            let trade_event = TradeEvent {
                symbol: trade_result.symbol.clone(),
                trade_result: trade_result.clone(),
                timestamp: crate::current_time_millis(),
                engine_seq: trade_result.engine_seq,
            };

            if let Err(e) = sender.send(trade_event) {
                error!("Failed to send trade event for {}: {}", symbol_clone, e);
            }
        });

        let book = OrderBook::with_trade_listener(symbol, trade_listener);
        self.books.insert(symbol.to_string(), book);
        info!("Added order book for symbol: {}", symbol);
        Ok(())
    }

    fn get_book(&self, symbol: &str) -> Option<&OrderBook<T>> {
        self.books.get(symbol)
    }

    fn get_book_mut(&mut self, symbol: &str) -> Option<&mut OrderBook<T>> {
        self.books.get_mut(symbol)
    }

    fn symbols(&self) -> Vec<String> {
        self.books.keys().cloned().collect()
    }

    fn remove_book(&mut self, symbol: &str) -> Option<OrderBook<T>> {
        let result = self.books.remove(symbol);
        if result.is_some() {
            info!("Removed order book for symbol: {}", symbol);
        }
        result
    }

    fn has_book(&self, symbol: &str) -> bool {
        self.books.contains_key(symbol)
    }

    fn book_count(&self) -> usize {
        self.books.len()
    }
}

impl<T> Default for BookManagerStd<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    fn default() -> Self {
        Self::new()
    }
}

/// BookManager implementation using Tokio mpsc channels.
///
/// # Trade-event channel is unbounded by design
///
/// Trade events are pushed onto a `tokio::sync::mpsc::unbounded_channel`. As with
/// [`BookManagerStd`], this is deliberate: the matching path must never block to
/// deliver an audit event, so the producer applies no backpressure (a bounded
/// channel would force the synchronous matching path to block or to silently drop
/// trade events).
///
/// **Start the processor before submitting orders.** The receiver sits in an
/// `Option` until [`start_trade_processor`](Self::start_trade_processor) is
/// called. If the consumer is never started — or lags persistently — trade
/// events buffer **without bound** and grow memory. Start the processor before
/// routing order flow and keep the consumer draining at least as fast as trades
/// are produced.
pub struct BookManagerTokio<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    /// Collection of order books indexed by symbol
    books: HashMap<String, OrderBook<T>>,
    /// Sender for trade events
    trade_sender: tokio::sync::mpsc::UnboundedSender<TradeEvent>,
    /// Receiver for trade events (taken when processor starts)
    trade_receiver: Option<tokio::sync::mpsc::UnboundedReceiver<TradeEvent>>,
}

impl<T> BookManagerTokio<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    /// Create a new BookManagerTokio with a Tokio unbounded mpsc channel.
    pub fn new() -> Self {
        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();

        Self {
            books: HashMap::new(),
            trade_sender: sender,
            trade_receiver: Some(receiver),
        }
    }

    /// Start the trade event processor as an async task.
    ///
    /// **Call this before submitting orders.** The trade-event channel is
    /// unbounded (see the [type-level docs](BookManagerTokio)); until the
    /// processor is running, every trade event buffers in the channel without
    /// bound. Once started, the spawned task drains the receiver for the
    /// manager's lifetime.
    ///
    /// # Errors
    ///
    /// Returns [`ManagerError::ProcessorAlreadyStarted`] if the processor has
    /// already been started (the receiver was already taken).
    pub fn start_trade_processor(&mut self) -> Result<tokio::task::JoinHandle<()>, ManagerError> {
        let mut receiver = self
            .trade_receiver
            .take()
            .ok_or(ManagerError::ProcessorAlreadyStarted)?;

        Ok(tokio::spawn(async move {
            info!("Trade processor started (Tokio)");

            while let Some(trade_event) = receiver.recv().await {
                Self::process_trade_event(trade_event);
            }

            info!("Trade processor stopped (Tokio)");
        }))
    }

    /// Process a single trade event.
    fn process_trade_event(event: TradeEvent) {
        info!(
            "Processing trade for {}: {} trades, executed quantity: {}",
            event.symbol,
            event.trade_result.match_result.trades().as_vec().len(),
            event
                .trade_result
                .match_result
                .executed_quantity()
                .map_or(0, |q| q.as_u64())
        );

        for trade in event.trade_result.match_result.trades().as_vec() {
            info!(
                "  Trade: {} units at price {} (ID: {})",
                trade.quantity(),
                trade.price(),
                trade.trade_id()
            );
        }
    }
}

impl<T> BookManagerTokio<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    /// Cancel all orders across all managed books.
    ///
    /// Returns a map from symbol to the [`MassCancelResult`] for that book.
    /// Books with no orders produce a result with `cancelled_count == 0`.
    #[must_use]
    pub fn cancel_all_across_books(&self) -> HashMap<String, MassCancelResult> {
        self.books
            .iter()
            .map(|(symbol, book)| (symbol.clone(), book.cancel_all_orders()))
            .collect()
    }

    /// Cancel all orders for a specific user across all managed books.
    ///
    /// Returns a map from symbol to the [`MassCancelResult`] for that book.
    ///
    /// # Arguments
    ///
    /// * `user_id` — the user whose orders should be cancelled
    #[must_use]
    pub fn cancel_by_user_across_books(
        &self,
        user_id: Hash32,
    ) -> HashMap<String, MassCancelResult> {
        self.books
            .iter()
            .map(|(symbol, book)| (symbol.clone(), book.cancel_orders_by_user(user_id)))
            .collect()
    }

    /// Cancel all orders on a specific side across all managed books.
    ///
    /// Returns a map from symbol to the [`MassCancelResult`] for that book.
    ///
    /// # Arguments
    ///
    /// * `side` — the side to cancel ([`Side::Buy`] or [`Side::Sell`])
    #[must_use]
    pub fn cancel_by_side_across_books(&self, side: Side) -> HashMap<String, MassCancelResult> {
        self.books
            .iter()
            .map(|(symbol, book)| (symbol.clone(), book.cancel_orders_by_side(side)))
            .collect()
    }

    /// Evict expired resting orders from a single managed book.
    ///
    /// Pass-through to [`OrderBook::evict_expired_orders`]. `now_ms` is
    /// caller-supplied Unix milliseconds (see that method for the boundary and
    /// determinism contract). Returns `None` when `symbol` is not managed, or
    /// `Some(evicted)` with the evicted orders in the book's documented
    /// deterministic order (empty when nothing expired).
    #[must_use]
    pub fn evict_expired_orders(
        &self,
        symbol: &str,
        now_ms: TimestampMs,
    ) -> Option<Vec<Arc<OrderType<T>>>> {
        self.books
            .get(symbol)
            .map(|book| book.evict_expired_orders(now_ms))
    }

    /// Evict expired resting orders across all managed books at `now_ms`.
    ///
    /// Returns a map from symbol to that book's evicted orders (in the book's
    /// documented deterministic order). Books with nothing expired map to an
    /// empty vector. `now_ms` is caller-supplied Unix milliseconds.
    #[must_use]
    pub fn evict_expired_across_books(
        &self,
        now_ms: TimestampMs,
    ) -> HashMap<String, Vec<Arc<OrderType<T>>>> {
        self.books
            .iter()
            .map(|(symbol, book)| (symbol.clone(), book.evict_expired_orders(now_ms)))
            .collect()
    }
}

impl<T> BookManager<T> for BookManagerTokio<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    fn add_book(&mut self, symbol: &str) -> Result<(), ManagerError> {
        if self.books.contains_key(symbol) {
            return Err(ManagerError::BookAlreadyExists {
                symbol: symbol.to_string(),
            });
        }

        let sender = self.trade_sender.clone();
        let symbol_clone = symbol.to_string();

        let trade_listener: TradeListener = Arc::new(move |trade_result: &TradeResult| {
            let trade_event = TradeEvent {
                symbol: trade_result.symbol.clone(),
                trade_result: trade_result.clone(),
                timestamp: crate::current_time_millis(),
                engine_seq: trade_result.engine_seq,
            };

            if let Err(e) = sender.send(trade_event) {
                error!("Failed to send trade event for {}: {}", symbol_clone, e);
            }
        });

        let book = OrderBook::with_trade_listener(symbol, trade_listener);
        self.books.insert(symbol.to_string(), book);
        info!("Added order book for symbol: {}", symbol);
        Ok(())
    }

    fn get_book(&self, symbol: &str) -> Option<&OrderBook<T>> {
        self.books.get(symbol)
    }

    fn get_book_mut(&mut self, symbol: &str) -> Option<&mut OrderBook<T>> {
        self.books.get_mut(symbol)
    }

    fn symbols(&self) -> Vec<String> {
        self.books.keys().cloned().collect()
    }

    fn remove_book(&mut self, symbol: &str) -> Option<OrderBook<T>> {
        let result = self.books.remove(symbol);
        if result.is_some() {
            info!("Removed order book for symbol: {}", symbol);
        }
        result
    }

    fn has_book(&self, symbol: &str) -> bool {
        self.books.contains_key(symbol)
    }

    fn book_count(&self) -> usize {
        self.books.len()
    }
}

impl<T> Default for BookManagerTokio<T>
where
    T: Clone + Send + Sync + Default + 'static,
{
    fn default() -> Self {
        Self::new()
    }
}
