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

//! This is the low level orderflow protocol

use crate::{
    packed_value, pool,
    protocol::{
        b2c2, coinbase, dvchain,
        limits::{LimitName, LimitScope, LimitSet},
        Account, Desk, Dir, Trader,
    },
    symbology::{Product, Route, TradableProduct, Venue},
};
use anyhow::{anyhow, Result};
use chrono::prelude::*;
use enumflags2::{bitflags, BitFlags};
use netidx::pool::Pooled;
use netidx_derive::Pack;
use rust_decimal::Decimal;
use schemars::{JsonSchema, JsonSchema_repr};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::{
    fmt,
    str::FromStr,
    sync::{
        atomic::{AtomicU32, AtomicU64, Ordering},
        Arc,
    },
};
use uuid::Uuid;

pool!(pub, pool_to_oms, Vec<ToOms>, 1_000, 1_000);
pool!(pub, pool_from_oms, Vec<FromOms>, 1_000, 1_000);

/// The state of an order
#[bitflags]
#[repr(u8)]
#[derive(
    Debug, Clone, Copy, Hash, PartialEq, Eq, Pack, Serialize, Deserialize, JsonSchema_repr,
)]
pub enum OrderState {
    Open,
    Acked,
    Filled,
    Canceled,
    Out,
    Rejected,
    NotOut,
}

packed_value!(OrderState);

/// The id of the channel an order is sent through
// ChanId is 24 bits, this leaves enough space for 6 new channels to
// be created every second for 30 days before wrapping. Even in that
// extreme scenario, any order still using a previous channel ID will
// surely be retired by that time.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Pack)]
pub struct ChanId(u32);

static CHAN_ID: AtomicU32 = AtomicU32::new(0);

impl ChanId {
    /// Create a new channel id. This is only really used by the OMS.
    pub fn new() -> ChanId {
        const MAX_CID: u32 = 0x00FF_FFFF;
        loop {
            let cid = CHAN_ID.fetch_add(1, Ordering::Relaxed);
            if cid > MAX_CID {
                let _ = CHAN_ID.compare_exchange(
                    cid,
                    0,
                    Ordering::Relaxed,
                    Ordering::Relaxed,
                );
            } else {
                break Self(cid);
            }
        }
    }

    /// Set the max used channel id for the current process. Do not use this
    /// unless you know exactly what you are doing.  
    /// Note if the value you try to set is less than the current
    /// value then nothing will happen
    pub fn maybe_set_max(self) {
        let _ = CHAN_ID.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
            if cur <= self.0 {
                Some(self.0 + 1)
            } else {
                None
            }
        });
    }
}

packed_value!(ChanId);

impl fmt::Display for ChanId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// The id of an order
// The order id has two components. The channel id identifies which
// order flow channel the order came in from, and thus which one
// messages must go back to. It occupies the high 24 bits of the
// u64. The remaing 40 bits is the sequence number. This allows for a
// single connection to send 35k orders per second for 1 year before
// it would wrap.
#[derive(
    Debug,
    Clone,
    Copy,
    Hash,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Pack,
    Serialize,
    Deserialize,
    JsonSchema,
)]
pub struct OrderId(u64);

packed_value!(OrderId);

const SEQ_MASK: u64 = 0xFFFF_FF00_0000_0000;
const MAX_SEQ: u64 = 0x0000_00FF_FFFF_FFFF;
static OID: AtomicU64 = AtomicU64::new(0);

impl OrderId {
    /// Create a new order id for use on a particular order flow channel
    pub fn new(chan: ChanId) -> OrderId {
        let seq = OID
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |seq| {
                if seq < MAX_SEQ {
                    Some(seq + 1)
                } else {
                    Some(0)
                }
            })
            .unwrap();
        Self::from_parts(chan, seq)
    }

    fn from_parts(chan: ChanId, seq: u64) -> Self {
        Self(((chan.0 as u64) << 40) | seq)
    }

    /// Extract the channel id portion of the orderid
    pub fn chan(&self) -> ChanId {
        ChanId(((self.0 & SEQ_MASK) >> 40) as u32)
    }

    /// Extract the sequence number portion of the orderid
    pub fn seq(&self) -> u64 {
        self.0 & !SEQ_MASK
    }

    /// Set the max used order id for the process. Do not
    /// use this unless you know exactly what you are doing
    /// Note if the value you try to set is less than the current
    /// value then nothing will happen
    pub fn maybe_set_max(self) {
        let _ = OID.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
            if v <= self.seq() {
                Some(self.seq() + 1)
            } else {
                None
            }
        });
    }
}

impl fmt::Display for OrderId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}:{}", self.chan(), self.seq())
    }
}

impl FromStr for OrderId {
    type Err = anyhow::Error;

    fn from_str(s: &str) -> Result<Self> {
        let (chan, seq) =
            s.split_once(':').ok_or_else(|| anyhow!("parse error can't split {}", s))?;
        let chan = ChanId(chan.parse::<u32>()?);
        let seq = seq.parse::<u64>()?;
        Ok(Self::from_parts(chan, seq))
    }
}

/// The ID of a fill
#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    Pack,
    Serialize,
    Deserialize,
    JsonSchema,
)]
pub struct FillId(Uuid);

packed_value!(FillId);

impl Default for FillId {
    fn default() -> Self {
        FillId(Uuid::new_v4())
    }
}

/// The best information available about an aberrant thing.
///
/// An aberrant thing is a thing that Architect doesn't fully
/// understand, or that seems to violate some invariant. In the case
/// it receives such a thing from a counterparty it will try to
/// process it as far as is safe, and then it will send the aberrant
/// thing back to the client.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack, Serialize, Deserialize, JsonSchema)]
pub struct AberrantInfo {
    pub id: Option<OrderId>,
    pub base: Option<Product>,
    pub quote: Option<Product>,
    pub route: Option<Route>,
    pub venue: Option<Venue>,
    pub price: Option<Decimal>,
    pub quantity: Option<Decimal>,
    pub account: Option<Account>,
    pub trader: Option<Trader>,
    pub desk: Option<Desk>,
    pub dir: Option<Dir>,
}

packed_value!(AberrantInfo);

impl AberrantInfo {
    /// If possible, extract the tradable product from this aberrant thing
    pub fn get_tradable_product(&self) -> Option<TradableProduct> {
        let base = self.base?;
        let quote = self.quote?;
        let route = self.route?;
        let venue = self.venue?;
        TradableProduct::get(base, quote, venue, route)
    }

    fn is_for(&self, id: ChanId) -> bool {
        self.id.map(|oid| oid.chan() == id).unwrap_or(true)
    }
}

/// This represents the scope of a halt in Architect
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "type", content = "value")]
pub enum HaltSet {
    /// The entire firm
    Firm,
    /// A trading desk
    Desk(Desk),
    /// A trader
    Trader(Trader),
    /// An account
    Account(Account),
}

packed_value!(HaltSet);

#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "type", content = "value")]
pub enum HaltTerm {
    Global,
    Product(Product),
}

packed_value!(HaltTerm);

/// A halted thing can't open new orders
///
/// but it can be canceled, and if it has open orders they can be
/// filled.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "type", content = "value")]
pub enum Halt {
    /// A global halt.
    /// - Halt::Global(HaltSet::Firm, HaltTerm::Global): halt everything for the entire firm
    /// - Halt::Global(HaltSet::Trader(bob), HaltTerm::Product(btc)): stop bob from trading btc everywhere
    Global(HaltSet, HaltTerm),
    /// Halt a tradable product.
    /// - Halt::TradableProduct(HaltSet::Firm, btc/usd*coinbase/direct): halt btc trading on coinbase firm wide.
    /// - Halt::TradableProduct(HaltSet::Account(acct), eth): halt eth trading in the wallet acct.
    TradableProduct(HaltSet, TradableProduct),
    /// Halt trading to a specific venue
    Venue(HaltSet, Venue, HaltTerm),
    /// Halt trading via a specific route
    Route(HaltSet, Route, HaltTerm),
    /// Halt trading to a specific venue/route pair
    VenueRoute(HaltSet, Venue, Route, HaltTerm),
}

packed_value!(Halt);

/// An order! Limit only for the moment.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack, Serialize, Deserialize, JsonSchema)]
pub struct Order {
    pub id: OrderId,
    /// Stamped on the oms receive
    pub timestamp: Option<DateTime<Utc>>,
    pub target: TradableProduct,
    pub account: Account,
    pub dir: Dir,
    pub price: Decimal,
    pub quantity: Decimal,
}

packed_value!(Order);

// not great. not terrible.
#[bitflags]
#[repr(u64)]
#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq, JsonSchema_repr)]
pub enum FillWarning {
    FillAfterOut,
}

/// A fill that was fully understood and processed normally.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack, Serialize, Deserialize, JsonSchema)]
pub struct NormalFill {
    /// Bad things about the fill that don't quite make it aberrant
    pub warnings: BitFlags<FillWarning>,
    pub id: OrderId,
    pub fill_id: FillId,
    pub target: TradableProduct,
    pub quantity: Decimal,
    pub price: Decimal,
    pub dir: Dir,
    /// When the oms received the fill
    pub recv_time: DateTime<Utc>,
    /// When the counterparty claims the trade happened
    pub trade_time: DateTime<Utc>,
}

packed_value!(NormalFill);

impl NormalFill {
    /// Turn a normal fill into an aberrant fill
    pub fn as_aberrant(&self, problems: BitFlags<FillProblem>) -> AberrantFill {
        AberrantFill {
            fill_id: self.fill_id,
            recv_time: self.recv_time,
            trade_time: self.trade_time,
            warnings: self.warnings,
            problems,
            info: AberrantInfo {
                id: Some(self.id),
                base: Some(self.target.base),
                quote: Some(self.target.quote),
                route: Some(self.target.route),
                venue: Some(self.target.venue),
                price: Some(self.price),
                quantity: Some(self.quantity),
                account: None,
                trader: None,
                desk: None,
                dir: Some(self.dir),
            },
        }
    }
}

/// The list of problems that make a fill aberrant
#[bitflags]
#[repr(u64)]
#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq, JsonSchema_repr)]
pub enum FillProblem {
    /// We received a fill for an order we don't know
    FillForUnknownOrder, // either
    /// We received fills totaling to a larger quantity than the order
    Overfilled, // oms
    /// We received a fill for a different symbol than the order was sent for
    FillForTheWrongInstrument, // oms
    /// We received a fill that violates the limit price on the order
    FillForTheWrongPrice, // oms
    /// unused
    MaxPriceVariation, // oms (unused)
    /// We received a fill for a tradable product we don't know
    FillForUnknownThing, // cpty
    /// We received a fill for a different raw symbol than the one we sent
    FillForTheWrongTradingSymbol, // cpty
    /// We received a fill for a crazy looking price (e.g. 0)
    FillForAbnormalPrice, // cpty
    /// We received a fill for a crazy looking quantity
    FillForAbnormalQuantity, // cpty
    /// We received a fill with no quantity
    FillForUnknownQuantity, // cpty
    /// We received a fill with no price
    FillForUnknownPrice, // cpty
    /// We received a fill that looks like a duplicate of a previous fill
    DuplicateFill, // cpty
    /// We received a fill in the wrong direction, e.g. the order was
    /// to buy and the fill was to sell
    FillForTheWrongDir, // cpty
    /// We received a fill with no direction
    FillForUnknownDir, // cpty
    /// We received a reversal or restatement that doesn't matche the
    /// original fill
    ReversalDiffersFromOriginal, // orderdb
    /// We received a reversal or correction for a fill we never received
    ReversalOrCorrectionForUnknownFill, // orderdb
    /// We don't even know what went wrong. This is mostly for
    /// protocol compatibility enhancement.
    UnknownFillProblem, // either, protocol compat
}

/// A fill that we couldn't process normally
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack, Serialize, Deserialize, JsonSchema)]
pub struct AberrantFill {
    pub warnings: BitFlags<FillWarning>,
    /// The reasons why this fill couldn't be processed normally
    pub problems: BitFlags<FillProblem>,
    pub fill_id: FillId,
    /// The time we received the fill
    pub recv_time: DateTime<Utc>,
    /// The time the counterparty claims the trade happened, if it was
    /// recoverable, otherwise the time we received the fill.
    pub trade_time: DateTime<Utc>,
    /// The information we could salvage
    pub info: AberrantInfo,
}

packed_value!(AberrantFill);

/// A fill
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack)]
pub enum Fill {
    /// A regular fill, either aberrant or not
    Normal(Result<NormalFill, AberrantFill>),
    /// A correction of a previous fill
    Correction(Result<NormalFill, AberrantFill>),
    /// A trade break
    Reversal(Result<NormalFill, AberrantFill>),
}

packed_value!(Fill);

/// A message to the OMS
#[derive(Debug, Clone, PartialEq, Eq, Pack)]
pub enum ToOms {
    Order(Order),
    Cancel(OrderId),
    Halt(Halt),
    Resume(Halt),
}

packed_value!(ToOms);

/// The list of problems that make an order ack aberrant
#[bitflags]
#[repr(u64)]
#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
pub enum OrderAckProblem {
    /// We received an ack for an order we didn't send
    AckForUnknownOrder,
    /// We received more than one ack for an order. Depending on the
    /// counterparty this may be ok, we will only generate an aberrant
    /// ack if we know it isn't ok.
    DuplicateAck,
    /// We received an ack for a filled order
    AckForFilledOrder,
    /// We received an ack for an outed order
    AckForOutedOrder,
    /// We received an ack, but we can't fully parse it
    CantParseAckResponse,
    /// We don't even know what went wrong. This is mostly for
    /// protocol compatibility enhancement.
    UnknownAckProblem,
}

/// An ack that we couldn't fully process
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack)]
pub struct AberrantAck {
    /// The reason(s) why we couldn't process this ack
    pub problems: BitFlags<OrderAckProblem>,
    pub info: AberrantInfo,
}

packed_value!(AberrantAck);

/// The reasons why the oms can reject an order
#[derive(Debug, Clone, PartialEq, Eq, Pack)]
pub enum OmsRejectReason {
    /// The order is a duplicate (the id is the same as a previous order)
    DuplicateOrder,
    /// The cancel was for an order we don't know
    CancelForUnknownOrOutedOrder,
    /// The order can't be sent because it violates a limit
    LimitViolated(LimitName, LimitSet, LimitScope), // which limit violated in which set and scope
    /// The order can't be sent because one or more halts apply to it
    Halted(Arc<Pooled<Vec<Halt>>>), // reasons why it's halted
    /// The system is misconfigured and can't send orders to this
    /// venue/route pair. There are multiple counterparties claiming
    /// to handle this order's venue/route pair
    RouteConflict,
    /// There is no counterparty that handles the order's venue/route pair
    NoRouteVenuePairForInstrument,
    /// The order tradable product is not tradable, e.g. it has no trading symbol
    NonTradableInstrument,
    /// The tradable product this order is for was removed from the system
    TradableProductWasRemoved,
    /// The tradable product this order is for is the
    /// INVALID_TRADABLE_PRODUCT placeholder. This likely means the
    /// core and the client's symbology are out of sync.
    InvalidTradableProduct,
    /// The order has a non positive price
    NonPositivePrice,
    /// The order has a non positive quantity
    NonPositiveQuantity,
    /// The order price varies excessively from the limit dollar price
    MaxPriceVariation,
}

packed_value!(OmsRejectReason);

/// A reject from the oms
#[derive(Debug, Clone, PartialEq, Eq, Pack)]
pub struct OmsReject {
    pub id: OrderId,
    pub reason: OmsRejectReason,
}

packed_value!(OmsReject);

/// A reject from the counterparty or exchange
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack)]
pub enum CptyRejectReason {
    /// The order would cause the rate limit for this counterparty to
    /// be exceeded, try again later.
    RateLimitExceeded,
    /// There was an IO error while trying to send the order
    IoError,
    /// Coinbase rejected the order
    CoinbaseReject {
        new_order_failure_reason: coinbase::NewOrderFailureReason,
        preview_failure_reason: coinbase::PreviewFailureReason,
    },
    /// B2C2 rejected the order
    B2C2Reject(b2c2::NewOrderFailureReason),
    /// DVChain rejected the order
    DVChainReject(dvchain::NewOrderFailureReason),
    /// The order could not be constructed by the counterparty
    FailedToConstructOrder,
    /// For protocol compatibility
    UnknownReason,
}

packed_value!(CptyRejectReason);

/// A reject from the counterparty
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack)]
pub struct NormalCptyReject {
    pub id: OrderId,
    pub reason: CptyRejectReason,
}

packed_value!(NormalCptyReject);

/// A counterparty reject that we didn't fully understand
#[derive(Debug, Clone, Copy, PartialEq, Eq, Pack)]
pub struct AberrantCptyReject {
    pub reason: CptyRejectReason,
    pub info: AberrantInfo,
}

packed_value!(AberrantCptyReject);

/// A message from the OMS
#[derive(Debug, Clone, PartialEq, Eq, Pack)]
pub enum FromOms {
    OrderAck(Result<OrderId, AberrantAck>),
    Fill(Fill),
    Out(Result<OrderId, AberrantInfo>),
    OmsReject(OmsReject),
    CptyReject(Result<NormalCptyReject, AberrantCptyReject>),
    /// We tried to cancel the order but we have not received an out
    NotOut(OrderId),
}

packed_value!(FromOms);

impl FromOms {
    pub fn is_for(&self, id: ChanId) -> bool {
        match self {
            FromOms::OrderAck(r) => match r {
                Ok(a) => a.chan() == id,
                Err(a) => a.info.is_for(id),
            },
            FromOms::Fill(r) => match r {
                Fill::Normal(r) => match r {
                    Ok(f) => f.id.chan() == id,
                    Err(f) => f.info.is_for(id),
                },
                Fill::Correction(r) => match r {
                    Ok(f) => f.id.chan() == id,
                    Err(f) => f.info.is_for(id),
                },
                Fill::Reversal(r) => match r {
                    Ok(f) => f.id.chan() == id,
                    Err(f) => f.info.is_for(id),
                },
            },
            FromOms::Out(r) => match r {
                Ok(a) => a.chan() == id,
                Err(a) => a.is_for(id),
            },
            FromOms::OmsReject(r) => r.id.chan() == id,
            FromOms::CptyReject(r) => match r {
                Ok(r) => r.id.chan() == id,
                Err(a) => a.info.is_for(id),
            },
            FromOms::NotOut(i) => i.chan() == id,
        }
    }
}