position_events.rs

//! `PositionEventSink` — единая точка, через которую все биржи сообщают
//! о событиях с позициями: position state updates, funding settlements,
//! и внешние закрытия.
//!
//! Зачем. До этого каждая биржа независимо детектировала внешнее закрытие
//! ноги (SL fired server-side, liquidation, manual close) и независимо
//! слала `ExternalLegClose` в shared mpsc-канал. Логика дублировалась на
//! 4 сайта (2 на lighter, 2 на extended, zero1 не детектил вообще — из-за
//! этого был инцидент 2026-04-13 с LIT).
//!
//! Теперь детекция специфичных причин остаётся near-raw (каждая биржа
//! знает свой формат лучше всех), но dispatch, in-flight check и
//! construction of `ExternalLegClose` централизованы в этом фасаде.
//! Новая биржа получает единственный контракт: на каждое внешнее
//! закрытие ноги вызови `sink.report_external_close(...)` с лучшим
//! reason, который умеешь. Остальное — автомат.

use std::sync::Arc;

use dashmap::DashMap;
use rust_decimal::Decimal;
use tokio::sync::mpsc;

use crate::types::{
    ActiveOrder, ExchangeId, ExchangePositionState, ExternalCloseReason,
    ExternalLegClose, OrderPurpose, SymbolId,
};

pub struct PositionEventSink {
    positions: Arc<DashMap<(i64, u32), ExchangePositionState>>,
    active_orders: Arc<DashMap<u64, ActiveOrder>>,
    leg_closed_tx: mpsc::UnboundedSender<ExternalLegClose>,
    symbol_names: Arc<Vec<String>>,
}

impl PositionEventSink {
    pub fn new(
        positions: Arc<DashMap<(i64, u32), ExchangePositionState>>,
        active_orders: Arc<DashMap<u64, ActiveOrder>>,
        leg_closed_tx: mpsc::UnboundedSender<ExternalLegClose>,
        symbol_names: Arc<Vec<String>>,
    ) -> Self {
        Self { positions, active_orders, leg_closed_tx, symbol_names }
    }

    /// Biriga detected that one of our legs was closed externally.
    /// Performs the in-flight check (our own close order still in flight?)
    /// and — if genuinely external — dispatches `ExternalLegClose` to the
    /// executor so it symmetrically closes the surviving leg.
    ///
    /// `source_order_id` carries the client_order_id of the SL order that
    /// fired (lighter/extended SL path) or 0 when unknown (liquidation,
    /// manual close, zero1 diagnostic path).
    pub fn report_external_close(
        &self,
        exchange: ExchangeId,
        symbol_id: SymbolId,
        reason: ExternalCloseReason,
        filled_qty: Decimal,
        fill_price: Decimal,
        source_order_id: u64,
    ) {
        if self.has_in_flight_close(exchange, symbol_id) {
            // `info` (not `debug`) during the rollout window so we can
            // eyeball how often the suppression fires in production —
            // zero1 taker fills on our own IOCs trip this path a lot,
            // and we want to confirm it's behaving before dropping the
            // level back to `debug`.
            tracing::info!(
                exchange = exchange.name(),
                symbol_id,
                reason = ?reason,
                "Ignoring reported external close — our order is in flight"
            );
            return;
        }

        let symbol_name = self
            .symbol_names
            .get(symbol_id as usize)
            .cloned()
            .unwrap_or_default();

        let event = ExternalLegClose {
            symbol_id,
            exchange,
            reason,
            client_order_id: source_order_id,
            filled_qty,
            fill_price,
            symbol_name,
        };

        if self.leg_closed_tx.send(event).is_err() {
            tracing::error!(
                exchange = exchange.name(),
                symbol_id,
                reason = ?reason,
                "leg_closed_tx channel dropped — external close NOT dispatched"
            );
        }
    }

    /// Update/insert position state from a WS position frame. Pure proxy:
    /// doesn't run any transition detection — detection is an event-level
    /// concern and lives near each exchange's WS handler, which calls
    /// `report_external_close` directly when it sees the triggering event.
    ///
    /// `cum_funding` is intentionally NOT written here — it flows through
    /// `update_funding` instead, to avoid racing the WS position update
    /// (which doesn't always carry funding) against the funding-settlement
    /// path.
    #[allow(clippy::too_many_arguments)]
    pub fn update_position(
        &self,
        exchange: ExchangeId,
        symbol_id: SymbolId,
        liq_price: Decimal,
        unrealized_pnl: Decimal,
        avg_entry_price: Decimal,
        margin: Decimal,
        now_ms: u64,
    ) {
        let key = (exchange.db_id(), symbol_id);
        self.positions
            .entry(key)
            .and_modify(|eps| {
                eps.liq_price = liq_price;
                eps.unrealized_pnl = unrealized_pnl;
                eps.avg_entry_price = avg_entry_price;
                eps.margin = margin;
                eps.updated_at_ms = now_ms;
                // cum_funding intentionally NOT touched
            })
            .or_insert(ExchangePositionState {
                liq_price,
                unrealized_pnl,
                avg_entry_price,
                cum_funding: Decimal::ZERO,
                margin,
                updated_at_ms: now_ms,
            });
    }

    /// Funding settlement arrived via the WS funding-histories path
    /// (lighter) or REST polling (extended). Accumulates into cum_funding
    /// on the existing position entry, or is a no-op when the position
    /// has already been cleared.
    pub fn add_funding(&self, exchange: ExchangeId, symbol_id: SymbolId, change: Decimal) -> bool {
        let key = (exchange.db_id(), symbol_id);
        if let Some(mut eps) = self.positions.get_mut(&key) {
            eps.cum_funding += change;
            true
        } else {
            false
        }
    }

    /// Read-only accessor for code paths that still touch `exchange_positions`
    /// directly (executor's SL monitor, dashboard snapshot builder, REST
    /// reconcile). The sink owns the map so eventually these can move here
    /// or take a `&PositionEventSink` instead of the raw `Arc<DashMap>`.
    pub fn positions(&self) -> &Arc<DashMap<(i64, u32), ExchangePositionState>> {
        &self.positions
    }

    /// Returns true if the bot has any taker-producing order in flight for
    /// `(exchange, symbol)` — used to suppress false external-close dispatches
    /// caused by our own fills arriving via the biriga's secondary WS-fill
    /// notification.
    ///
    /// White-list exception: `StopLoss1`/`StopLoss2` are explicitly NOT
    /// suppressed. An SL trigger firing is exactly the external event we
    /// want to catch — the biriga's engine executes it, not us.
    ///
    /// Every other purpose (`Entry1`/`Entry2`/`Close1`/`Close2`/`ExitUnwind1`/
    /// `ExitUnwind2`) represents an order we decided to place and whose fill
    /// must NOT be confused for an external close. This matters for zero1's
    /// ws-fill detection path in particular: the account WS delivers a
    /// secondary fill notification for every IOC order, and without this
    /// suppression a fresh entry fill would race with `active_orders`
    /// cleanup and trigger a self-close of the position the bot just
    /// opened (audit 2026-04-14 finding #1).
    fn has_in_flight_close(&self, exchange: ExchangeId, symbol_id: SymbolId) -> bool {
        self.active_orders.iter().any(|entry| {
            let ao = entry.value();
            ao.exchange == exchange
                && ao.symbol_id == symbol_id
                && !matches!(ao.purpose, OrderPurpose::StopLoss1 | OrderPurpose::StopLoss2)
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{
        ActiveOrder, ExchangeId, OrderPurpose, OrderType,
    };
    use rust_decimal_macros::dec;
    use uuid::Uuid;

    fn init_exchange_ids() {
        crate::types::ExchangeId::init_names([
            (1i64, "lighter".to_string()),
            (2i64, "extended".to_string()),
            (3i64, "zero1".to_string()),
        ]);
    }

    fn fresh_sink() -> (PositionEventSink, mpsc::UnboundedReceiver<ExternalLegClose>) {
        let positions = Arc::new(DashMap::new());
        let active_orders = Arc::new(DashMap::new());
        let (tx, rx) = mpsc::unbounded_channel();
        let names = Arc::new(vec!["BTC".to_string(), "ETH".to_string(), "LIT".to_string()]);
        (
            PositionEventSink::new(positions, active_orders, tx, names),
            rx,
        )
    }

    fn lighter() -> ExchangeId { ExchangeId::from_db_id(1) }
    fn extended() -> ExchangeId { ExchangeId::from_db_id(2) }

    fn active_order(
        order_id: u64,
        trade_id: Uuid,
        exchange: ExchangeId,
        symbol_id: SymbolId,
        purpose: OrderPurpose,
    ) -> ActiveOrder {
        ActiveOrder {
            order_id,
            session_seq: 0,
            symbol_id,
            exchange,
            side: crate::types::Side::Sell,
            qty: dec!(1),
            limit_price: dec!(100),
            placed_at_ns: 0,
            exchange_ts_ns: 0,
            recv_ts_ns: 0,
            purpose,
            order_type: OrderType::Ioc,
            trade_id,
            signal_price: dec!(100),
            trigger_price: dec!(0),
            liq_price_at_placement: dec!(0),
            exchange_order_id: 0,
        }
    }

    #[test]
    fn report_external_close_dispatches_event() {
        init_exchange_ids();
        let (sink, mut rx) = fresh_sink();
        sink.report_external_close(
            lighter(),
            2, // LIT
            ExternalCloseReason::ExternalManual,
            dec!(13.89),
            dec!(1.0667),
            0,
        );
        let event = rx.try_recv().expect("event should be dispatched");
        assert_eq!(event.symbol_id, 2);
        assert_eq!(event.exchange, lighter());
        assert!(matches!(event.reason, ExternalCloseReason::ExternalManual));
        assert_eq!(event.filled_qty, dec!(13.89));
        assert_eq!(event.symbol_name, "LIT");
    }

    #[test]
    fn report_external_close_preserves_stoploss_reason() {
        init_exchange_ids();
        let (sink, mut rx) = fresh_sink();
        sink.report_external_close(
            extended(),
            0,
            ExternalCloseReason::StopLoss,
            dec!(5),
            dec!(50000),
            123,
        );
        let event = rx.try_recv().expect("event");
        assert!(matches!(event.reason, ExternalCloseReason::StopLoss));
        assert_eq!(event.client_order_id, 123);
    }

    #[test]
    fn in_flight_close_blocks_dispatch() {
        init_exchange_ids();
        let (sink, mut rx) = fresh_sink();
        let trade_id = Uuid::new_v4();
        sink.active_orders.insert(
            42,
            active_order(42, trade_id, lighter(), 2, OrderPurpose::Close1),
        );
        sink.report_external_close(
            lighter(),
            2,
            ExternalCloseReason::ExternalManual,
            dec!(13.89),
            dec!(1.0),
            0,
        );
        assert!(rx.try_recv().is_err(), "must NOT dispatch when our Close1 is in flight");
    }

    #[test]
    fn in_flight_entry_blocks_dispatch() {
        // Regression guard for audit finding #1: when a zero1 IOC entry
        // fills, the account WS delivers a secondary fill notification
        // that used to race with active_orders cleanup and trigger a
        // self-close of the freshly opened position. The sink must now
        // suppress dispatch while ANY non-StopLoss active order exists.
        init_exchange_ids();
        let (sink, mut rx) = fresh_sink();
        let trade_id = Uuid::new_v4();
        sink.active_orders.insert(
            11,
            active_order(11, trade_id, lighter(), 2, OrderPurpose::Entry1),
        );
        sink.report_external_close(
            lighter(),
            2,
            ExternalCloseReason::ExternalManual,
            dec!(13.89),
            dec!(1.0667),
            0,
        );
        assert!(
            rx.try_recv().is_err(),
            "must NOT dispatch while an Entry1 for the same (exchange, symbol) is tracked"
        );

        // Symmetric: Entry2 for exchange_2 of a pair.
        sink.active_orders.insert(
            12,
            active_order(12, trade_id, extended(), 1, OrderPurpose::Entry2),
        );
        sink.report_external_close(
            extended(),
            1,
            ExternalCloseReason::ExternalManual,
            dec!(5),
            dec!(100),
            0,
        );
        assert!(rx.try_recv().is_err(), "must suppress Entry2 case too");
    }

    #[test]
    fn in_flight_unwind_blocks_dispatch() {
        init_exchange_ids();
        let (sink, mut rx) = fresh_sink();
        let trade_id = Uuid::new_v4();
        sink.active_orders.insert(
            7,
            active_order(7, trade_id, extended(), 1, OrderPurpose::ExitUnwind2),
        );
        sink.report_external_close(
            extended(),
            1,
            ExternalCloseReason::Liquidation,
            dec!(1),
            dec!(3000),
            0,
        );
        assert!(rx.try_recv().is_err());
    }

    #[test]
    fn stoploss_order_in_flight_does_not_block_dispatch() {
        // An active SL order is a server-side trigger — when it fires, that's
        // exactly the external event we want to catch. Must NOT suppress.
        init_exchange_ids();
        let (sink, mut rx) = fresh_sink();
        let trade_id = Uuid::new_v4();
        sink.active_orders.insert(
            9,
            active_order(9, trade_id, lighter(), 2, OrderPurpose::StopLoss1),
        );
        sink.report_external_close(
            lighter(),
            2,
            ExternalCloseReason::StopLoss,
            dec!(13.89),
            dec!(0.72),
            9,
        );
        let event = rx.try_recv().expect("SL fire must dispatch");
        assert!(matches!(event.reason, ExternalCloseReason::StopLoss));
    }

    #[test]
    fn suppression_covers_every_order_purpose() {
        // Exhaustiveness guard: iterate every OrderPurpose variant and
        // assert that only StopLoss1/StopLoss2 allow dispatch. If a new
        // purpose is added to the enum, the `match` inside ensures every
        // variant is considered, and this test pins the intended policy.
        use crate::types::OrderPurpose::*;
        init_exchange_ids();

        let purposes = [
            Entry1, Entry2, Close1, Close2, ExitUnwind1, ExitUnwind2, StopLoss1, StopLoss2,
        ];
        for purpose in purposes {
            let allow_dispatch = match purpose {
                // SL fire on the biriga IS the external event we want to catch.
                StopLoss1 | StopLoss2 => true,
                // Any of our own taker-producing orders → our fill, not external.
                Entry1 | Entry2 | Close1 | Close2 | ExitUnwind1 | ExitUnwind2 => false,
            };

            let (sink, mut rx) = fresh_sink();
            let trade_id = Uuid::new_v4();
            sink.active_orders.insert(
                1,
                active_order(1, trade_id, lighter(), 2, purpose),
            );
            sink.report_external_close(
                lighter(),
                2,
                ExternalCloseReason::ExternalManual,
                dec!(1),
                dec!(1),
                0,
            );
            let got = rx.try_recv().is_ok();
            assert_eq!(
                got, allow_dispatch,
                "purpose {:?}: expected dispatch={} got dispatch={}",
                purpose, allow_dispatch, got
            );
        }
    }

    #[test]
    fn in_flight_close_on_different_exchange_does_not_block() {
        // Close for lighter doesn't protect an external close on extended.
        init_exchange_ids();
        let (sink, mut rx) = fresh_sink();
        let trade_id = Uuid::new_v4();
        sink.active_orders.insert(
            1,
            active_order(1, trade_id, lighter(), 2, OrderPurpose::Close1),
        );
        sink.report_external_close(
            extended(),
            2,
            ExternalCloseReason::ExternalManual,
            dec!(13.89),
            dec!(1.0),
            0,
        );
        let _ = rx.try_recv().expect("dispatch on the other exchange");
    }

    #[test]
    fn update_position_is_proxy_and_does_not_dispatch() {
        init_exchange_ids();
        let (sink, mut rx) = fresh_sink();
        sink.update_position(lighter(), 2, dec!(0.5), dec!(0), dec!(1.0), dec!(10), 1234);
        assert!(rx.try_recv().is_err());
        let key = (lighter().db_id(), 2u32);
        let entry = sink.positions.get(&key).unwrap();
        assert_eq!(entry.liq_price, dec!(0.5));
        assert_eq!(entry.avg_entry_price, dec!(1.0));
        assert_eq!(entry.margin, dec!(10));
        assert_eq!(entry.cum_funding, dec!(0));
    }

    #[test]
    fn update_position_preserves_cum_funding_on_re_update() {
        init_exchange_ids();
        let (sink, _rx) = fresh_sink();
        let key = (lighter().db_id(), 2u32);
        sink.update_position(lighter(), 2, dec!(0.5), dec!(0), dec!(1.0), dec!(10), 1);
        assert!(sink.add_funding(lighter(), 2, dec!(0.0025)));
        sink.update_position(lighter(), 2, dec!(0.6), dec!(1), dec!(1.05), dec!(11), 2);
        let entry = sink.positions.get(&key).unwrap();
        assert_eq!(entry.cum_funding, dec!(0.0025), "cum_funding must survive position update");
        assert_eq!(entry.liq_price, dec!(0.6));
        assert_eq!(entry.margin, dec!(11));
    }

    #[test]
    fn add_funding_is_no_op_when_position_missing() {
        init_exchange_ids();
        let (sink, _rx) = fresh_sink();
        assert!(!sink.add_funding(lighter(), 99, dec!(0.001)));
    }
}