zrsclient 0.3.0

use std::collections::HashMap;
use std::io::{Cursor, Seek, SeekFrom};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use byteorder::{BigEndian, ReadBytesExt};
use futures_util::{SinkExt, StreamExt};
use log::debug;
use serde::{Serialize, Deserialize};
use serde_json::{json, Value as JsonValue};
use tokio::sync::mpsc;
use tokio_tungstenite::connect_async;
use tokio_tungstenite::tungstenite::Message;

/// Result of an outbound ticker command (subscribe / unsubscribe / set_mode).
/// An error means the connection's writer task has gone away.
pub type SendResult = Result<(), mpsc::error::SendError<String>>;

/// Streaming mode a tick was received in.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Mode {
    Ltp,
    Quote,
    Full,
}

/// Open/High/Low/Close prices carried by quote and full ticks.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Ohlc {
    pub open: f64,
    pub high: f64,
    pub low: f64,
    pub close: f64,
}

/// A single level of the market-depth (order book) ladder.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct DepthItem {
    pub quantity: i64,
    pub price: f64,
    pub orders: i64,
}

/// Five-level buy/sell market depth (present only in `full` mode for tradable
/// instruments).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct MarketDepth {
    pub buy: Vec<DepthItem>,
    pub sell: Vec<DepthItem>,
}

/// A single decoded market-data tick.
///
/// Fields absent in the received [`Mode`] (or for index instruments) are `None`.
/// `ltp` carries only `last_price`; `quote` adds volumes and `ohlc`; `full` adds
/// open interest, timestamps and `depth`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Tick {
    pub mode: Mode,
    pub instrument_token: u32,
    pub tradable: bool,
    pub last_price: f64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_quantity: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub average_price: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub volume: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub buy_quantity: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub sell_quantity: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ohlc: Option<Ohlc>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub change: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_trade_time: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub oi: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub oi_day_high: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub oi_day_low: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub timestamp: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub depth: Option<MarketDepth>,
}

impl Tick {
    /// A tick with only the fields common to every mode; callers fill in the rest.
    fn base(mode: Mode, instrument_token: u32, tradable: bool, last_price: f64) -> Self {
        Tick {
            mode,
            instrument_token,
            tradable,
            last_price,
            last_quantity: None,
            average_price: None,
            volume: None,
            buy_quantity: None,
            sell_quantity: None,
            ohlc: None,
            change: None,
            last_trade_time: None,
            oi: None,
            oi_day_high: None,
            oi_day_low: None,
            timestamp: None,
            depth: None,
        }
    }

    /// Percentage change vs. the close, or `None` when close is zero/absent.
    fn compute_change(last_price: f64, close: f64) -> Option<f64> {
        if close != 0.0 {
            Some((last_price - close) * 100.0 / close)
        } else {
            None
        }
    }
}

/// KiteTickerHandler lets the user write the business logic inside
/// the corresponding callbacks which are basically proxied from the
/// underlying websocket events.
pub trait KiteTickerHandler {

    fn on_open<T>(&mut self, _ws: &mut WebSocketHandler<T>)
    where T: KiteTickerHandler {
        debug!("Connection opened");
    }

    fn on_ticks<T>(&mut self, _ws: &mut WebSocketHandler<T>, tick: Vec<Tick>)
    where T: KiteTickerHandler {
        debug!("{:?}", tick);
    }

    fn on_close<T>(&mut self, _ws: &mut WebSocketHandler<T>)
    where T: KiteTickerHandler {
        debug!("Connection closed");
    }

    fn on_error<T>(&mut self, _ws: &mut WebSocketHandler<T>)
    where T: KiteTickerHandler {
        debug!("Error");
    }

    /// Called for `order` postback text frames pushed over the ticker socket.
    /// `order` is the `data` object of the postback.
    fn on_order_update<T>(&mut self, _ws: &mut WebSocketHandler<T>, order: JsonValue)
    where T: KiteTickerHandler {
        debug!("Order update: {:?}", order);
    }

    /// Called for `message` text frames (broker messages / alerts).
    fn on_message<T>(&mut self, _ws: &mut WebSocketHandler<T>, message: JsonValue)
    where T: KiteTickerHandler {
        debug!("Message: {:?}", message);
    }
}

/// Handle passed to every callback. Outbound frames are pushed onto an async
/// channel drained by the connection's writer; subscription state is shared
/// across reconnects so it survives a dropped connection.
pub struct WebSocketHandler<T> where T: KiteTickerHandler {
    handler: Arc<Mutex<Box<T>>>,
    cmd_tx: mpsc::UnboundedSender<String>,
    subscribed_tokens: Arc<Mutex<HashMap<u32, String>>>
}

impl<T> WebSocketHandler<T> where T: KiteTickerHandler {
    /// Subscribe to a list of instrument_tokens.
    pub fn subscribe(&mut self, instrument_tokens: Vec<u32>) -> SendResult {
        let data = json!({
            "a": "subscribe",
            "v": instrument_tokens
        });
        {
            let mut tokens = self.subscribed_tokens.lock().unwrap();
            for token in &instrument_tokens {
                tokens.entry(*token).or_insert_with(|| "quote".to_string());
            }
        }
        self.cmd_tx.send(data.to_string())
    }

    /// Unsubscribe the given list of instrument_tokens.
    pub fn unsubscribe(&mut self, instrument_tokens: Vec<u32>) -> SendResult {
        let data = json!({
            "a": "unsubscribe",
            "v": instrument_tokens
        });
        {
            let mut tokens = self.subscribed_tokens.lock().unwrap();
            for token in &instrument_tokens {
                tokens.remove(token);
            }
        }
        self.cmd_tx.send(data.to_string())
    }

    /// Resubscribe to all currently subscribed tokens, restoring their modes.
    ///
    /// Because subscription state is shared across reconnects, calling this from
    /// `on_open` after a reconnect restores every prior subscription.
    pub fn resubscribe(&mut self) -> SendResult {
        // Snapshot and release the lock before calling subscribe/set_mode, which
        // re-acquire it (holding it across those calls would deadlock).
        let mut modes: HashMap<String, Vec<u32>> = HashMap::new();
        {
            let tokens = self.subscribed_tokens.lock().unwrap();
            for (token, mode) in tokens.iter() {
                modes.entry(mode.clone()).or_default().push(*token);
            }
        }

        for (mode, tokens) in modes.iter() {
            debug!("Resubscribing and set mode: {} - {:?}", mode, tokens);
            self.subscribe(tokens.clone())?;
            self.set_mode(mode.as_str(), tokens.clone())?;
        }
        Ok(())
    }

    /// Set streaming mode for the given list of tokens.
    pub fn set_mode(&mut self, mode: &str, instrument_tokens: Vec<u32>) -> SendResult {
        let data = json!({
            "a": "mode",
            "v": [mode.to_string(), instrument_tokens]
        });
        {
            let mut tokens = self.subscribed_tokens.lock().unwrap();
            for token in &instrument_tokens {
                *tokens.entry(*token).or_default() = mode.to_string();
            }
        }
        self.cmd_tx.send(data.to_string())
    }
}

/// Decode a binary market-data frame into a list of [`Tick`]s.
///
/// The frame is `[i16 packet_count]` followed by, per packet,
/// `[i16 packet_length][packet_length bytes]`. Malformed or truncated frames
/// are skipped rather than panicking.
fn parse_binary_message(data: &[u8]) -> Vec<Tick> {
    let mut tick_data: Vec<Tick> = Vec::new();
    if data.len() <= 2 {
        return tick_data;
    }

    let buf_len = data.len() as u64;
    let mut reader = Cursor::new(data);
    let number_of_packets = reader.read_i16::<BigEndian>().unwrap();

    for _ in 0..number_of_packets {
        // Need the 2-byte packet-length header.
        if buf_len - reader.position() < 2 {
            debug!("truncated tick stream: missing packet length header");
            break;
        }
        let packet_length = reader.read_i16::<BigEndian>().unwrap();

        // A packet is at least a 4-byte instrument token. Bail on a
        // truncated/garbage frame instead of unwrap-panicking. One check here
        // keeps every read below in-bounds.
        if packet_length < 4 || buf_len - reader.position() < packet_length as u64 {
            debug!("skipping malformed packet (length {})", packet_length);
            break;
        }
        // Start of this packet's body.
        let packet_start = reader.position();

        let instrument_token = reader.read_u32::<BigEndian>().unwrap();
        let segment = instrument_token & 0xFF;
        let divisor: f64 = match segment {
            3 => 10_000_000.0, // NSE currency (CDS)
            6 => 10_000.0,     // BSE currency (BCD)
            _ => 100.0,
        };
        // indices (segment 9) are not tradable
        let tradable = segment != 9;

        // Read a price field (paise -> rupees).
        let price = |r: &mut Cursor<&[u8]>| r.read_i32::<BigEndian>().unwrap() as f64 / divisor;

        match packet_length {
            // LTP
            8 => {
                let last_price = price(&mut reader);
                tick_data.push(Tick::base(Mode::Ltp, instrument_token, tradable, last_price));
            },

            // Index quote/full
            28 | 32 => {
                let mode = if packet_length == 28 { Mode::Quote } else { Mode::Full };
                let last_price = price(&mut reader);
                let ohlc = Ohlc {
                    high: price(&mut reader),
                    low: price(&mut reader),
                    open: price(&mut reader),
                    close: price(&mut reader),
                };
                let mut tick = Tick::base(mode, instrument_token, tradable, last_price);
                tick.change = Tick::compute_change(last_price, ohlc.close);
                tick.ohlc = Some(ohlc);
                if packet_length == 32 {  // full mode
                    // Bytes 24-28 are the exchange-sent price change, which we
                    // compute ourselves; the exchange timestamp (epoch seconds)
                    // follows at bytes 28-32.
                    reader.read_i32::<BigEndian>().unwrap(); // price change, discarded
                    tick.timestamp = Some(reader.read_i32::<BigEndian>().unwrap() as i64);
                }
                tick_data.push(tick);
            },

            // Quote/Full (tradable)
            44 | 184 => {
                let mode = if packet_length == 44 { Mode::Quote } else { Mode::Full };
                let last_price = price(&mut reader);
                let last_quantity = reader.read_i32::<BigEndian>().unwrap() as i64;
                let average_price = price(&mut reader);
                let volume = reader.read_i32::<BigEndian>().unwrap() as i64;
                let buy_quantity = reader.read_i32::<BigEndian>().unwrap() as i64;
                let sell_quantity = reader.read_i32::<BigEndian>().unwrap() as i64;
                let ohlc = Ohlc {
                    open: price(&mut reader),
                    high: price(&mut reader),
                    low: price(&mut reader),
                    close: price(&mut reader),
                };

                let mut tick = Tick::base(mode, instrument_token, tradable, last_price);
                tick.change = Tick::compute_change(last_price, ohlc.close);
                tick.last_quantity = Some(last_quantity);
                tick.average_price = Some(average_price);
                tick.volume = Some(volume);
                tick.buy_quantity = Some(buy_quantity);
                tick.sell_quantity = Some(sell_quantity);
                tick.ohlc = Some(ohlc);

                if packet_length == 184 {
                    tick.last_trade_time = Some(reader.read_i32::<BigEndian>().unwrap() as i64);
                    tick.oi = Some(reader.read_i32::<BigEndian>().unwrap() as i64);
                    tick.oi_day_high = Some(reader.read_i32::<BigEndian>().unwrap() as i64);
                    tick.oi_day_low = Some(reader.read_i32::<BigEndian>().unwrap() as i64);
                    tick.timestamp = Some(reader.read_i32::<BigEndian>().unwrap() as i64);

                    // Remaining (184 - 64) / 12 = 10 depth entries: 5 buy then 5 sell.
                    let mut buy = Vec::with_capacity(5);
                    let mut sell = Vec::with_capacity(5);
                    for index in 0..10 {
                        let item = DepthItem {
                            quantity: reader.read_i32::<BigEndian>().unwrap() as i64,
                            price: price(&mut reader),
                            orders: reader.read_i16::<BigEndian>().unwrap() as i64,
                        };
                        if index < 5 { buy.push(item); } else { sell.push(item); }
                        // 2 bytes padding, ignored.
                        reader.read_i16::<BigEndian>().unwrap();
                    }
                    tick.depth = Some(MarketDepth { buy, sell });
                }

                tick_data.push(tick);
            }

            _ => {
                debug!("undefined packet length received: {}", packet_length)
            }
        }

        // Advance to the end of this packet, regardless of how many bytes the
        // matched arm actually consumed. Infallible for an in-range cursor.
        reader.seek(SeekFrom::Start(packet_start + packet_length as u64)).unwrap();
    }

    tick_data
}

/// Dispatch a text (postback) frame to the appropriate callback.
fn dispatch_text<T>(text: &str, ws: &mut WebSocketHandler<T>)
where T: KiteTickerHandler {
    // Non-binary frames are postbacks/updates, shaped as
    // {"type": "order" | "error" | "message", "data": {...}}
    match serde_json::from_str::<JsonValue>(text) {
        Ok(parsed) => {
            let msg_type = parsed.get("type").and_then(|t| t.as_str()).unwrap_or("");
            let data = parsed.get("data").cloned().unwrap_or(JsonValue::Null);
            let handler = ws.handler.clone();
            match msg_type {
                "order" => handler.lock().unwrap().on_order_update(ws, data),
                "message" => handler.lock().unwrap().on_message(ws, data),
                "error" => {
                    debug!("Ticker error message: {:?}", data);
                    handler.lock().unwrap().on_error(ws);
                },
                _ => debug!("Unhandled text message type '{}': {}", msg_type, text),
            }
        },
        Err(e) => debug!("Failed to parse text message '{}': {}", text, e),
    }
}

/// Async connection driver: connects, pumps reads to callbacks and outbound
/// commands to the socket, and reconnects with exponential backoff.
async fn run_ticker<F>(
    url: String,
    shared_handler: Arc<Mutex<Box<F>>>,
    shared_tokens: Arc<Mutex<HashMap<u32, String>>>,
    reconnect: bool,
    max_retries: u32,
    base: u64,
    max_delay: u64,
) where F: KiteTickerHandler + Send + 'static {
    let mut attempt: u32 = 0;
    loop {
        match connect_async(&url).await {
            Ok((ws_stream, _resp)) => {
                attempt = 0; // a live connection resets the backoff
                debug!("kiteticker: connected");
                let (mut write, mut read) = ws_stream.split();
                // Per-connection command channel; the handle owns the only sender,
                // so recv() never yields None while the connection is alive.
                let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<String>();
                let mut ws = WebSocketHandler {
                    handler: shared_handler.clone(),
                    cmd_tx,
                    subscribed_tokens: shared_tokens.clone(),
                };

                // on_open fires the user's subscribe logic, which queues commands.
                {
                    let handler = ws.handler.clone();
                    handler.lock().unwrap().on_open(&mut ws);
                }

                loop {
                    tokio::select! {
                        maybe_msg = read.next() => {
                            match maybe_msg {
                                Some(Ok(Message::Binary(bytes))) => {
                                    let ticks = parse_binary_message(&bytes);
                                    if !ticks.is_empty() {
                                        let handler = ws.handler.clone();
                                        handler.lock().unwrap().on_ticks(&mut ws, ticks);
                                    }
                                }
                                Some(Ok(Message::Text(text))) => dispatch_text(&text, &mut ws),
                                Some(Ok(Message::Ping(payload))) => {
                                    let _ = write.send(Message::Pong(payload)).await;
                                }
                                Some(Ok(Message::Close(_))) => {
                                    debug!("kiteticker: server closed the connection");
                                    break;
                                }
                                Some(Ok(_)) => {}
                                Some(Err(e)) => {
                                    debug!("kiteticker: read error: {:?}", e);
                                    break;
                                }
                                None => {
                                    debug!("kiteticker: stream ended");
                                    break;
                                }
                            }
                        }
                        maybe_cmd = cmd_rx.recv() => {
                            if let Some(text) = maybe_cmd {
                                if let Err(e) = write.send(Message::Text(text)).await {
                                    debug!("kiteticker: write error: {:?}", e);
                                    break;
                                }
                            }
                        }
                    }
                }

                {
                    let handler = ws.handler.clone();
                    handler.lock().unwrap().on_close(&mut ws);
                }
            }
            Err(e) => {
                debug!("kiteticker: connect failed: {:?}", e);
            }
        }

        if !reconnect {
            break;
        }
        attempt += 1;
        if max_retries != 0 && attempt > max_retries {
            debug!("kiteticker: giving up after {} reconnect attempts", attempt - 1);
            break;
        }
        let shift = (attempt - 1).min(16);
        let delay = std::cmp::min(base.saturating_mul(1u64 << shift), max_delay);
        debug!("kiteticker: reconnecting in {}s (attempt {})", delay, attempt);
        tokio::time::sleep(Duration::from_secs(delay)).await;
    }
}

pub struct KiteTicker {
    api_key: String,
    access_token: String,
    reconnect: bool,
    max_retries: u32,
    base_delay_secs: u64,
    max_delay_secs: u64
}

impl KiteTicker {

    /// Constructor. Automatic reconnection is enabled by default (retry forever,
    /// exponential backoff from 1s capped at 30s). Tune it with
    /// [`KiteTicker::set_reconnect`].
    pub fn new(api_key: &str, access_token: &str) -> Self {
        Self {
            api_key: api_key.to_string(),
            access_token: access_token.to_string(),
            reconnect: true,
            max_retries: 0,
            base_delay_secs: 1,
            max_delay_secs: 30
        }
    }

    /// Configure automatic reconnection. `max_retries` of `0` retries forever.
    /// Delay grows exponentially from `base_delay_secs`, capped at `max_delay_secs`.
    /// Must be called before [`KiteTicker::connect`].
    pub fn set_reconnect(&mut self, enabled: bool, max_retries: u32,
        base_delay_secs: u64, max_delay_secs: u64) {
        self.reconnect = enabled;
        self.max_retries = max_retries;
        self.base_delay_secs = base_delay_secs.max(1);
        self.max_delay_secs = max_delay_secs.max(1);
    }

    /// Connect the ticker on a background thread.
    ///
    /// `uri` may be a bare host (`ws.kite.trade`, the default when `None`), in
    /// which case `wss://` and the api_key/access_token query params are added,
    /// or a full `ws://`/`wss://` URL, which is used verbatim. On a dropped or
    /// failed connection the driver reconnects (if enabled) with exponential
    /// backoff; subscription state is shared across reconnects, so a handler that
    /// (re)subscribes in `on_open` — or calls `resubscribe()` — is restored.
    pub fn connect<F>(&mut self, handler: F, uri: Option<&str>) -> Result<(), Box<dyn std::error::Error>>
        where F: KiteTickerHandler + Send + 'static {
        let url = match uri {
            Some(u) if u.contains("://") => u.to_string(),
            Some(u) => format!("wss://{}?api_key={}&access_token={}", u, self.api_key, self.access_token),
            None => format!("wss://ws.kite.trade?api_key={}&access_token={}", self.api_key, self.access_token),
        };

        let shared_handler = Arc::new(Mutex::new(Box::new(handler)));
        let shared_tokens: Arc<Mutex<HashMap<u32, String>>> = Arc::new(Mutex::new(HashMap::new()));
        let reconnect = self.reconnect;
        let max_retries = self.max_retries;
        let base = self.base_delay_secs;
        let max_delay = self.max_delay_secs;

        std::thread::spawn(move || {
            let rt = match tokio::runtime::Builder::new_current_thread().enable_all().build() {
                Ok(rt) => rt,
                Err(e) => { debug!("kiteticker: failed to build runtime: {:?}", e); return; }
            };
            rt.block_on(run_ticker(
                url, shared_handler, shared_tokens,
                reconnect, max_retries, base, max_delay,
            ));
        });

        Ok(())
    }
}

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

    /// Handler that records ticks delivered to `on_ticks`.
    struct CaptureHandler {
        ticks: Arc<Mutex<Vec<Tick>>>,
    }
    impl KiteTickerHandler for CaptureHandler {
        fn on_ticks<T>(&mut self, _ws: &mut WebSocketHandler<T>, tick: Vec<Tick>)
        where T: KiteTickerHandler {
            self.ticks.lock().unwrap().extend(tick);
        }
    }

    /// Build a WebSocketHandler backed by an in-memory command channel, for
    /// exercising subscription bookkeeping without a live socket. The returned
    /// receiver is kept so the channel stays open.
    fn test_handler() -> (WebSocketHandler<CaptureHandler>, mpsc::UnboundedReceiver<String>) {
        let (tx, rx) = mpsc::unbounded_channel::<String>();
        let handler = Arc::new(Mutex::new(Box::new(CaptureHandler {
            ticks: Arc::new(Mutex::new(Vec::new())),
        })));
        let ws = WebSocketHandler {
            handler,
            cmd_tx: tx,
            subscribed_tokens: Arc::new(Mutex::new(HashMap::new())),
        };
        (ws, rx)
    }

    fn ltp_frame(token: u32, price: i32) -> Vec<u8> {
        let mut buf = Vec::new();
        buf.extend_from_slice(&1i16.to_be_bytes());
        buf.extend_from_slice(&8i16.to_be_bytes());
        buf.extend_from_slice(&token.to_be_bytes());
        buf.extend_from_slice(&price.to_be_bytes());
        buf
    }

    /// Two packets in one frame — an LTP (8) and a quote (44) — exercise both the
    /// per-packet decoding and the loop's packet-advance logic.
    #[test]
    fn test_parse_binary_ltp_and_quote() {
        let mut buf: Vec<u8> = Vec::new();
        buf.extend_from_slice(&2i16.to_be_bytes());          // 2 packets
        // Packet A: LTP, token 408065 (segment 1 -> divisor 100), price 15075 -> 150.75
        buf.extend_from_slice(&8i16.to_be_bytes());
        buf.extend_from_slice(&408065u32.to_be_bytes());
        buf.extend_from_slice(&15075i32.to_be_bytes());
        // Packet B: quote (44), same token
        buf.extend_from_slice(&44i16.to_be_bytes());
        buf.extend_from_slice(&408065u32.to_be_bytes());
        for v in [15075, 10, 15000, 100000, 500, 600, 14900, 15200, 14800, 14950] {
            buf.extend_from_slice(&(v as i32).to_be_bytes());
        }

        let ticks = parse_binary_message(&buf);
        assert_eq!(ticks.len(), 2);
        assert_eq!(ticks[0].mode, Mode::Ltp);
        assert_eq!(ticks[0].instrument_token, 408065);
        assert_eq!(ticks[0].last_price, 150.75);
        assert_eq!(ticks[1].mode, Mode::Quote);
        assert_eq!(ticks[1].last_price, 150.75);
        assert_eq!(ticks[1].volume, Some(100000));
        let ohlc = ticks[1].ohlc.as_ref().unwrap();
        assert_eq!(ohlc.open, 149.0);
        assert_eq!(ohlc.close, 149.5);
        assert_eq!(ticks[1].last_quantity, Some(10));
    }

    /// A frame claiming 2 packets but carrying only a truncated one must be
    /// skipped gracefully — never panic.
    #[test]
    fn test_parse_binary_truncated_does_not_panic() {
        let mut buf: Vec<u8> = Vec::new();
        buf.extend_from_slice(&2i16.to_be_bytes());   // claims 2 packets
        buf.extend_from_slice(&8i16.to_be_bytes());   // says length 8...
        buf.extend_from_slice(&408065u32.to_be_bytes()); // ...but only 4 bytes follow

        let ticks = parse_binary_message(&buf);
        assert_eq!(ticks.len(), 0);
    }

    /// Subscription state is shared across connections, so a handler created for
    /// a reconnect sees prior subscriptions and `resubscribe()` can replay them.
    #[test]
    fn test_subscription_state_survives_reconnect() {
        let shared = Arc::new(Mutex::new(HashMap::new()));
        let handler = Arc::new(Mutex::new(Box::new(CaptureHandler {
            ticks: Arc::new(Mutex::new(Vec::new())),
        })));
        let (tx1, _rx1) = mpsc::unbounded_channel::<String>();

        // First "connection".
        let mut h1 = WebSocketHandler {
            handler: handler.clone(),
            cmd_tx: tx1,
            subscribed_tokens: shared.clone(),
        };
        h1.subscribe(vec![408065, 5633]).unwrap();
        h1.set_mode("full", vec![408065]).unwrap();

        // A reconnect builds a fresh handler around the SAME shared state.
        let (tx2, _rx2) = mpsc::unbounded_channel::<String>();
        let mut h2 = WebSocketHandler {
            handler: handler.clone(),
            cmd_tx: tx2,
            subscribed_tokens: shared.clone(),
        };
        {
            let map = h2.subscribed_tokens.lock().unwrap();
            assert_eq!(map.len(), 2);
            assert_eq!(map.get(&408065), Some(&"full".to_string()));
            assert_eq!(map.get(&5633), Some(&"quote".to_string()));
        }

        // Must not deadlock (resubscribe snapshots before re-locking) and must
        // preserve the modes.
        h2.resubscribe().unwrap();
        let map = h2.subscribed_tokens.lock().unwrap();
        assert_eq!(map.get(&408065), Some(&"full".to_string()));
        assert_eq!(map.get(&5633), Some(&"quote".to_string()));
    }

    /// End-to-end over a local tokio-tungstenite server: the client connects,
    /// the server pushes a binary LTP tick, and the client's on_ticks receives it.
    #[test]
    fn test_ticker_end_to_end() {
        let (port_tx, port_rx) = std::sync::mpsc::channel::<u16>();

        // Server thread: accept one connection, send a tick, then read until close.
        std::thread::spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
            rt.block_on(async move {
                let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
                port_tx.send(listener.local_addr().unwrap().port()).unwrap();
                if let Ok((stream, _)) = listener.accept().await {
                    let mut server = tokio_tungstenite::accept_async(stream).await.unwrap();
                    server.send(Message::Binary(ltp_frame(408065, 15075))).await.unwrap();
                    // Drain until the client goes away.
                    while let Some(Ok(msg)) = server.next().await {
                        if msg.is_close() { break; }
                    }
                }
            });
        });

        let port = port_rx.recv().unwrap();
        let url = format!("ws://127.0.0.1:{}", port);

        let captured: Arc<Mutex<Vec<Tick>>> = Arc::new(Mutex::new(Vec::new()));
        struct H { out: Arc<Mutex<Vec<Tick>>> }
        impl KiteTickerHandler for H {
            fn on_ticks<T>(&mut self, _ws: &mut WebSocketHandler<T>, tick: Vec<Tick>)
            where T: KiteTickerHandler {
                self.out.lock().unwrap().extend(tick);
            }
        }

        let mut ticker = KiteTicker::new("<API-KEY>", "<ACCESS-TOKEN>");
        ticker.set_reconnect(false, 0, 1, 1);
        ticker.connect(H { out: captured.clone() }, Some(&url)).unwrap();

        // Poll for the tick to arrive (up to ~2s).
        let mut got = false;
        for _ in 0..40 {
            if !captured.lock().unwrap().is_empty() { got = true; break; }
            std::thread::sleep(Duration::from_millis(50));
        }
        assert!(got, "client did not receive the pushed tick");
        let ticks = captured.lock().unwrap();
        assert_eq!(ticks[0].mode, Mode::Ltp);
        assert_eq!(ticks[0].instrument_token, 408065);
        assert_eq!(ticks[0].last_price, 150.75);
    }

    // ---- binary protocol builders -------------------------------------------

    /// Wrap packet bodies into a full frame: `[i16 count][i16 len][body]...`.
    fn frame(packets: &[Vec<u8>]) -> Vec<u8> {
        let mut buf = Vec::new();
        buf.extend_from_slice(&(packets.len() as i16).to_be_bytes());
        for p in packets {
            buf.extend_from_slice(&(p.len() as i16).to_be_bytes());
            buf.extend_from_slice(p);
        }
        buf
    }

    /// A packet body: a u32 token followed by big-endian i32 fields.
    fn body_i32s(token: u32, vals: &[i32]) -> Vec<u8> {
        let mut b = Vec::new();
        b.extend_from_slice(&token.to_be_bytes());
        for v in vals {
            b.extend_from_slice(&v.to_be_bytes());
        }
        b
    }

    /// A 184-byte full-mode body: 15 i32 quote/OI fields + 10 depth entries
    /// (`i32 qty, i32 price, i16 orders, i16 padding`).
    fn full_body(token: u32, fields: &[i32; 15], depth: &[(i32, i32, i16); 10]) -> Vec<u8> {
        let mut b = body_i32s(token, fields);
        for &(qty, price, orders) in depth.iter() {
            b.extend_from_slice(&qty.to_be_bytes());
            b.extend_from_slice(&price.to_be_bytes());
            b.extend_from_slice(&orders.to_be_bytes());
            b.extend_from_slice(&0i16.to_be_bytes()); // padding
        }
        b
    }

    // ---- parser coverage ----------------------------------------------------

    /// 32-byte index-full packet: non-tradable, computed change, and — critically
    /// — the exchange timestamp read from bytes 28-32, *after* the price-change
    /// field at 24-28 (a distinct sentinel proves the correct offset is used).
    #[test]
    fn test_parse_index_full_packet() {
        // token 256265 has segment (token & 0xFF) == 9 -> index, non-tradable.
        // fields: ltp, high, low, open, close, price_change(ignored), timestamp
        let body = body_i32s(256265, &[1_750_000, 1_760_000, 1_390_000, 1_745_000,
                                        1_400_000, 999, 1_700_000_000]);
        let ticks = parse_binary_message(&frame(&[body]));

        assert_eq!(ticks.len(), 1);
        let t = &ticks[0];
        assert_eq!(t.mode, Mode::Full);
        assert!(!t.tradable);
        assert_eq!(t.instrument_token, 256265);
        assert_eq!(t.last_price, 17500.0);
        let ohlc = t.ohlc.as_ref().unwrap();
        assert_eq!(ohlc.high, 17600.0);
        assert_eq!(ohlc.low, 13900.0);
        assert_eq!(ohlc.open, 17450.0);
        assert_eq!(ohlc.close, 14000.0);
        // (17500 - 14000) * 100 / 14000 = 25.0
        assert_eq!(t.change, Some(25.0));
        // Must be the real timestamp (1_700_000_000), NOT the price_change (999).
        assert_eq!(t.timestamp, Some(1_700_000_000));
        // Index ticks carry no volume/depth.
        assert_eq!(t.volume, None);
        assert!(t.depth.is_none());
    }

    /// 184-byte full-mode tradable packet: OI, timestamps, and 5+5 market depth.
    #[test]
    fn test_parse_full_packet_with_depth() {
        let fields = [
            15075,        // last_price -> 150.75
            10,           // last_quantity
            15000,        // average_price -> 150.00
            100_000,      // volume
            500,          // buy_quantity
            600,          // sell_quantity
            14900,        // open -> 149.00
            15200,        // high -> 152.00
            14800,        // low  -> 148.00
            14950,        // close -> 149.50
            1_699_999_999, // last_trade_time
            12345,        // oi
            20000,        // oi_day_high
            8000,         // oi_day_low
            1_700_000_000, // timestamp
        ];
        let depth = [
            (100, 15070, 1), (90, 15060, 2), (80, 15050, 3), (70, 15040, 4), (60, 15030, 5), // buy
            (110, 15080, 6), (120, 15090, 7), (130, 15100, 8), (140, 15110, 9), (150, 15120, 10), // sell
        ];
        let ticks = parse_binary_message(&frame(&[full_body(408065, &fields, &depth)]));

        assert_eq!(ticks.len(), 1);
        let t = &ticks[0];
        assert_eq!(t.mode, Mode::Full);
        assert!(t.tradable);
        assert_eq!(t.last_price, 150.75);
        assert_eq!(t.average_price, Some(150.0));
        assert_eq!(t.volume, Some(100_000));
        assert_eq!(t.buy_quantity, Some(500));
        assert_eq!(t.sell_quantity, Some(600));
        assert_eq!(t.oi, Some(12345));
        assert_eq!(t.oi_day_high, Some(20000));
        assert_eq!(t.oi_day_low, Some(8000));
        assert_eq!(t.last_trade_time, Some(1_699_999_999));
        assert_eq!(t.timestamp, Some(1_700_000_000));

        let d = t.depth.as_ref().unwrap();
        assert_eq!(d.buy.len(), 5);
        assert_eq!(d.sell.len(), 5);
        assert_eq!(d.buy[0].quantity, 100);
        assert_eq!(d.buy[0].price, 150.70);
        assert_eq!(d.buy[0].orders, 1);
        assert_eq!(d.sell[0].quantity, 110);
        assert_eq!(d.sell[0].price, 150.80);
        assert_eq!(d.sell[0].orders, 6);
    }

    /// CDS instruments (segment 3) scale prices by 10^7, not 100.
    #[test]
    fn test_parse_cds_price_divisor() {
        // token 4099 -> segment (4099 & 0xFF) == 3 (NSE currency).
        let body = body_i32s(4099, &[835_000_000]); // 83.5 with a 10^7 divisor
        let ticks = parse_binary_message(&frame(&[body]));
        assert_eq!(ticks.len(), 1);
        assert_eq!(ticks[0].mode, Mode::Ltp);
        assert_eq!(ticks[0].last_price, 83.5);
    }

    /// An unknown packet length is skipped without derailing later packets.
    #[test]
    fn test_unknown_packet_length_is_skipped() {
        let unknown = body_i32s(408065, &[0, 0, 0]); // 16-byte body, no matching arm
        let ltp = body_i32s(408065, &[15075]);       // 8-byte LTP
        let ticks = parse_binary_message(&frame(&[unknown, ltp]));
        assert_eq!(ticks.len(), 1);
        assert_eq!(ticks[0].mode, Mode::Ltp);
        assert_eq!(ticks[0].last_price, 150.75);
    }

    /// Empty and zero-packet frames yield no ticks and never panic.
    #[test]
    fn test_empty_and_zero_packet_frames() {
        assert!(parse_binary_message(&[]).is_empty());
        assert!(parse_binary_message(&[0x00, 0x00]).is_empty());        // len <= 2
        assert!(parse_binary_message(&[0x00, 0x00, 0x00]).is_empty());  // 0 packets
    }

    // ---- outbound commands --------------------------------------------------

    /// subscribe / unsubscribe / set_mode emit the correct JSON frames and keep
    /// the shared subscription map in sync.
    #[test]
    fn test_commands_emit_expected_json() {
        let (mut ws, mut rx) = test_handler();

        ws.subscribe(vec![408065, 5633]).unwrap();
        let sub: JsonValue = serde_json::from_str(&rx.try_recv().unwrap()).unwrap();
        assert_eq!(sub, json!({"a": "subscribe", "v": [408065, 5633]}));

        ws.set_mode("full", vec![408065]).unwrap();
        let mode: JsonValue = serde_json::from_str(&rx.try_recv().unwrap()).unwrap();
        assert_eq!(mode, json!({"a": "mode", "v": ["full", [408065]]}));

        ws.unsubscribe(vec![5633]).unwrap();
        let unsub: JsonValue = serde_json::from_str(&rx.try_recv().unwrap()).unwrap();
        assert_eq!(unsub, json!({"a": "unsubscribe", "v": [5633]}));

        let map = ws.subscribed_tokens.lock().unwrap();
        assert_eq!(map.get(&408065), Some(&"full".to_string())); // mode updated
        assert!(!map.contains_key(&5633));                       // removed
    }

    // ---- text (postback) routing --------------------------------------------

    struct RecordingHandler {
        events: Arc<Mutex<Vec<(String, JsonValue)>>>,
    }
    impl KiteTickerHandler for RecordingHandler {
        fn on_order_update<T>(&mut self, _ws: &mut WebSocketHandler<T>, order: JsonValue)
        where T: KiteTickerHandler {
            self.events.lock().unwrap().push(("order".into(), order));
        }
        fn on_message<T>(&mut self, _ws: &mut WebSocketHandler<T>, message: JsonValue)
        where T: KiteTickerHandler {
            self.events.lock().unwrap().push(("message".into(), message));
        }
        fn on_error<T>(&mut self, _ws: &mut WebSocketHandler<T>)
        where T: KiteTickerHandler {
            self.events.lock().unwrap().push(("error".into(), JsonValue::Null));
        }
    }

    #[test]
    fn test_dispatch_text_routes_to_callbacks() {
        let events = Arc::new(Mutex::new(Vec::new()));
        let (tx, _rx) = mpsc::unbounded_channel::<String>();
        let mut ws = WebSocketHandler {
            handler: Arc::new(Mutex::new(Box::new(RecordingHandler { events: events.clone() }))),
            cmd_tx: tx,
            subscribed_tokens: Arc::new(Mutex::new(HashMap::new())),
        };

        dispatch_text(r#"{"type":"order","data":{"order_id":"123","status":"COMPLETE"}}"#, &mut ws);
        dispatch_text(r#"{"type":"message","data":{"msg":"hello"}}"#, &mut ws);
        dispatch_text(r#"{"type":"error","data":"boom"}"#, &mut ws);
        dispatch_text(r#"{"type":"unknown","data":{}}"#, &mut ws); // ignored
        dispatch_text("not json at all", &mut ws);                 // ignored, no panic

        let events = events.lock().unwrap();
        assert_eq!(events.len(), 3);
        assert_eq!(events[0].0, "order");
        assert_eq!(events[0].1["order_id"], "123");
        assert_eq!(events[1].0, "message");
        assert_eq!(events[1].1["msg"], "hello");
        assert_eq!(events[2].0, "error");
    }

    // ---- serialization ------------------------------------------------------

    /// A Tick round-trips to JSON and omits fields that are `None` for its mode.
    #[test]
    fn test_ltp_tick_serializes_without_absent_fields() {
        let ticks = parse_binary_message(&ltp_frame(408065, 15075));
        let v = serde_json::to_value(&ticks[0]).unwrap();
        assert_eq!(v["mode"], "ltp");
        assert_eq!(v["instrument_token"], 408065);
        assert_eq!(v["last_price"], 150.75);
        assert!(v.get("volume").is_none());
        assert!(v.get("ohlc").is_none());
        assert!(v.get("depth").is_none());
        assert!(v.get("change").is_none());
    }
}