lasersell-sdk 1.1.0

//! Low-level stream websocket client and outbound command sender.
//!
//! The client handles initial handshake, configure flow, and automatic
//! reconnects while keeping an in-memory queue of outbound messages.

use std::collections::VecDeque;
use std::time::Duration;

use futures_util::{SinkExt, Stream, StreamExt};
use secrecy::{ExposeSecret, SecretString};
use thiserror::Error;
use tokio::sync::{mpsc, oneshot};
use tokio_tungstenite::{connect_async, connect_async_tls_with_config, Connector};
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::http::header::InvalidHeaderValue;
use tokio_tungstenite::tungstenite::{Error as WsError, Message};

use tracing::{debug, info, trace, warn};

use crate::exit_api::ExitApiClient;
use crate::stream::proto::{ClientMessage, MirrorConfigMsg, ServerMessage, StrategyConfigMsg, TakeProfitLevelMsg, WatchWalletEntryMsg};

const MIN_RECONNECT_BACKOFF: Duration = Duration::from_millis(100);
const MAX_RECONNECT_BACKOFF: Duration = Duration::from_secs(2);
const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
/// Production websocket endpoint for the stream service.
pub const STREAM_ENDPOINT: &str = "wss://stream.lasersell.io/v1/ws";
/// Local development websocket endpoint for the stream service.
pub const LOCAL_STREAM_ENDPOINT: &str = "ws://localhost:8082/v1/ws";

/// Entry point for creating stream connections.
#[derive(Clone)]
pub struct StreamClient {
    api_key: SecretString,
    local: bool,
    endpoint_override: Option<String>,
    spki_pins: Option<Vec<String>>,
}

impl StreamClient {
    /// Creates a stream client for production mode.
    pub fn new(api_key: SecretString) -> Self {
        Self {
            api_key,
            local: false,
            endpoint_override: None,
            spki_pins: None,
        }
    }

    /// Enables or disables local mode endpoint routing.
    pub fn with_local_mode(mut self, local: bool) -> Self {
        self.local = local;
        self
    }

    /// Sets an explicit stream endpoint override.
    ///
    /// The override takes precedence over local mode when set.
    pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
        let endpoint = endpoint.into();
        self.endpoint_override = Some(endpoint.trim_end().to_string());
        self
    }

    /// Enables TLS certificate pinning for the WebSocket connection.
    ///
    /// `spki_sha256_b64` are base64-encoded SHA-256 hashes of intermediate CA
    /// SubjectPublicKeyInfo (SPKI) fields. At least one certificate in the
    /// server's chain must match for the connection to succeed.
    pub fn with_spki_pins(mut self, spki_sha256_b64: Vec<String>) -> Self {
        self.spki_pins = Some(spki_sha256_b64);
        self
    }

    /// Opens a configured stream connection.
    ///
    /// This spawns a background worker that owns the websocket and returns a
    /// handle pair for sending client messages and receiving server messages.
    pub async fn connect(
        &self,
        configure: StreamConfigure,
    ) -> Result<StreamConnection, StreamClientError> {
        validate_strategy_thresholds(&configure.strategy, configure.deadline_timeout_sec)?;

        let (outbound_tx, outbound_rx) = mpsc::unbounded_channel();
        let (inbound_tx, inbound_rx) = mpsc::unbounded_channel();
        let (status_tx, status_rx) = mpsc::unbounded_channel();
        let (ready_tx, ready_rx) = oneshot::channel();

        let url = self.endpoint().to_string();
        let api_key = self.api_key.clone();
        let tls_connector = self.spki_pins.as_ref().map(|pins| {
            let config = super::tls::build_pinned_tls_config(pins);
            Connector::Rustls(std::sync::Arc::new(config))
        });

        tokio::spawn(async move {
            stream_connection_worker(
                url,
                api_key,
                configure,
                outbound_rx,
                inbound_tx,
                status_tx,
                ready_tx,
                tls_connector,
            )
            .await;
        });

        match ready_rx.await {
            Ok(Ok(())) => Ok(StreamConnection {
                sender: StreamSender { tx: outbound_tx },
                receiver: inbound_rx,
                status: Some(status_rx),
            }),
            Ok(Err(err)) => Err(err),
            Err(_) => Err(StreamClientError::Protocol(
                "stream worker stopped before initial connect".to_string(),
            )),
        }
    }

    /// Registers wallet proofs via the LaserSell API, then opens a stream
    /// connection.
    ///
    /// This is the recommended way to connect. Generate proofs with
    /// [`prove_ownership`](crate::exit_api::prove_ownership) first — a pure
    /// local operation that never touches the network.
    ///
    /// ```rust,no_run
    /// # use lasersell_sdk::{prove_ownership, stream::client::*};
    /// # use secrecy::SecretString;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let proof = prove_ownership(&wallet_keypair);
    /// let client = StreamClient::new(SecretString::from("your-api-key"));
    /// let conn = client.connect_with_wallets(
    ///     &[proof],
    ///     strategy_config_from_optional(Some(50.0), Some(25.0), None, None),
    ///     45,
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect_with_wallets(
        &self,
        proofs: &[crate::exit_api::WalletProof],
        strategy: StrategyConfigMsg,
        deadline_timeout_sec: u64,
    ) -> Result<StreamConnection, StreamClientError> {
        let exit_client = ExitApiClient::with_api_key(self.api_key.clone())
            .map_err(|e| StreamClientError::Protocol(format!("failed to create exit client: {e}")))?;

        for proof in proofs {
            exit_client
                .register_wallet(proof, None)
                .await
                .map_err(|e| StreamClientError::Protocol(format!("wallet registration failed: {e}")))?;
        }

        let wallet_pubkeys = proofs.iter().map(|p| p.wallet_pubkey.clone()).collect();

        self.connect(StreamConfigure {
            wallet_pubkeys,
            strategy,
            deadline_timeout_sec,
            send_mode: None,
            tip_lamports: None,
            watch_wallets: Vec::new(),
            mirror_config: None,
        })
        .await
    }

    fn endpoint(&self) -> &str {
        if let Some(endpoint) = self.endpoint_override.as_deref() {
            return endpoint;
        }
        if self.local {
            LOCAL_STREAM_ENDPOINT
        } else {
            STREAM_ENDPOINT
        }
    }
}

/// Stream configuration sent during initial websocket setup.
///
/// Use [`StreamConfigure::single_wallet`] or [`StreamConfigure::single_wallet_optional`]
/// to construct. Mirror trading fields default to disabled.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct StreamConfigure {
    /// Wallet public keys to track.
    pub wallet_pubkeys: Vec<String>,
    /// Strategy parameters evaluated server-side.
    pub strategy: StrategyConfigMsg,
    /// SDK-local deadline timer configuration in seconds.
    ///
    /// This is enforced by `StreamSession` and is not sent to the stream
    /// server.
    pub deadline_timeout_sec: u64,
    /// How the client will submit the signed transaction (e.g. "helius_sender", "rpc", "astralane").
    pub send_mode: Option<String>,
    /// Priority fee tip in lamports (required for some send modes).
    pub tip_lamports: Option<u64>,
    /// External wallets to monitor for copy trading.
    pub watch_wallets: Vec<WatchWalletEntryMsg>,
    /// Mirror trading hardening configuration.
    pub mirror_config: Option<MirrorConfigMsg>,
}

impl StreamConfigure {
    /// Constructor for one or more wallets with a strategy.
    ///
    /// Mirror trading fields default to disabled. Set them on the returned
    /// struct if needed.
    pub fn new(wallet_pubkeys: Vec<String>, strategy: StrategyConfigMsg) -> Self {
        Self {
            wallet_pubkeys,
            strategy,
            deadline_timeout_sec: 0,
            send_mode: None,
            tip_lamports: None,
            watch_wallets: Vec::new(),
            mirror_config: None,
        }
    }

    /// Convenience constructor for a single wallet.
    pub fn single_wallet(wallet_pubkey: impl Into<String>, strategy: StrategyConfigMsg) -> Self {
        Self {
            wallet_pubkeys: vec![wallet_pubkey.into()],
            strategy,
            deadline_timeout_sec: 0,
            send_mode: None,
            tip_lamports: None,
            watch_wallets: Vec::new(),
            mirror_config: None,
        }
    }

    /// Convenience constructor for a single wallet with optional thresholds.
    ///
    /// Unset strategy values default to `0.0` (disabled), and unset deadline
    /// defaults to `0` (disabled).
    pub fn single_wallet_optional(
        wallet_pubkey: impl Into<String>,
        target_profit_pct: Option<f64>,
        stop_loss_pct: Option<f64>,
        deadline_timeout_sec: Option<u64>,
    ) -> Self {
        Self {
            wallet_pubkeys: vec![wallet_pubkey.into()],
            strategy: strategy_config_from_optional(target_profit_pct, stop_loss_pct, None, None),
            deadline_timeout_sec: deadline_timeout_sec.unwrap_or(0),
            send_mode: None,
            tip_lamports: None,
            watch_wallets: Vec::new(),
            mirror_config: None,
        }
    }
}

/// Builds wire strategy config from optional TP/SL settings.
///
/// Unset values default to `0.0` (disabled). For more control, use
/// [`StrategyConfigBuilder`].
pub fn strategy_config_from_optional(
    target_profit_pct: Option<f64>,
    stop_loss_pct: Option<f64>,
    trailing_stop_pct: Option<f64>,
    sell_on_graduation: Option<bool>,
) -> StrategyConfigMsg {
    StrategyConfigBuilder::new()
        .target_profit_pct(target_profit_pct.unwrap_or(0.0))
        .stop_loss_pct(stop_loss_pct.unwrap_or(0.0))
        .trailing_stop_pct(trailing_stop_pct.unwrap_or(0.0))
        .sell_on_graduation(sell_on_graduation.unwrap_or(false))
        .build()
}

/// Builder for constructing a [`StrategyConfigMsg`] with all strategy fields.
///
/// ```
/// use lasersell_sdk::stream::client::StrategyConfigBuilder;
///
/// let strategy = StrategyConfigBuilder::new()
///     .target_profit_pct(25.0)
///     .stop_loss_pct(10.0)
///     .trailing_stop_pct(5.0)
///     .breakeven_trail_pct(20.0)
///     .build();
/// ```
pub struct StrategyConfigBuilder {
    msg: StrategyConfigMsg,
}

impl StrategyConfigBuilder {
    pub fn new() -> Self {
        Self {
            msg: StrategyConfigMsg {
                target_profit_pct: 0.0,
                stop_loss_pct: 0.0,
                trailing_stop_pct: 0.0,
                sell_on_graduation: false,
                take_profit_levels: Vec::new(),
                liquidity_guard: false,
                breakeven_trail_pct: 0.0,
            },
        }
    }

    pub fn target_profit_pct(mut self, pct: f64) -> Self {
        self.msg.target_profit_pct = pct;
        self
    }

    pub fn stop_loss_pct(mut self, pct: f64) -> Self {
        self.msg.stop_loss_pct = pct;
        self
    }

    pub fn trailing_stop_pct(mut self, pct: f64) -> Self {
        self.msg.trailing_stop_pct = pct;
        self
    }

    pub fn sell_on_graduation(mut self, enabled: bool) -> Self {
        self.msg.sell_on_graduation = enabled;
        self
    }

    pub fn take_profit_levels(mut self, levels: Vec<TakeProfitLevelMsg>) -> Self {
        self.msg.take_profit_levels = levels;
        self
    }

    pub fn liquidity_guard(mut self, enabled: bool) -> Self {
        self.msg.liquidity_guard = enabled;
        self
    }

    pub fn breakeven_trail_pct(mut self, pct: f64) -> Self {
        self.msg.breakeven_trail_pct = pct;
        self
    }

    pub fn build(self) -> StrategyConfigMsg {
        self.msg
    }
}

impl Default for StrategyConfigBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// Connection lifecycle updates produced by the stream worker.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StreamConnectionStatus {
    Connected,
    Disconnected,
}

/// Active stream connection channels.
///
/// Internally, messages are produced by the background websocket worker.
#[derive(Debug)]
pub struct StreamConnection {
    sender: StreamSender,
    receiver: mpsc::UnboundedReceiver<ServerMessage>,
    status: Option<mpsc::UnboundedReceiver<StreamConnectionStatus>>,
}

impl StreamConnection {
    /// Returns a cloneable sender for client commands.
    pub fn sender(&self) -> StreamSender {
        self.sender.clone()
    }

    /// Splits into sender and raw inbound message receiver.
    pub fn split(self) -> (StreamSender, mpsc::UnboundedReceiver<ServerMessage>) {
        (self.sender, self.receiver)
    }

    /// Splits into sender, raw inbound message receiver, and connection status
    /// receiver.
    pub fn split_with_status(
        self,
    ) -> (
        StreamSender,
        mpsc::UnboundedReceiver<ServerMessage>,
        mpsc::UnboundedReceiver<StreamConnectionStatus>,
    ) {
        (self.sender, self.receiver, self.status.unwrap_or_else(|| mpsc::unbounded_channel().1))
    }

    /// Takes the status channel receiver, leaving `None` in its place.
    /// Can only be called once — subsequent calls return `None`.
    pub fn take_status(&mut self) -> Option<mpsc::UnboundedReceiver<StreamConnectionStatus>> {
        self.status.take()
    }

    /// Receives the next server message from the stream worker.
    pub async fn recv(&mut self) -> Option<ServerMessage> {
        self.receiver.recv().await
    }

    /// Splits the inbound stream into high- and low-priority lanes.
    ///
    /// This does **not** create a second websocket connection; it spawns a small
    /// demux task that reads from the single underlying socket and routes:
    ///
    /// - **High**: all messages except `PnlUpdate`
    /// - **Low**: `PnlUpdate` (best-effort; dropped when the low lane is full)
    ///
    /// `low_capacity` controls how many low-priority messages to buffer.
    pub fn into_lanes(self, low_capacity: usize) -> StreamConnectionLanes {
        let (high_tx, high_rx) = mpsc::unbounded_channel();
        let (low_tx, low_rx) = mpsc::channel(low_capacity);

        let _status = self.status; // Intentionally dropped — lanes don't use status
        let mut receiver = self.receiver;
        tokio::spawn(async move {
            while let Some(message) = receiver.recv().await {
                match message {
                    ServerMessage::PnlUpdate { .. } => {
                        // Best-effort: drop when low lane is backpressured.
                        let _ = low_tx.try_send(message);
                    }
                    _ => {
                        let _ = high_tx.send(message);
                    }
                }
            }
        });

        StreamConnectionLanes {
            sender: self.sender,
            high: high_rx,
            low: low_rx,
        }
    }
}

/// Stream connection with split inbound priority lanes.
///
/// Intended for hot paths where you cannot afford to let frequent low-value
/// messages (eg, PnL updates) delay exit signals / tx delivery.
#[derive(Debug)]
pub struct StreamConnectionLanes {
    sender: StreamSender,
    high: mpsc::UnboundedReceiver<ServerMessage>,
    low: mpsc::Receiver<ServerMessage>,
}

impl StreamConnectionLanes {
    /// Returns a cloneable sender for client commands.
    pub fn sender(&self) -> StreamSender {
        self.sender.clone()
    }

    /// Splits into sender, high-priority receiver, and low-priority receiver.
    pub fn split(
        self,
    ) -> (
        StreamSender,
        mpsc::UnboundedReceiver<ServerMessage>,
        mpsc::Receiver<ServerMessage>,
    ) {
        (self.sender, self.high, self.low)
    }

    /// Receives the next high-priority server message.
    pub async fn recv_high(&mut self) -> Option<ServerMessage> {
        self.high.recv().await
    }

    /// Receives the next low-priority server message.
    pub async fn recv_low(&mut self) -> Option<ServerMessage> {
        self.low.recv().await
    }
}

/// Selects a position either by token account or numeric position id.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PositionSelector {
    /// Select by token account address.
    TokenAccount(String),
    /// Select by position id.
    PositionId(u64),
}

/// Conversion trait for position selector inputs.
pub trait IntoPositionSelector {
    /// Converts an input into a canonical [`PositionSelector`].
    fn into_position_selector(self) -> PositionSelector;
}

impl IntoPositionSelector for PositionSelector {
    fn into_position_selector(self) -> PositionSelector {
        self
    }
}

impl IntoPositionSelector for String {
    fn into_position_selector(self) -> PositionSelector {
        PositionSelector::TokenAccount(self)
    }
}

impl IntoPositionSelector for &String {
    fn into_position_selector(self) -> PositionSelector {
        PositionSelector::TokenAccount(self.clone())
    }
}

impl IntoPositionSelector for &str {
    fn into_position_selector(self) -> PositionSelector {
        PositionSelector::TokenAccount(self.to_string())
    }
}

impl IntoPositionSelector for u64 {
    fn into_position_selector(self) -> PositionSelector {
        PositionSelector::PositionId(self)
    }
}

/// Cloneable sender for outbound stream client messages.
#[derive(Clone, Debug)]
pub struct StreamSender {
    tx: mpsc::UnboundedSender<ClientMessage>,
}

impl StreamSender {
    /// Sends a raw client message to the stream worker queue.
    pub fn send(&self, message: ClientMessage) -> Result<(), StreamClientError> {
        self.tx
            .send(message)
            .map_err(|_| StreamClientError::SendQueueClosed)
    }

    /// Sends a heartbeat ping with client timestamp.
    pub fn ping(&self, client_time_ms: u64) -> Result<(), StreamClientError> {
        self.send(ClientMessage::Ping { client_time_ms })
    }

    /// Updates strategy parameters for the active stream session.
    pub fn update_strategy(&self, strategy: StrategyConfigMsg) -> Result<(), StreamClientError> {
        self.send(ClientMessage::UpdateStrategy { strategy })
    }

    /// Overrides the strategy for a single active position.
    ///
    /// The provided strategy fully replaces the session-level strategy for this
    /// position only. Other positions are unaffected.
    pub fn update_position_strategy(
        &self,
        position_id: u64,
        strategy: StrategyConfigMsg,
    ) -> Result<(), StreamClientError> {
        self.send(ClientMessage::UpdatePositionStrategy {
            position_id,
            strategy,
        })
    }

    /// Replaces the set of monitored wallets for the active session.
    pub fn update_wallets(&self, wallet_pubkeys: Vec<String>) -> Result<(), StreamClientError> {
        self.send(ClientMessage::UpdateWallets { wallet_pubkeys })
    }

    /// Replaces the set of watched wallets for copy trading.
    pub fn update_watch_wallets(&self, watch_wallets: Vec<WatchWalletEntryMsg>) -> Result<(), StreamClientError> {
        self.send(ClientMessage::UpdateWatchWallets { watch_wallets })
    }

    /// Requests a position close by token account or position id.
    pub fn close_position<S>(&self, selector: S) -> Result<(), StreamClientError>
    where
        S: IntoPositionSelector,
    {
        self.send(close_message(selector.into_position_selector()))
    }

    /// Convenience wrapper for closing by position id.
    pub fn close_by_id(&self, position_id: u64) -> Result<(), StreamClientError> {
        self.close_position(PositionSelector::PositionId(position_id))
    }

    /// Requests an exit signal by token account or position id.
    ///
    /// `slippage_bps` overrides stream strategy slippage for this request when
    /// provided.
    pub fn request_exit_signal<S>(
        &self,
        selector: S,
        slippage_bps: Option<u16>,
    ) -> Result<(), StreamClientError>
    where
        S: IntoPositionSelector,
    {
        self.send(request_exit_signal_message(
            selector.into_position_selector(),
            slippage_bps,
        ))
    }

    /// Convenience wrapper for requesting an exit signal by position id.
    pub fn request_exit_signal_by_id(
        &self,
        position_id: u64,
        slippage_bps: Option<u16>,
    ) -> Result<(), StreamClientError> {
        self.request_exit_signal(PositionSelector::PositionId(position_id), slippage_bps)
    }

    /// Reports the outcome of a mirror buy transaction back to the stream.
    ///
    /// Call this after signing and submitting (or failing to submit) a
    /// [`StreamEvent::MirrorBuySignal`] transaction. Allows the stream to
    /// immediately clear pending state instead of waiting for the periodic
    /// cleanup sweep.
    pub fn mirror_buy_result(
        &self,
        mint: impl Into<String>,
        success: bool,
    ) -> Result<(), StreamClientError> {
        self.send(ClientMessage::MirrorBuyResult {
            mint: mint.into(),
            success,
        })
    }
}

fn close_message(selector: PositionSelector) -> ClientMessage {
    match selector {
        PositionSelector::TokenAccount(token_account) => ClientMessage::ClosePosition {
            position_id: None,
            token_account: Some(token_account),
        },
        PositionSelector::PositionId(position_id) => ClientMessage::ClosePosition {
            position_id: Some(position_id),
            token_account: None,
        },
    }
}

fn request_exit_signal_message(
    selector: PositionSelector,
    slippage_bps: Option<u16>,
) -> ClientMessage {
    match selector {
        PositionSelector::TokenAccount(token_account) => ClientMessage::RequestExitSignal {
            position_id: None,
            token_account: Some(token_account),
            slippage_bps,
        },
        PositionSelector::PositionId(position_id) => ClientMessage::RequestExitSignal {
            position_id: Some(position_id),
            token_account: None,
            slippage_bps,
        },
    }
}

/// Errors produced by stream transport and protocol handling.
#[derive(Debug, Error)]
pub enum StreamClientError {
    /// Websocket transport error.
    #[error("websocket error: {0}")]
    WebSocket(#[from] WsError),

    /// JSON serialization/deserialization error.
    #[error("json error: {0}")]
    Json(#[from] serde_json::Error),

    /// API key could not be converted to a valid HTTP header value.
    #[error("invalid api-key header: {0}")]
    InvalidApiKeyHeader(#[from] InvalidHeaderValue),

    /// Outbound message queue has been closed.
    #[error("send queue is closed")]
    SendQueueClosed,

    /// Stream protocol or handshake contract error.
    #[error("protocol error: {0}")]
    Protocol(String),
}

pub(crate) fn validate_strategy_thresholds(
    strategy: &StrategyConfigMsg,
    deadline_timeout_sec: u64,
) -> Result<(), StreamClientError> {
    validate_strategy_value(strategy.target_profit_pct, "strategy.target_profit_pct")?;
    validate_strategy_value(strategy.stop_loss_pct, "strategy.stop_loss_pct")?;
    validate_strategy_value(strategy.trailing_stop_pct, "strategy.trailing_stop_pct")?;

    if strategy.target_profit_pct > 0.0
        || strategy.stop_loss_pct > 0.0
        || strategy.trailing_stop_pct > 0.0
        || deadline_timeout_sec > 0
    {
        return Ok(());
    }

    Err(StreamClientError::Protocol(
        "at least one of strategy.target_profit_pct, strategy.stop_loss_pct, strategy.trailing_stop_pct, or deadline_timeout_sec must be > 0"
            .to_string(),
    ))
}

fn validate_strategy_value(value: f64, field: &str) -> Result<(), StreamClientError> {
    if !value.is_finite() {
        return Err(StreamClientError::Protocol(format!(
            "{field} must be a finite number"
        )));
    }
    if value < 0.0 {
        return Err(StreamClientError::Protocol(format!("{field} must be >= 0")));
    }
    Ok(())
}

enum SessionOutcome {
    GracefulShutdown,
    Reconnect,
}

async fn stream_connection_worker(
    url: String,
    api_key: SecretString,
    configure: StreamConfigure,
    mut outbound_rx: mpsc::UnboundedReceiver<ClientMessage>,
    inbound_tx: mpsc::UnboundedSender<ServerMessage>,
    status_tx: mpsc::UnboundedSender<StreamConnectionStatus>,
    ready_tx: oneshot::Sender<Result<(), StreamClientError>>,
    tls_connector: Option<Connector>,
) {
    let mut ready_tx = Some(ready_tx);
    let mut pending = VecDeque::new();
    let mut backoff = MIN_RECONNECT_BACKOFF;

    debug!(event = "stream_connecting");

    loop {
        match run_connected_session(
            &url,
            &api_key,
            &configure,
            &mut outbound_rx,
            &inbound_tx,
            &status_tx,
            &mut pending,
            &mut ready_tx,
            tls_connector.clone(),
        )
        .await
        {
            Ok(SessionOutcome::GracefulShutdown) => {
                info!(event = "stream_worker_graceful_shutdown");
                let _ = status_tx.send(StreamConnectionStatus::Disconnected);
                if let Some(tx) = ready_tx.take() {
                    let _ = tx.send(Err(StreamClientError::SendQueueClosed));
                }
                break;
            }
            Ok(SessionOutcome::Reconnect) => {
                warn!(event = "stream_worker_reconnect");
                let _ = status_tx.send(StreamConnectionStatus::Disconnected);
                backoff = MIN_RECONNECT_BACKOFF;
            }
            Err(err) => {
                warn!(event = "stream_worker_connect_error", error = %err);
                if let Some(tx) = ready_tx.take() {
                    let _ = tx.send(Err(err));
                    return;
                }
            }
        }

        if outbound_rx.is_closed() {
            break;
        }

        debug!(event = "stream_reconnect_backoff", delay_ms = backoff.as_millis() as u64);
        if !collect_messages_during_delay(backoff, &mut outbound_rx, &mut pending).await {
            break;
        }

        backoff = std::cmp::min(backoff.saturating_mul(2), MAX_RECONNECT_BACKOFF);
    }
}

async fn run_connected_session(
    url: &str,
    api_key: &SecretString,
    configure: &StreamConfigure,
    outbound_rx: &mut mpsc::UnboundedReceiver<ClientMessage>,
    inbound_tx: &mpsc::UnboundedSender<ServerMessage>,
    status_tx: &mpsc::UnboundedSender<StreamConnectionStatus>,
    pending: &mut VecDeque<ClientMessage>,
    ready_tx: &mut Option<oneshot::Sender<Result<(), StreamClientError>>>,
    tls_connector: Option<Connector>,
) -> Result<SessionOutcome, StreamClientError> {
    let mut request = url.into_client_request()?;
    let api_key_header = api_key.expose_secret().parse()?;
    request.headers_mut().insert("x-api-key", api_key_header);

    let (mut socket, _) = if let Some(connector) = tls_connector {
        connect_async_tls_with_config(request, None, false, Some(connector)).await?
    } else {
        connect_async(request).await?
    };
    debug!(event = "stream_ws_connected");

    // Send configure immediately — server waits for it before sending hello_ok.
    let configure_msg = ClientMessage::Configure {
        wallet_pubkeys: configure.wallet_pubkeys.clone(),
        strategy: configure.strategy.clone(),
        send_mode: configure.send_mode.clone(),
        tip_lamports: configure.tip_lamports,
        watch_wallets: configure.watch_wallets.clone(),
        mirror_config: configure.mirror_config.clone(),
    };
    send_client_message(&mut socket, &configure_msg).await?;
    debug!(event = "stream_configure_sent");

    // Server responds with hello_ok after processing configure.
    let hello_message = recv_server_message_before_configure(&mut socket).await?;

    if !matches!(&hello_message, ServerMessage::HelloOk { .. }) {
        let detail = match &hello_message {
            ServerMessage::Error { code, message } => format!("server error [{code}]: {message}"),
            other => format!("unexpected message: {}", serde_json::to_string(other).unwrap_or_else(|_| format!("{other:?}"))),
        };
        return Err(StreamClientError::Protocol(
            format!("expected hello_ok after configure, got: {detail}"),
        ));
    }
    debug!(event = "stream_hello_ok_received");
    let _ = inbound_tx.send(hello_message);
    let _ = status_tx.send(StreamConnectionStatus::Connected);
    info!(event = "stream_configured");

    if let Some(tx) = ready_tx.take() {
        let _ = tx.send(Ok(()));
    }

    while let Some(next) = pending.pop_front() {
        if send_client_message(&mut socket, &next).await.is_err() {
            pending.push_front(next);
            return Ok(SessionOutcome::Reconnect);
        }
    }

    let mut keepalive = tokio::time::interval(KEEPALIVE_INTERVAL);
    keepalive.reset();

    loop {
        tokio::select! {
            _ = keepalive.tick() => {
                if socket.send(Message::Ping(vec![].into())).await.is_err() {
                    return Ok(SessionOutcome::Reconnect);
                }
            }
            maybe_outbound = outbound_rx.recv() => {
                match maybe_outbound {
                    Some(client_msg) => {
                        if send_client_message(&mut socket, &client_msg).await.is_err() {
                            pending.push_front(client_msg);
                            return Ok(SessionOutcome::Reconnect);
                        }
                    }
                    None => {
                        let _ = socket.close(None).await;
                        return Ok(SessionOutcome::GracefulShutdown);
                    }
                }
            }
            maybe_inbound = socket.next() => {
                match maybe_inbound {
                    Some(Ok(Message::Text(text))) => {
                        trace!(event = "stream_raw_message", raw = %text);
                        match parse_server_message(&text) {
                            Ok(server_msg) => {
                                debug!(event = "stream_server_msg", msg_type = server_msg_label(&server_msg));
                                let _ = inbound_tx.send(server_msg);
                            }
                            Err(err) => {
                                warn!(
                                    event = "stream_msg_parse_error",
                                    error = %err,
                                    raw_message = %text,
                                    "skipping unparseable server message"
                                );
                            }
                        }
                    }
                    Some(Ok(Message::Ping(payload))) => {
                        if socket.send(Message::Pong(payload)).await.is_err() {
                            return Ok(SessionOutcome::Reconnect);
                        }
                    }
                    Some(Ok(Message::Pong(_))) => {}
                    Some(Ok(Message::Close(_))) => {
                        debug!(event = "stream_ws_close_received");
                        return Ok(SessionOutcome::Reconnect);
                    }
                    Some(Ok(_)) => return Ok(SessionOutcome::Reconnect),
                    Some(Err(_)) => {
                        warn!(event = "stream_ws_error");
                        return Ok(SessionOutcome::Reconnect);
                    }
                    None => {
                        debug!(event = "stream_ws_ended");
                        return Ok(SessionOutcome::Reconnect);
                    }
                }
            }
        }
    }
}

async fn recv_server_message_before_configure<S>(
    socket: &mut tokio_tungstenite::WebSocketStream<S>,
) -> Result<ServerMessage, StreamClientError>
where
    tokio_tungstenite::WebSocketStream<S>: futures_util::Sink<Message, Error = WsError>
        + Stream<Item = Result<Message, WsError>>
        + Unpin,
{
    loop {
        match socket.next().await {
            Some(Ok(Message::Text(text))) => return parse_server_message(&text),
            Some(Ok(Message::Ping(payload))) => {
                socket.send(Message::Pong(payload)).await?;
            }
            Some(Ok(Message::Pong(_))) => {}
            Some(Ok(Message::Close(_))) => {
                return Err(StreamClientError::Protocol(
                    "socket closed before hello_ok".to_string(),
                ));
            }
            Some(Ok(_)) => {
                return Err(StreamClientError::Protocol(
                    "received non-text frame before hello_ok".to_string(),
                ));
            }
            Some(Err(err)) => return Err(StreamClientError::WebSocket(err)),
            None => {
                return Err(StreamClientError::Protocol(
                    "socket ended before hello_ok".to_string(),
                ));
            }
        }
    }
}

fn server_msg_label(msg: &ServerMessage) -> &'static str {
    match msg {
        ServerMessage::HelloOk { .. } => "hello_ok",
        ServerMessage::BalanceUpdate { .. } => "balance_update",
        ServerMessage::PositionOpened { .. } => "position_opened",
        ServerMessage::PositionClosed { .. } => "position_closed",
        ServerMessage::ExitSignalWithTx { .. } => "exit_signal_with_tx",
        ServerMessage::PnlUpdate { .. } => "pnl_update",
        ServerMessage::LiquiditySnapshot { .. } => "liquidity_snapshot",
        ServerMessage::TradeTick { .. } => "trade_tick",
        ServerMessage::Pong { .. } => "pong",
        ServerMessage::Error { .. } => "error",
        ServerMessage::MirrorBuySignal { .. } => "mirror_buy_signal",
        ServerMessage::MirrorBuyFailed { .. } => "mirror_buy_failed",
        ServerMessage::MirrorWalletAutoDisabled { .. } => "mirror_wallet_auto_disabled",
    }
}

fn parse_server_message(text: &str) -> Result<ServerMessage, StreamClientError> {
    serde_json::from_str(text).map_err(StreamClientError::Json)
}

async fn send_client_message<S>(
    socket: &mut tokio_tungstenite::WebSocketStream<S>,
    message: &ClientMessage,
) -> Result<(), StreamClientError>
where
    tokio_tungstenite::WebSocketStream<S>: futures_util::Sink<Message, Error = WsError> + Unpin,
{
    let text = serde_json::to_string(message)?;
    socket.send(Message::Text(text)).await?;
    Ok(())
}

async fn collect_messages_during_delay(
    delay: Duration,
    outbound_rx: &mut mpsc::UnboundedReceiver<ClientMessage>,
    pending: &mut VecDeque<ClientMessage>,
) -> bool {
    let sleep = tokio::time::sleep(delay);
    tokio::pin!(sleep);

    loop {
        tokio::select! {
            _ = &mut sleep => return true,
            maybe_message = outbound_rx.recv() => {
                match maybe_message {
                    Some(message) => pending.push_back(message),
                    None => return false,
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use secrecy::SecretString;

    use super::{
        strategy_config_from_optional, validate_strategy_thresholds, StreamClient,
        LOCAL_STREAM_ENDPOINT, STREAM_ENDPOINT,
    };

    #[test]
    fn stream_client_uses_production_endpoint_by_default() {
        let client = StreamClient::new(SecretString::new("test-api-key".to_string()));
        assert_eq!(client.endpoint(), STREAM_ENDPOINT);
    }

    #[test]
    fn stream_client_uses_local_endpoint_when_enabled() {
        let client =
            StreamClient::new(SecretString::new("test-api-key".to_string())).with_local_mode(true);
        assert_eq!(client.endpoint(), LOCAL_STREAM_ENDPOINT);
    }

    #[test]
    fn stream_client_endpoint_override_takes_precedence() {
        let client = StreamClient::new(SecretString::new("test-api-key".to_string()))
            .with_local_mode(true)
            .with_endpoint("wss://stream-dev.example/ws   \n");
        assert_eq!(client.endpoint(), "wss://stream-dev.example/ws");
    }

    #[test]
    fn optional_strategy_builder_defaults_unset_values_to_zero() {
        let strategy = strategy_config_from_optional(None, Some(1.5), None, None);
        assert_eq!(strategy.target_profit_pct, 0.0);
        assert_eq!(strategy.stop_loss_pct, 1.5);
    }

    #[test]
    fn validation_accepts_target_only() {
        let strategy = strategy_config_from_optional(Some(2.0), None, None, None);
        assert!(validate_strategy_thresholds(&strategy, 0).is_ok());
    }

    #[test]
    fn validation_accepts_stop_only() {
        let strategy = strategy_config_from_optional(None, Some(1.0), None, None);
        assert!(validate_strategy_thresholds(&strategy, 0).is_ok());
    }

    #[test]
    fn validation_accepts_deadline_only() {
        let strategy = strategy_config_from_optional(None, None, None, None);
        assert!(validate_strategy_thresholds(&strategy, 45).is_ok());
    }

    #[test]
    fn validation_rejects_when_all_thresholds_disabled() {
        let strategy = strategy_config_from_optional(None, None, None, None);
        assert!(validate_strategy_thresholds(&strategy, 0).is_err());
    }

    #[test]
    fn validation_rejects_negative_values() {
        let strategy = strategy_config_from_optional(Some(-1.0), None, None, None);
        assert!(validate_strategy_thresholds(&strategy, 0).is_err());
    }
}