client.rs

//! High-level streaming client returned by [`crate::IgClient::streaming`].
//!
//! Obtain a [`StreamingApi`] from [`crate::IgClient::streaming`], then call
//! [`StreamingApi::connect`] or [`StreamingApi::connect_with`] to open a
//! Lightstreamer session and receive a [`StreamingClient`].
//!
//! # Example
//!
//! ```no_run
//! # use trading_ig::{IgClient, Environment, Credentials};
//! # async fn run() -> trading_ig::Result<()> {
//! # let client = IgClient::builder()
//! #     .environment(Environment::Demo)
//! #     .api_key("key")
//! #     .credentials(Credentials::password("u", "p"))
//! #     .build()?;
//! client.session().login_v2().await?;
//! let (stream, _events) = client.streaming().connect_with(Default::default()).await?;
//! let mut rx = stream.subscribe_price("CS.D.GBPUSD.TODAY.IP").await?;
//! while let Some(update) = rx.recv().await {
//!     println!("{} bid={:?}", update.epic, update.bid);
//! }
//! # Ok(()) }
//! ```

use std::sync::Once;

use tokio::sync::{mpsc, watch};
use tracing::{instrument, warn};

use crate::IgClient;
use crate::error::Result;
use crate::session::SessionHandle;
use crate::streaming::connection::{self, CreateParams, LsConnection, SharedConn};
use crate::streaming::events::{
    AccountUpdate, CandleScale, ChartCandleUpdate, ChartTickUpdate, MarketUpdate, PriceUpdate,
    TradeUpdate,
};
use crate::streaming::reconnect::{AutoReconnect, StreamingEvent};
use crate::streaming::subscription::{Registry, SubscriptionKind};

/// Subscription channel capacity.  Keep modest so a slow consumer causes
/// back-pressure rather than unbounded queue growth.
const CHANNEL_CAP: usize = 256;

/// Capacity of the optional lifecycle-event channel.
const EVENT_CHAN_CAP: usize = 64;

// ---------------------------------------------------------------------------
// StreamingApi — accessor on IgClient
// ---------------------------------------------------------------------------

/// Entry point for streaming.  Obtain via [`crate::IgClient::streaming`].
#[derive(Debug)]
pub struct StreamingApi<'a> {
    pub(crate) client: &'a IgClient,
}

impl StreamingApi<'_> {
    /// Connect to the Lightstreamer streaming endpoint with the default
    /// [`AutoReconnect`] policy (enabled, 5 attempts, 1 s–30 s back-off).
    ///
    /// The underlying session must already be authenticated before calling
    /// this method.  For a **v3 (OAuth)** session, call
    /// `client.session().read(true).await?` first so that CST/XST tokens
    /// are stored locally — Lightstreamer requires the
    /// `CST-<cst>|XST-<xst>` password format regardless of auth flavour.
    ///
    /// Returns `(StreamingClient, Receiver<StreamingEvent>)`.  The event
    /// channel emits [`StreamingEvent::Reconnected`],
    /// [`StreamingEvent::ReconnectFailed`], and
    /// [`StreamingEvent::Disconnected`] lifecycle events.
    ///
    /// Equivalent to `connect_with(AutoReconnect::default())`.
    #[instrument(skip_all, name = "streaming.connect")]
    pub async fn connect(&self) -> Result<(StreamingClient, mpsc::Receiver<StreamingEvent>)> {
        self.connect_with(AutoReconnect::default()).await
    }

    /// Connect with an explicit [`AutoReconnect`] policy.
    ///
    /// Set `policy.enabled = false` to disable auto-reconnect and get the
    /// pre-reconnect behaviour: the stream terminates on `END` and all
    /// subscriber channels close.
    ///
    /// Returns `(StreamingClient, Receiver<StreamingEvent>)`.
    #[instrument(skip_all, name = "streaming.connect_with")]
    pub async fn connect_with(
        &self,
        policy: AutoReconnect,
    ) -> Result<(StreamingClient, mpsc::Receiver<StreamingEvent>)> {
        let state = self.client.session.require_authenticated().await?;

        let account_id = state.account_id.ok_or_else(|| {
            crate::error::Error::Auth(
                "no account ID in session — call session().login() first".into(),
            )
        })?;
        let endpoint = state.lightstreamer_endpoint.ok_or_else(|| {
            crate::error::Error::Auth(
                "no Lightstreamer endpoint in session — call session().login() first".into(),
            )
        })?;

        // Build Lightstreamer password from the streaming CST/XST pair.
        // v2 logins populate this directly ; v3 (OAuth) sessions need
        // an explicit `client.session().read(true).await?` first to
        // populate the streaming surface without losing OAuth.
        let password = match state.tokens.streaming.as_ref() {
            Some(s) => format!("CST-{}|XST-{}", s.cst, s.x_security_token),
            None => {
                return Err(crate::error::Error::Auth(
                    "no streaming tokens (CST/XST). Call \
                     client.session().read(true).await? first to populate \
                     the streaming surface, then connect()."
                        .into(),
                ));
            }
        };

        let registry = Registry::new();
        let (shutdown_tx, _shutdown_rx) = watch::channel(false);
        let (event_tx, event_rx) = mpsc::channel(EVENT_CHAN_CAP);

        // Build a SessionHandle so the reconnect path can call login_v2().
        let session_handle = SessionHandle {
            transport: self.client.transport.clone(),
            session: self.client.session.clone(),
            credentials: self.client.credentials.clone(),
        };

        let conn = LsConnection::create(CreateParams {
            endpoint,
            username: account_id.clone(),
            password,
            registry: registry.clone(),
            shutdown_tx: shutdown_tx.clone(),
            policy,
            event_tx: Some(event_tx),
            session_handle,
        })
        .await?;

        let client = StreamingClient {
            conn,
            registry,
            shutdown_tx,
            account_id,
        };
        Ok((client, event_rx))
    }
}

// ---------------------------------------------------------------------------
// StreamingClient
// ---------------------------------------------------------------------------

/// A live Lightstreamer session with active subscriptions.
///
/// Obtained via [`StreamingApi::connect`] or [`StreamingApi::connect_with`].
/// All subscription methods return a `tokio::sync::mpsc::Receiver<T>`.
/// Dropping the receiver automatically cancels the subscription server-side
/// the next time the server sends an update for that item.
///
/// Call [`StreamingClient::disconnect`] to cleanly tear down the session.
#[derive(Debug)]
pub struct StreamingClient {
    conn: SharedConn,
    registry: Registry,
    shutdown_tx: watch::Sender<bool>,
    /// Account this Lightstreamer session authenticated as. Deliberately a
    /// connect-time snapshot: a later `switch_account` does not re-authenticate
    /// the stream, so the server still knows us as this account.
    account_id: String,
}

/// `#[deprecated]` only fires at compile time; long-running bots that already
/// call `subscribe_market` need to see the EOL in their logs too.
fn warn_market_deprecated() {
    static ONCE: Once = Once::new();
    ONCE.call_once(|| {
        warn!(
            "MARKET subscription is deprecated by IG (EOL 1 May 2026, \
             decommissioned 8 May 2026) — migrate to subscribe_price()"
        );
    });
}

impl StreamingClient {
    // ------------------------------------------------------------------
    // Price
    // ------------------------------------------------------------------

    /// Subscribe to `PRICE:<accountId>:<epic>` for the account this stream
    /// was connected with — *not* whichever account the REST session was last
    /// switched to. Replaces the deprecated
    /// [`subscribe_market`](Self::subscribe_market).
    ///
    /// Each received value is a snapshot of all changed fields merged with the
    /// previous state — no field is ever "missing".
    #[instrument(skip(self), fields(%epic))]
    pub async fn subscribe_price(&self, epic: &str) -> Result<mpsc::Receiver<PriceUpdate>> {
        self.subscribe_price_for_account(&self.account_id, epic)
            .await
    }

    /// Subscribe to `PRICE:<accountId>:<epic>` for an explicit account.
    ///
    /// Escape hatch for callers that track account ids themselves. IG may
    /// reject an account the Lightstreamer session did not authenticate as —
    /// unverified against a live session.
    #[instrument(skip(self), fields(%account_id, %epic))]
    pub async fn subscribe_price_for_account(
        &self,
        account_id: &str,
        epic: &str,
    ) -> Result<mpsc::Receiver<PriceUpdate>> {
        let (tx, rx) = mpsc::channel(CHANNEL_CAP);
        let spec = self.registry.register(SubscriptionKind::Price {
            account_id: account_id.to_owned(),
            epic: epic.to_owned(),
            tx,
        });
        connection::control(&self.conn, "add", &spec).await?;
        Ok(rx)
    }

    // ------------------------------------------------------------------
    // Market — DEPRECATED by IG
    // ------------------------------------------------------------------

    /// Subscribe to market price updates for `epic`.
    ///
    /// Returns a `Receiver<MarketUpdate>`.
    #[deprecated(
        since = "0.1.6",
        note = "IG deprecated the MARKET subscription: end of life 1 May 2026, \
                decommissioned 8 May 2026. Use `subscribe_price` instead."
    )]
    #[instrument(skip(self), fields(%epic))]
    pub async fn subscribe_market(&self, epic: &str) -> Result<mpsc::Receiver<MarketUpdate>> {
        warn_market_deprecated();
        let (tx, rx) = mpsc::channel(CHANNEL_CAP);
        let spec = self.registry.register(SubscriptionKind::Market {
            epic: epic.to_owned(),
            tx,
        });
        connection::control(&self.conn, "add", &spec).await?;
        Ok(rx)
    }

    // ------------------------------------------------------------------
    // Chart tick
    // ------------------------------------------------------------------

    /// Subscribe to chart tick data for `epic`.
    ///
    /// Returns a `Receiver<ChartTickUpdate>`.  This is a `DISTINCT`-mode
    /// subscription — every message is a fresh tick, not a merge.
    #[instrument(skip(self), fields(%epic))]
    pub async fn subscribe_chart_tick(
        &self,
        epic: &str,
    ) -> Result<mpsc::Receiver<ChartTickUpdate>> {
        let (tx, rx) = mpsc::channel(CHANNEL_CAP);
        let spec = self.registry.register(SubscriptionKind::ChartTick {
            epic: epic.to_owned(),
            tx,
        });
        connection::control(&self.conn, "add", &spec).await?;
        Ok(rx)
    }

    // ------------------------------------------------------------------
    // Chart candle
    // ------------------------------------------------------------------

    /// Subscribe to OHLC candle data for `epic` at `scale`.
    ///
    /// Returns a `Receiver<ChartCandleUpdate>`.  This is a `MERGE`-mode
    /// subscription — fields are merged across updates for the current candle.
    #[instrument(skip(self), fields(%epic, scale = %scale))]
    pub async fn subscribe_chart_candle(
        &self,
        epic: &str,
        scale: CandleScale,
    ) -> Result<mpsc::Receiver<ChartCandleUpdate>> {
        let (tx, rx) = mpsc::channel(CHANNEL_CAP);
        let spec = self.registry.register(SubscriptionKind::ChartCandle {
            epic: epic.to_owned(),
            scale,
            tx,
        });
        connection::control(&self.conn, "add", &spec).await?;
        Ok(rx)
    }

    // ------------------------------------------------------------------
    // Account
    // ------------------------------------------------------------------

    /// Subscribe to account balance and margin updates.
    ///
    /// Returns a `Receiver<AccountUpdate>`.
    #[instrument(skip(self), fields(%account_id))]
    pub async fn subscribe_account(
        &self,
        account_id: &str,
    ) -> Result<mpsc::Receiver<AccountUpdate>> {
        let (tx, rx) = mpsc::channel(CHANNEL_CAP);
        let spec = self.registry.register(SubscriptionKind::Account {
            account_id: account_id.to_owned(),
            tx,
        });
        connection::control(&self.conn, "add", &spec).await?;
        Ok(rx)
    }

    // ------------------------------------------------------------------
    // Trade
    // ------------------------------------------------------------------

    /// Subscribe to trade confirmations and working-order updates.
    ///
    /// Returns a `Receiver<TradeUpdate>`.
    #[instrument(skip(self), fields(%account_id))]
    pub async fn subscribe_trade(&self, account_id: &str) -> Result<mpsc::Receiver<TradeUpdate>> {
        let (tx, rx) = mpsc::channel(CHANNEL_CAP);
        let spec = self.registry.register(SubscriptionKind::Trade {
            account_id: account_id.to_owned(),
            tx,
        });
        connection::control(&self.conn, "add", &spec).await?;
        Ok(rx)
    }

    // ------------------------------------------------------------------
    // Lifecycle
    // ------------------------------------------------------------------

    /// Disconnect from Lightstreamer and stop the background read-loop task.
    ///
    /// After this call all pending `Receiver`s will no longer receive updates.
    ///
    /// The method signature is `async` for forward compatibility (future
    /// implementations may need to await a clean shutdown handshake with the
    /// server).
    #[allow(clippy::unused_async)]
    pub async fn disconnect(self) -> Result<()> {
        // Signal the background read-loop to stop.
        let _ = self.shutdown_tx.send(true);
        Ok(())
    }

    /// Return the current Lightstreamer session ID.
    ///
    /// Now `async` and returns an owned `String`: the connection lives behind a
    /// shared async lock (P1-15) so a reconnect-driven session swap is observed
    /// here too, rather than a stale clone.
    pub async fn session_id(&self) -> String {
        self.conn.read().await.session_id.clone()
    }
}