/* Copyright 2023 Architect Financial Technologies LLC. This is free
 * software released under the GNU Affero Public License version 3. */

//! clients for talking to various parts of architect

use crate::{
    config::Common,
    pool,
    protocol::{
        alerts::Alert,
        limits::{LimitScope, LimitSet},
        orderflow::{
            AberrantFill, Fill, FillId, FromOms, NormalCptyReject, NormalFill, OmsReject,
            Order, OrderId, ToOms,
        },
        Account, Dir,
    },
    symbology::{Product, Route, TradableProduct, Venue},
};
use anyhow::{anyhow, bail, Result};
use futures::{channel::mpsc as fmpsc, StreamExt};
use fxhash::FxHashMap;
use log::debug;
use netidx::{
    pool::Pooled,
    subscriber::{Event, FromValue, SubId, UpdatesFlags, Value},
};
use netidx_archive::logfile::Seek;
use netidx_core::path::Path;
use rust_decimal::Decimal;
use std::{collections::HashMap, ops::Deref, result, sync::Arc};
use tokio::{
    sync::{broadcast, RwLock},
    task::JoinHandle,
};

pub mod alerts;
pub mod limits;
pub mod managed_order;
pub mod managed_product;
pub mod orderflow;
pub mod secrets;
pub mod simple_orderflow;
pub mod symbology;
pub mod symbology_loader;

pub struct ClientInner {
    common: Common,
    symbology: symbology_loader::Client,
    orderflow: orderflow::Client,
    simple_orderflow: simple_orderflow::OmsQueryApi,
    limits: limits::OmsLimitsApi,
    secrets: secrets::SecretsQueryApi,
    alerts_tx: broadcast::Sender<Pooled<Vec<Alert>>>,
    _alerts_rx: broadcast::Receiver<Pooled<Vec<Alert>>>,
    default_account: Account,
    managed_orders: RwLock<FxHashMap<OrderId, Arc<managed_order::ManagedOrder>>>,
}

/// Managed client for Architect.
///
/// Implements some default behavior for setup and provides a simple
/// and clean interface + convenient helpers for getting started
/// without obfuscating the actual core API--that is, it should be
/// easy to graduate from this interface as the user requirements
/// become more complex.  Think "training wheels" and not "opaque
/// re-skin".
///
/// That being said, the interface should be sufficient for all kinds
/// of low-frequency, discretionary trading.  Probably around >1000
/// open orders at any given time is the cutoff for setting things up
/// more manually.
///
/// The lower level oms api is in `orderflow`, lower level qf
/// functionality is direct from netidx.
#[derive(Clone)]
pub struct Client(Arc<ClientInner>);

impl Deref for Client {
    type Target = ClientInner;

    fn deref(&self) -> &Self::Target {
        &*self.0
    }
}

impl Client {
    /// This will initialize every api, in some cases it can take a
    /// while. It should not be necessary to have more than one of
    /// these in a process.
    pub async fn new(common: &Common) -> Result<(Self, JoinHandle<Result<()>>)> {
        debug!("client loading symbology");
        let symbology = symbology_loader::Client::new(common).await?;
        debug!("waiting for symbology...");
        symbology.wait_caught_up().await?;
        debug!("client connecting simple orderflow");
        let simple_orderflow = simple_orderflow::OmsQueryApi::new(common).await?;
        let chanid = simple_orderflow.get_channel_id().await?;
        debug!("client connecting orderflow");
        let orderflow = orderflow::Client::new(common, Some(chanid)).await?;
        debug!("client connecting limits api");
        let limits = limits::OmsLimitsApi::new(common).await?;
        debug!("client connecting secrets api");
        let secrets = secrets::SecretsQueryApi::new(common).await?;
        let (alerts_tx, alerts_rx) = broadcast::channel::<Pooled<Vec<Alert>>>(10000);
        debug!("client started");
        let t = Self(Arc::new(ClientInner {
            common: common.clone(),
            symbology,
            orderflow,
            simple_orderflow,
            limits,
            secrets,
            alerts_tx,
            _alerts_rx: alerts_rx,
            default_account: Account::get("DEFAULT"),
            managed_orders: RwLock::new(HashMap::default()),
        }));

        let closed = t.run();
        Ok((t, closed))
    }

    /// This is a wrapper around new that will load the default api
    /// configuration if you don't already have one. If you do have
    /// one, then it is the same as calling new.
    pub async fn initialize(
        common: Option<&Common>,
    ) -> Result<(Self, tokio::task::JoinHandle<Result<()>>)> {
        let common = match common {
            Some(common) => common.clone(),
            None => Common::load_default().await?,
        };
        Self::new(&common).await
    }

    /// Return a reference to the api common config.
    pub fn common(&self) -> &Common {
        &self.common
    }

    /// Return a reference to the symbology client
    pub fn symbology(&self) -> &symbology_loader::Client {
        &self.symbology
    }

    // CR alee: don't let user [recv] on the orderflow, that would be mistake; still want to let the
    // user access [orderflow] for sending [ToOms] though
    /// Return a reference to the low level orderflow client. If you
    /// are using simple_orderflow directly, or ManagedOrder then it
    /// isn't safe to receive messages directly from this client.
    pub fn orderflow(&self) -> &orderflow::Client {
        &self.orderflow
    }

    /// Return a reference to the simple orderflow client. Normally
    /// you would call `send_limit_order`, however if you want a
    /// somewhat lower level interface you can use simple_orderflow
    /// directly.
    pub fn simple_orderflow(&self) -> &simple_orderflow::OmsQueryApi {
        &self.simple_orderflow
    }

    /// Return a reference to the limits api
    pub fn limits(&self) -> &limits::OmsLimitsApi {
        &self.limits
    }

    /// Return a reference to the secrets db api
    pub fn secrets(&self) -> &secrets::SecretsQueryApi {
        &self.secrets
    }

    // CR estokes: the broadcast channel clones everything you put
    // into it for every subscriber.
    /// Return a reference to the alerts channel. You must also call
    /// `subscribe_to_alerts` if you want to receive system alerts.
    pub fn alerts(&self) -> broadcast::Receiver<Pooled<Vec<Alert>>> {
        self.alerts_tx.subscribe()
    }

    /// Subscribe to system alerts. To use the alerts call `alerts`
    /// and use the returned broadcast channel.
    pub fn subscribe_to_alerts(&self, path: Path, seek: Seek) {
        pool!(alerts_pool, Vec<Alert>, 10000, 1000);
        let common = self.common.clone();
        let tx = self.alerts_tx.clone();
        tokio::spawn(async move {
            let api = alerts::AlertsApi::new(&common, path).await?;
            let (inner_tx, mut rx) = fmpsc::channel::<Pooled<Vec<(SubId, Event)>>>(10000);
            api.last_alert().updates(UpdatesFlags::BEGIN_WITH_LAST, inner_tx);
            api.start_session(&seek).await?;
            loop {
                if let Some(updates) = rx.next().await {
                    let mut alerts = alerts_pool().take();
                    for (_, event) in updates.iter() {
                        match event {
                            Event::Unsubscribed => (),
                            Event::Update(Value::Null) => (),
                            Event::Update(value) => {
                                if let Ok(alert) = Alert::from_value(value.clone()) {
                                    alerts.push(alert);
                                }
                            }
                        }
                    }
                    if !alerts.is_empty() {
                        tx.send(alerts).unwrap();
                    }
                }
            }
            #[allow(unreachable_code)]
            Ok::<(), anyhow::Error>(())
        });
    }

    /// Return the default account
    pub fn default_account(&self) -> Account {
        self.default_account
    }

    /// Subscribe to the specified product by name. Base is the
    /// product, e.g. BTC, quote is the quote currency,
    /// e.g. USD. Venue and Route are the exchange the product trades
    /// on and qf route to use for the data. For most exchanges this
    /// will be Route::get("DIRECT").
    pub fn managed_product(
        &self,
        base: &str,
        quote: &str,
        venue: Venue,
        route: Route,
    ) -> Result<managed_product::ManagedProduct> {
        let base = Product::get(base).ok_or(anyhow!("no such product: {}", base))?;
        let quote = Product::get(quote).ok_or(anyhow!("no such product: {}", quote))?;
        let tradable_product = TradableProduct::get(base, quote, venue, route)
                .ok_or(anyhow!(
                    "no tradable product found for ({}, {}, {}, {}): is symbology up, and the cpty loaded?", 
                    base, quote, venue, route
            ))?;

        Ok(managed_product::ManagedProduct::new(&self.common, tradable_product))
    }

    /// Wrap an order in a managed order
    pub async fn managed_order(
        &self,
        order_id: OrderId,
    ) -> Arc<managed_order::ManagedOrder> {
        let mo = managed_order::ManagedOrder::new(order_id);
        let mo = Arc::new(mo);
        let mut managed_orders = self.managed_orders.write().await;
        managed_orders.insert(order_id, mo.clone());
        mo
    }

    /// Send a limit order, returning the managed order
    pub async fn send_limit_order(
        &self,
        product: TradableProduct,
        account: Account,
        dir: Dir,
        price: Decimal,
        quantity: Decimal,
    ) -> Result<Arc<managed_order::ManagedOrder>> {
        let oid = self.orderflow.orderid();
        self.orderflow.send_one(&ToOms::Order(Order {
            id: oid,
            timestamp: None,
            target: product,
            account,
            dir,
            price,
            quantity,
        }))?;
        let mo = self.managed_order(oid).await;
        Ok(mo)
    }

    /// Return a list of open orders in the specified set and
    /// scope. E.G. to return all open orders, you'd use
    /// LimitSet::Firm, LimitScope::Global.
    pub async fn list_open_orders(
        &self,
        set: LimitSet,
        scope: LimitScope,
    ) -> Result<Pooled<Vec<Order>>> {
        let oids = self.simple_orderflow.list_open(set, scope).await?;
        let orders = self.simple_orderflow.get_order_details(&oids).await?;
        Ok(orders)
    }

    /// return a list of fills in the specified set and scope.
    ///
    /// However be aware that because the OMS is designed to support
    /// very high frequency trading applications it is not possible
    /// for it to keep every order forever. Filled orders are dropped
    /// after 1 day by default.
    pub async fn list_fills(
        &self,
        set: LimitSet,
        scope: LimitScope,
    ) -> Result<Pooled<Vec<result::Result<NormalFill, AberrantFill>>>> {
        let oids_and_fillids = self.simple_orderflow.list_fills(set, scope).await?;
        let fillids: Vec<FillId> =
            oids_and_fillids.iter().map(|(_, fillid)| *fillid).collect();
        let fills =
            self.simple_orderflow.get_fill_details(&Pooled::orphan(fillids)).await?;
        Ok(fills)
    }

    fn run(&self) -> tokio::task::JoinHandle<Result<()>> {
        let t = self.clone();
        tokio::task::spawn(async move {
            while let Ok(msg) = t.orderflow.recv_one().await {
                debug!("from orderflow: {:?}", msg);
                match msg {
                    FromOms::OrderAck(Ok(oid))
                    | FromOms::Fill(Fill::Normal(Ok(NormalFill { id: oid, .. })))
                    | FromOms::Fill(Fill::Correction(Ok(NormalFill {
                        id: oid, ..
                    })))
                    | FromOms::Fill(Fill::Reversal(Ok(NormalFill { id: oid, .. })))
                    | FromOms::Out(Ok(oid))
                    | FromOms::OmsReject(OmsReject { id: oid, .. })
                    | FromOms::CptyReject(Ok(NormalCptyReject { id: oid, .. }))
                    | FromOms::NotOut(oid) => {
                        // CR estokes: update this code to handle fill Corrections and Reversals
                        // CR alee: consider another concurrent queue for updating order state
                        // CR alee: maybe move to managed_orders.rs
                        let mos = t.managed_orders.read().await;
                        if let Some(mo) = mos.get(&oid) {
                            let res = t
                                .simple_orderflow
                                .get_order_state(&Pooled::orphan(vec![oid]))
                                .await?;
                            let state = res
                                .first()
                                .ok_or(anyhow!("expected exactly one item in vec"))?;
                            mo.set_order_state(*state);
                        }
                    }
                    other => bail!("aberrant response: {:?}", other),
                }
            }
            Ok(())
        })
    }
}

/// Helper functions
pub mod utils {
    use crate::protocol::orderflow::{AberrantFill, NormalFill};
    use crate::protocol::Dir;
    use anyhow::{anyhow, Result};
    use rust_decimal::Decimal;
    use std::result;

    /// Return true if the first price is more aggressive than the
    /// second for the specified direction.
    ///
    /// e.g. is_more_agg_then(dec!(3), dec!(2), Dir::Sell) -> false
    /// e.g. is_more_agg_then(dec!(3), dec!(2), Dir::Buy) -> true
    pub fn is_more_agg_than(price: Decimal, than: Decimal, dir: Dir) -> bool {
        match dir {
            Dir::Buy => price > than,
            Dir::Sell => price < than,
        }
    }

    /// Round the decimal to the nearest multiple of increment in the
    /// direction of zero
    pub fn round_to_nearest_towards_zero(x: Decimal, increment: Decimal) -> Decimal {
        if x.is_sign_positive() || x.is_zero() {
            (x / increment).floor() * increment
        } else {
            (x / increment).ceil() * increment
        }
    }

    /// Check the list of fills and return an error if any of them are
    /// abberant, otherwise return a vec of normal fills.
    pub fn all_normal(
        fills: &Vec<result::Result<NormalFill, AberrantFill>>,
    ) -> Result<Vec<&NormalFill>> {
        let mut ret = Vec::new();
        for result in fills {
            match result {
                Ok(fill) => ret.push(fill),
                Err(_) => return Err(anyhow!("there are aberrant fills")),
            }
        }
        Ok(ret)
    }

    #[cfg(test)]
    mod test {
        use super::*;
        use rust_decimal_macros::dec;

        #[test]
        fn it_rounds_toward_zero() {
            assert_eq!(round_to_nearest_towards_zero(dec!(0), dec!(0.25)), dec!(0));
            assert_eq!(round_to_nearest_towards_zero(dec!(1.33), dec!(0.25)), dec!(1.25));
            assert_eq!(
                round_to_nearest_towards_zero(dec!(-1.33), dec!(0.25)),
                dec!(-1.25)
            );
        }
    }
}

// CR estokes: move to symbology. Think carefully about init.
/// List of known venues for convenience.
pub mod venues {
    use crate::symbology::Venue;
    use once_cell::sync::Lazy;

    pub static COINBASE: Lazy<Venue> = Lazy::new(|| Venue::get("COINBASE").unwrap());
    pub static UNISWAPV3F100: Lazy<Venue> =
        Lazy::new(|| Venue::get("UNISWAPV3F100").unwrap());
    pub static UNISWAPV3F500: Lazy<Venue> =
        Lazy::new(|| Venue::get("UNISWAPV3F500").unwrap());
    pub static B2C2: Lazy<Venue> = Lazy::new(|| Venue::get("B2C2").unwrap());
}

/// List of known routes for convenience
pub mod routes {
    use crate::symbology::Route;
    use once_cell::sync::Lazy;

    pub static DIRECT: Lazy<Route> = Lazy::new(|| Route::get("DIRECT").unwrap());
}