ibapi 5.5.0

use super::*;
use crate::common::test_utils::helpers;
use crate::common::test_utils::helpers::assert_rejects_text_framing;
use crate::common::test_utils::helpers::{binary_text, proto_error_response, proto_response, CapturingSink};
use crate::messages::IncomingMessages;
use crate::messages::{HANDSHAKE_DECODE_FAILURE_CODE, HANDSHAKE_UNKNOWN_FRAME_CODE, MESSAGE_ID_LEN, UNKNOWN_MESSAGE_TYPE_CODE};
use crate::testdata::builders::orders::next_valid_id;
use crate::testdata::builders::ResponseProtoEncoder;
use std::sync::{Arc, Mutex};
use time::macros::datetime;
use time_tz::TimeZone;

const CLIENT_ID: i32 = 100;

/// Test sink that drops every notice. Used when the test cares about the
/// startup callback, not the notice fan-out.
#[derive(Default)]
struct DiscardingSink;
impl NoticeSink for DiscardingSink {
    fn deliver(&self, _: Notice) {}
}

fn empty_ctx<'a>() -> StartupHandshakeContext<'a> {
    static SINK: DiscardingSink = DiscardingSink;
    StartupHandshakeContext {
        client_id: CLIENT_ID,
        startup: None,
        notice_sink: &SINK,
    }
}

fn startup_ctx<'a>(cb: &'a (dyn Fn(StartupMessage) + Send + Sync)) -> StartupHandshakeContext<'a> {
    static SINK: DiscardingSink = DiscardingSink;
    StartupHandshakeContext {
        client_id: CLIENT_ID,
        startup: Some(cb),
        notice_sink: &SINK,
    }
}

fn notice_sink_ctx(sink: &CapturingSink) -> StartupHandshakeContext<'_> {
    StartupHandshakeContext {
        client_id: CLIENT_ID,
        startup: None,
        notice_sink: sink,
    }
}

/// Context with both a startup callback AND a capturing notice sink. Used to
/// verify decode-failure routing (callback should NOT fire; sink should
/// receive the synthesized notice).
fn full_ctx<'a>(cb: &'a (dyn Fn(StartupMessage) + Send + Sync), sink: &'a CapturingSink) -> StartupHandshakeContext<'a> {
    StartupHandshakeContext {
        client_id: CLIENT_ID,
        startup: Some(cb),
        notice_sink: sink,
    }
}

#[test]
fn dispatch_unsolicited_open_order_decode_failure_emits_notice() {
    // Sparse text-framed OpenOrder — decoder calls proto() and rejects.
    let message = ResponseMessage::from("5\0123\0AAPL\0STK\0");

    let cb_fired = Arc::new(Mutex::new(false));
    let cb_fired_clone = cb_fired.clone();
    let cb = move |_msg: StartupMessage| *cb_fired_clone.lock().unwrap() = true;
    let sink = CapturingSink::default();

    dispatch_unsolicited_message(&message, &full_ctx(&cb, &sink));

    assert!(!*cb_fired.lock().unwrap(), "callback must not fire on decode failure");
    let notice = sink.last().expect("notice sink should receive decode-failure notice");
    assert_eq!(notice.code, HANDSHAKE_DECODE_FAILURE_CODE);
    assert!(
        notice.message.contains("OpenOrder"),
        "notice message should name the kind: {}",
        notice.message
    );
}

#[test]
fn dispatch_unsolicited_order_status_decode_failure_emits_notice() {
    let message = ResponseMessage::from("3\0456\0Filled\0100\0");

    let cb_fired = Arc::new(Mutex::new(false));
    let cb_fired_clone = cb_fired.clone();
    let cb = move |_msg: StartupMessage| *cb_fired_clone.lock().unwrap() = true;
    let sink = CapturingSink::default();

    dispatch_unsolicited_message(&message, &full_ctx(&cb, &sink));

    assert!(!*cb_fired.lock().unwrap(), "callback must not fire on decode failure");
    let notice = sink.last().expect("notice sink should receive decode-failure notice");
    assert_eq!(notice.code, HANDSHAKE_DECODE_FAILURE_CODE);
    assert!(notice.message.contains("OrderStatus"), "notice should name the kind: {}", notice.message);
}

#[test]
fn dispatch_unsolicited_account_value_typed() {
    use crate::messages::IncomingMessages;
    use crate::testdata::builders::accounts::account_value;
    use crate::testdata::builders::ResponseProtoEncoder;

    let message = proto_response(
        IncomingMessages::AccountValue,
        account_value()
            .key("NetLiquidation")
            .value("123456.78")
            .currency("USD")
            .account("DU1234567")
            .encode_proto(),
    );

    let captured: Arc<Mutex<Option<crate::accounts::AccountValue>>> = Arc::new(Mutex::new(None));
    let captured_clone = captured.clone();
    let cb = move |msg: StartupMessage| {
        if let StartupMessage::AccountUpdate(crate::accounts::AccountUpdate::AccountValue(av)) = msg {
            *captured.lock().unwrap() = Some(av);
        } else {
            panic!("expected AccountUpdate::AccountValue, got {msg:?}");
        }
    };

    dispatch_unsolicited_message(&message, &startup_ctx(&cb));
    let got = captured_clone.lock().unwrap().take().expect("callback didn't fire");
    assert_eq!(got.key, "NetLiquidation");
    assert_eq!(got.value, "123456.78");
    assert_eq!(got.currency, "USD");
    assert_eq!(got.account.as_deref(), Some("DU1234567"));
}

#[test]
fn dispatch_unsolicited_open_order_end_typed() {
    // OpenOrderEnd: msg_type=53, version=1
    let message = ResponseMessage::from("53\01\0");

    let captured: Arc<Mutex<bool>> = Arc::new(Mutex::new(false));
    let captured_clone = captured.clone();
    let cb = move |msg: StartupMessage| {
        if matches!(msg, StartupMessage::OpenOrderEnd) {
            *captured.lock().unwrap() = true;
        } else {
            panic!("expected OpenOrderEnd, got {msg:?}");
        }
    };

    dispatch_unsolicited_message(&message, &startup_ctx(&cb));
    assert!(*captured_clone.lock().unwrap(), "OpenOrderEnd not delivered as typed variant");
}

#[test]
fn dispatch_unsolicited_account_download_end_typed() {
    // AccountDownloadEnd: msg_type=54, version=1, account
    let message = ResponseMessage::from("54\01\0DU1234567\0");

    let captured: Arc<Mutex<bool>> = Arc::new(Mutex::new(false));
    let captured_clone = captured.clone();
    let cb = move |msg: StartupMessage| {
        if matches!(msg, StartupMessage::AccountUpdate(crate::accounts::AccountUpdate::End)) {
            *captured.lock().unwrap() = true;
        }
    };

    dispatch_unsolicited_message(&message, &startup_ctx(&cb));
    assert!(*captured_clone.lock().unwrap(), "End variant not delivered");
}

#[test]
fn dispatch_unsolicited_unknown_emits_notice() {
    // NewsBulletins (msg_type=14) — no typed variant in StartupMessage.
    // Catch-all fires the notice sink with HANDSHAKE_UNKNOWN_FRAME_CODE.
    let message = ResponseMessage::from("14\0\0");

    let cb_fired = Arc::new(Mutex::new(false));
    let cb_fired_clone = cb_fired.clone();
    let cb = move |_msg: StartupMessage| *cb_fired_clone.lock().unwrap() = true;
    let sink = CapturingSink::default();

    dispatch_unsolicited_message(&message, &full_ctx(&cb, &sink));

    assert!(!*cb_fired.lock().unwrap(), "callback must not fire for unknown handshake frame");
    let notice = sink.last().expect("notice sink should receive unknown-frame notice");
    assert_eq!(notice.code, HANDSHAKE_UNKNOWN_FRAME_CODE);
    assert!(
        notice.message.contains("NewsBulletins"),
        "notice should name the kind: {}",
        notice.message
    );
    assert!(notice.is_handshake_synthetic());
}

#[test]
fn dispatch_unsolicited_notice_warning_invokes_notice_sink() {
    // Error frame with code 2104 (farm-status warning), proto-framed.
    let message = proto_error_response(-1, 2104, "Market data farm OK");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));

    let got = sink.last().expect("notice sink didn't receive notice");
    assert_eq!(got.code, 2104);
    assert_eq!(got.message, "Market data farm OK");
}

/// 326 ends the handshake with `ClientIdInUse`; any other handshake error
/// is only a notice. Both reach the notice sink.
#[test]
fn parse_account_info_error_frames() {
    // (name, code, client id in use)
    let cases = [("client id in use", 326, true), ("farm warning", 2104, false), ("hard error", 504, false)];

    for (name, code, in_use) in cases {
        let message = proto_error_response(-1, code, "msg");
        let sink = CapturingSink::default();

        let result = ConnectionHandler::default().parse_account_info(&message, &notice_sink_ctx(&sink));

        match result {
            Err(Error::ClientIdInUse { client_id }) => {
                assert!(in_use, "{name}: unexpected ClientIdInUse");
                assert_eq!(client_id, CLIENT_ID, "{name}");
            }
            Ok(info) => {
                assert!(!in_use, "{name}: expected ClientIdInUse");
                assert_eq!(info.next_order_id, None, "{name}");
            }
            Err(e) => panic!("{name}: unexpected error {e:?}"),
        }
        assert_eq!(sink.last().map(|n| n.code), Some(code), "{name}: notice still delivered");
    }
}

#[test]
fn dispatch_unsolicited_notice_hard_error_invokes_notice_sink() {
    // Error frame with code 504 — non-warning, non-system message.
    let message = proto_error_response(-1, 504, "Not connected");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));

    assert_eq!(sink.last().expect("sink missed").code, 504);
}

#[test]
fn dispatch_unsolicited_notice_only_fires_notice_sink() {
    // Error frame should fire the notice sink, NOT the startup callback.
    let message = proto_error_response(-1, 2104, "farm OK");

    let startup_fired = Arc::new(Mutex::new(false));
    let startup_fired_clone = startup_fired.clone();
    let startup_cb = move |_msg: StartupMessage| {
        *startup_fired_clone.lock().unwrap() = true;
    };
    let sink = CapturingSink::default();

    dispatch_unsolicited_message(
        &message,
        &StartupHandshakeContext {
            client_id: CLIENT_ID,
            startup: Some(&startup_cb),
            notice_sink: &sink,
        },
    );
    assert!(!*startup_fired.lock().unwrap(), "startup callback should not fire on Error");
    assert_eq!(sink.count(), 1, "notice sink should receive exactly one notice on Error");
}

#[test]
fn parse_account_info_callback_not_invoked_for_next_valid_id() {
    // NextValidId is consumed internally — neither callback fires.
    let handler = ConnectionHandler::default();
    let bytes = next_valid_id().order_id(1000).encode_proto();
    let message = proto_response(IncomingMessages::NextValidId, bytes);

    let fired = Arc::new(Mutex::new(false));
    let fired_clone = fired.clone();
    let cb = move |_: StartupMessage| {
        *fired_clone.lock().unwrap() = true;
    };

    let result = handler.parse_account_info(&message, &startup_ctx(&cb));
    assert!(result.is_ok());
    assert!(!*fired.lock().unwrap(), "callback should NOT be invoked for NextValidId");
}

#[test]
fn parse_account_info_callback_not_invoked_for_managed_accounts() {
    use prost::Message;

    let handler = ConnectionHandler::default();
    let bytes = crate::proto::ManagedAccounts {
        accounts_list: Some("DU123".to_string()),
    }
    .encode_to_vec();
    let message = proto_response(IncomingMessages::ManagedAccounts, bytes);

    let fired = Arc::new(Mutex::new(false));
    let fired_clone = fired.clone();
    let cb = move |_: StartupMessage| {
        *fired_clone.lock().unwrap() = true;
    };

    let result = handler.parse_account_info(&message, &startup_ctx(&cb));
    assert!(result.is_ok());
    assert!(!*fired.lock().unwrap(), "callback should NOT be invoked for ManagedAccounts");
}

#[test]
fn parse_account_info_multiple_messages_callback() {
    let handler = ConnectionHandler::default();
    let count = Arc::new(Mutex::new(0));
    let count_clone = count.clone();
    let cb = move |_: StartupMessage| {
        *count_clone.lock().unwrap() += 1;
    };
    let cbs = startup_ctx(&cb);

    // First message: OpenOrderEnd (unit marker — no decoder, callback fires).
    let msg1 = ResponseMessage::from("53\01\0");
    handler.parse_account_info(&msg1, &cbs).unwrap();

    // Second message: CompletedOrdersEnd (unit marker — callback fires).
    let msg2 = ResponseMessage::from("102\0");
    handler.parse_account_info(&msg2, &cbs).unwrap();

    // Third message: NextValidId (consumed internally — should NOT trigger callback)
    let bytes = next_valid_id().order_id(1000).encode_proto();
    let msg3 = proto_response(IncomingMessages::NextValidId, bytes);
    handler.parse_account_info(&msg3, &cbs).unwrap();

    assert_eq!(*count.lock().unwrap(), 2, "callback should be invoked exactly twice");
}

#[test]
fn parse_account_info_next_valid_id_rejects_text_framing() {
    // The handshake is the one reader still advancing a cursor, so it takes
    // `&mut` where every decoder takes `&` — hence the clone rather than a
    // direct hand-off to the shared assertion.
    let handler = ConnectionHandler::default();
    assert_rejects_text_framing(IncomingMessages::NextValidId, "9\01\01000\0", |message| {
        handler.parse_account_info(&message.clone(), &empty_ctx())
    });
}

#[test]
fn parse_account_info_managed_accounts_rejects_text_framing() {
    let handler = ConnectionHandler::default();
    assert_rejects_text_framing(IncomingMessages::ManagedAccounts, "15\01\0DU123,DU456\0", |message| {
        handler.parse_account_info(&message.clone(), &empty_ctx())
    });
}

#[test]
fn require_protobuf_support_accepts_minimum() {
    require_protobuf_support(server_versions::PROTOBUF_REST_MESSAGES_3).expect("floor version must be accepted");
}

#[test]
fn require_protobuf_support_accepts_newer() {
    require_protobuf_support(server_versions::PROTOBUF_REST_MESSAGES_3 + 5).expect("newer versions must be accepted");
}

#[test]
fn require_protobuf_support_rejects_older() {
    let actual = server_versions::PROTOBUF_REST_MESSAGES_3 - 1;
    let err = require_protobuf_support(actual).expect_err("older versions must be rejected");

    match &err {
        Error::ServerVersion(required, got, msg) => {
            assert_eq!(*required, server_versions::PROTOBUF_REST_MESSAGES_3);
            assert_eq!(*got, actual);
            assert!(msg.contains("protobuf"), "message should mention protobuf: {msg}");
            assert!(msg.contains("upgrade"), "message should tell user to upgrade: {msg}");
        }
        other => panic!("expected Error::ServerVersion, got {other:?}"),
    }

    let rendered = err.to_string();
    let expected_required = format!("server version {} required", server_versions::PROTOBUF_REST_MESSAGES_3);
    assert!(rendered.contains(&expected_required), "rendered: {rendered}");
    assert!(rendered.contains(&actual.to_string()), "rendered: {rendered}");
}

#[test]
fn require_protobuf_support_rejects_previous_scan_data_floor() {
    // Servers in [210, 212] don't yet emit the REST_MESSAGES_1/2 families
    // (ExerciseOptions, calc, FA, SecDefOpt, FamilyCodes, MarketRule, …) in
    // protobuf and we no longer carry text-decode fallback, so the floor
    // must reject them.
    let previous_floor = server_versions::PROTOBUF_SCAN_DATA;
    let err = require_protobuf_support(previous_floor).expect_err("previous floor must now be rejected");
    match err {
        Error::ServerVersion(required, got, _) => {
            assert_eq!(required, server_versions::PROTOBUF_REST_MESSAGES_3);
            assert_eq!(got, previous_floor);
        }
        other => panic!("expected Error::ServerVersion, got {other:?}"),
    }
}

#[test]
fn parse_connection_time_works() {
    // TWS sends the abbreviation in force on the date: PST in winter, PDT in summer.
    let (connection_time, timezone) = parse_connection_time("20230105 22:20:39 PST");
    assert_eq!(connection_time, Some(datetime!(2023-01-05 22:20:39 -08:00)));
    assert_eq!(timezone.map(|tz| tz.name()), Some("America/Los_Angeles"));

    let (connection_time, timezone) = parse_connection_time("20230405 22:20:39 PDT");
    assert_eq!(connection_time, Some(datetime!(2023-04-05 22:20:39 -07:00)));
    assert_eq!(timezone.map(|tz| tz.name()), Some("America/Los_Angeles"));
}

#[test]
fn parse_connection_time_china_standard_time() {
    let example = "20230405 22:20:39 China Standard Time";
    let (connection_time, timezone) = parse_connection_time(example);

    assert!(connection_time.is_some());
    assert!(timezone.is_some());
    assert_eq!(timezone.unwrap().name(), "Asia/Shanghai");
}

#[test]
fn parse_connection_time_brasilia_standard_time() {
    // As sent by a pt-BR 10.51 gateway (#964). Brazil has no DST since 2019.
    let (connection_time, timezone) = parse_connection_time("20261009 10:00:00 Brasilia Standard Time");
    assert_eq!(connection_time, Some(datetime!(2026-10-09 10:00:00 -03:00)));
    assert_eq!(timezone.map(|tz| tz.name()), Some("America/Sao_Paulo"));
}

#[test]
fn parse_connection_time_chinese_utf8() {
    let example = "20230405 22:20:39 中国标准时间";
    let (connection_time, timezone) = parse_connection_time(example);

    assert!(connection_time.is_some());
    assert!(timezone.is_some());
    assert_eq!(timezone.unwrap().name(), "Asia/Shanghai");
}

#[test]
fn parse_connection_time_mojibake() {
    // Simulate GB2312 timezone decoded as UTF-8 lossy
    let example = "20230405 22:20:39 \u{FFFD}\u{FFFD}\u{FFFD}";
    let (connection_time, timezone) = parse_connection_time(example);

    assert!(connection_time.is_some());
    assert!(timezone.is_some());
    assert_eq!(timezone.unwrap().name(), "Asia/Shanghai");
}

#[test]
fn parse_connection_time_unknown_timezone_is_soft() {
    // Only `Client::time_zone` and `connection_time` depend on the zone, so an
    // unmatched name leaves them unset instead of failing the handshake. A
    // partial name is not guessed at either (#809).
    for name in ["Bogus Standard Time", "York"] {
        let (time, tz) = parse_connection_time(&format!("20230405 22:20:39 {name}"));
        assert!(time.is_none(), "{name}");
        assert!(tz.is_none(), "{name}");
    }
}

#[test]
fn parse_connection_time_in_dst_fold_resolves() {
    // Connecting during a fall-back hour: the reading is ambiguous. It used to
    // log "Error setting timezone" and yield None; it now takes the earlier
    // occurrence like every other wall-clock reading from TWS.
    let (time, tz) = parse_connection_time("20251102 01:30:00 US/Eastern");
    assert_eq!(time, Some(datetime!(2025-11-02 05:30:00 UTC)));
    assert!(tz.is_some());
}

#[test]
fn parse_connection_time_short_input_still_ok() {
    // Truncated wire data — preserve current tolerance, no error.
    let (time, tz) = parse_connection_time("20230405");
    assert!(time.is_none());
    assert!(tz.is_none());
}

#[test]
fn parse_connection_time_unparseable_date_still_ok() {
    // Timezone resolves; only the wall-clock fails. Preserve tolerance.
    let (time, tz) = parse_connection_time("BADDATE 99:99:99 PST");
    assert!(time.is_none());
    assert!(tz.is_some());
}

#[test]
fn connection_handler_handshake() {
    let handler = ConnectionHandler::default();
    let handshake = handler.format_handshake();

    // Should start with "API\0"
    assert_eq!(&handshake[0..4], b"API\0");

    // Should contain version string
    let version_part = &handshake[4..];
    assert!(!version_part.is_empty());
}

#[test]
fn connection_handler_start_api() {
    use crate::messages::PROTOBUF_MSG_ID;

    let handler = ConnectionHandler::default();
    let data = handler.format_start_api(123);

    // First 4 bytes: big-endian (StartApi=71 + 200)
    let msg_id = i32::from_be_bytes([data[0], data[1], data[2], data[3]]);
    assert_eq!(msg_id, 71 + PROTOBUF_MSG_ID);

    // Remaining bytes: protobuf-encoded StartApiRequest with client_id=123
    let request: crate::proto::StartApiRequest = prost::Message::decode(&data[4..]).unwrap();
    assert_eq!(request.client_id, Some(123));
}

#[test]
fn connection_handler_start_api_protobuf() {
    let handler = ConnectionHandler::default();
    let data = handler.format_start_api(123);

    // First 4 bytes: msg_id (71 + 200 = 271)
    let msg_id = i32::from_be_bytes([data[0], data[1], data[2], data[3]]);
    assert_eq!(msg_id, 271);

    // Decode protobuf payload
    use prost::Message;
    let req = crate::proto::StartApiRequest::decode(&data[4..]).unwrap();
    assert_eq!(req.client_id, Some(123));
}

#[test]
fn parse_raw_message_protobuf() {
    use crate::messages::PROTOBUF_MSG_ID;

    // Simulate a protobuf message: msg_id = 5 + 200 = 205, then some payload
    let msg_id: i32 = 5 + PROTOBUF_MSG_ID;
    let payload = vec![0x08, 0x64]; // varint tag=1, value=100
    let mut data = msg_id.to_be_bytes().to_vec();
    data.extend_from_slice(&payload);

    let message = parse_raw_message(&data).expect("well-formed protobuf frame");
    assert!(message.raw_bytes().is_some(), "protobuf framing must populate raw_bytes");
    assert_eq!(message.message_type(), IncomingMessages::OpenOrder);
    assert_eq!(message.raw_bytes(), Some(payload.as_slice()));
}

#[test]
fn parse_raw_message_binary_id_text_payload() {
    // Simulate a text message at server >= 201: binary msg_id=9, then NUL-delimited text
    let data = binary_text(IncomingMessages::NextValidId as i32, "1\01000\0");

    let message = parse_raw_message(&data).expect("well-formed text frame");
    assert!(message.raw_bytes().is_none(), "text framing must leave raw_bytes empty");
    assert_eq!(message.message_type(), IncomingMessages::NextValidId);
    assert_eq!(message.fields[1], "1"); // version field
    assert_eq!(message.peek_int(2).unwrap(), 1000); // next_order_id
}

/// A body too short to hold the 4-byte message id must be reported, not
/// indexed past. This used to panic the dispatcher outright ("index out of
/// bounds: the len is 0 but the index is 0") — in sync it killed the
/// dispatcher thread, in async the dispatcher task.
#[test]
fn parse_raw_message_rejects_body_shorter_than_message_id() {
    for length in 0..MESSAGE_ID_LEN {
        let data = vec![0_u8; length];
        let err = parse_raw_message(&data).expect_err("a body too short for a message id must be rejected");
        assert!(matches!(err, Error::InvalidFrame(_)), "short body must raise InvalidFrame, got {err:?}");
    }

    // A bare message id with no payload is the smallest legal frame.
    let message = parse_raw_message(&(IncomingMessages::NextValidId as i32).to_be_bytes()).expect("bare message id is a legal frame");
    assert_eq!(message.message_type(), IncomingMessages::NextValidId);
}

/// An unrecognized id survives parsing on both framings. `IncomingMessages`
/// collapses every unknown value to the single `NotValid` variant, so without
/// this the id an unroutable-frame diagnostic needs is gone by the time routing
/// sees the message — and the protobuf path has no `fields[0]` to fall back on.
#[test]
fn parse_raw_message_retains_an_unrecognized_message_id() {
    // Protobuf framing: wire id carries the PROTOBUF_MSG_ID offset, and
    // message_id() reports the value actually looked up.
    let message = parse_raw_message(&helpers::unknown_message_frame()).expect("well-formed frame");
    assert_eq!(message.message_type(), IncomingMessages::NotValid);
    assert_eq!(message.message_id(), Some(helpers::UNKNOWN_MESSAGE_ID));

    // Text framing: same accessor, id taken from fields[0]. Must be at or below
    // PROTOBUF_MSG_ID to reach the text branch at all, so it cannot reuse
    // UNKNOWN_MESSAGE_ID.
    let text_frame = binary_text(150, "1\0");
    let message = parse_raw_message(&text_frame).expect("well-formed frame");
    assert_eq!(message.message_type(), IncomingMessages::NotValid);
    assert_eq!(message.message_id(), Some(150));

    // A recognized id reports itself too — the accessor is not unknown-only.
    let message = parse_raw_message(&(IncomingMessages::NextValidId as i32).to_be_bytes()).expect("bare message id is a legal frame");
    assert_eq!(message.message_type(), IncomingMessages::NextValidId);
    assert_eq!(message.message_id(), Some(IncomingMessages::NextValidId as i32));
}

/// Test handling of non-UTF8 encoded data from IB Gateway (issue #352)
/// Some IB Gateway installations send timezone names in GB2312/GBK encoding
/// (e.g., Chinese "中国标准时间" for "China Standard Time")
#[test]
fn non_utf8_handshake_response() {
    // Actual bytes from issue #352: "173\020251205 23:13:45 中国标准时间\0"
    // where the Chinese characters are GB2312 encoded, not UTF-8
    let gb2312_bytes: Vec<u8> = vec![
        49, 55, 51, 0, // "173\0" - server version
        50, 48, 50, 53, 49, 50, 48, 53, 32, // "20251205 " - date
        50, 51, 58, 49, 51, 58, 52, 53, 32, // "23:13:45 " - time
        214, 208, 185, 250, 177, 234, 215, 188, 202, 177, 188, 228, // GB2312: 中国标准时间
        0,   // null terminator
    ];

    // from_utf8_lossy should handle this without error
    let raw_string = String::from_utf8_lossy(&gb2312_bytes).into_owned();

    // Should contain the ASCII portions intact
    assert!(raw_string.contains("173"));
    assert!(raw_string.contains("20251205"));
    assert!(raw_string.contains("23:13:45"));

    // Non-UTF8 bytes are replaced with replacement character
    assert!(raw_string.contains('\u{FFFD}'));

    // Parse as ResponseMessage and extract handshake data
    let mut response = ResponseMessage::from(&raw_string);
    let handler = ConnectionHandler::default();
    let result = handler.parse_handshake_response(&mut response);

    assert!(result.is_ok());
    let handshake_data = result.unwrap();
    assert_eq!(handshake_data.server_version, 173);
    // server_time will contain replacement characters but parsing succeeds
    assert!(handshake_data.server_time.contains("20251205"));
}

#[test]
fn startup_message_message_type_typed_variants() {
    use crate::accounts::{AccountPortfolioValue, AccountUpdateTime, AccountValue};
    use crate::orders::{CommissionReport, ExecutionData, OrderData, OrderStatus};

    assert_eq!(
        StartupMessage::OpenOrder(OrderData::default()).message_type(),
        IncomingMessages::OpenOrder
    );
    assert_eq!(
        StartupMessage::OrderStatus(OrderStatus::default()).message_type(),
        IncomingMessages::OrderStatus
    );
    assert_eq!(StartupMessage::OpenOrderEnd.message_type(), IncomingMessages::OpenOrderEnd);
    assert_eq!(
        StartupMessage::AccountUpdate(AccountUpdate::AccountValue(AccountValue::default())).message_type(),
        IncomingMessages::AccountValue
    );
    assert_eq!(
        StartupMessage::AccountUpdate(AccountUpdate::PortfolioValue(AccountPortfolioValue::default())).message_type(),
        IncomingMessages::PortfolioValue
    );
    assert_eq!(
        StartupMessage::AccountUpdate(AccountUpdate::UpdateTime(AccountUpdateTime::default())).message_type(),
        IncomingMessages::AccountUpdateTime
    );
    assert_eq!(
        StartupMessage::AccountUpdate(AccountUpdate::End).message_type(),
        IncomingMessages::AccountDownloadEnd
    );
    assert_eq!(
        StartupMessage::Execution(ExecutionData::default()).message_type(),
        IncomingMessages::ExecutionData
    );
    assert_eq!(
        StartupMessage::CommissionReport(CommissionReport::default()).message_type(),
        IncomingMessages::CommissionsReport
    );
    assert_eq!(
        StartupMessage::CompletedOrder(OrderData::default()).message_type(),
        IncomingMessages::CompletedOrder
    );
    assert_eq!(StartupMessage::ExecutionDataEnd.message_type(), IncomingMessages::ExecutionDataEnd);
    assert_eq!(StartupMessage::CompletedOrdersEnd.message_type(), IncomingMessages::CompletedOrdersEnd);
}

#[test]
fn parse_account_info_next_valid_id_protobuf() {
    let handler = ConnectionHandler::default();
    let bytes = next_valid_id().order_id(4242).encode_proto();
    let message = proto_response(IncomingMessages::NextValidId, bytes);

    let info = handler
        .parse_account_info(&message, &empty_ctx())
        .expect("protobuf NextValidId must parse");
    assert_eq!(info.next_order_id, Some(4242));
    assert_eq!(info.managed_accounts, None);
}

#[test]
fn parse_account_info_managed_accounts_protobuf() {
    use prost::Message;

    let handler = ConnectionHandler::default();
    let proto = crate::proto::ManagedAccounts {
        accounts_list: Some("DU111,DU222".to_string()),
    };
    let bytes = proto.encode_to_vec();
    let message = proto_response(IncomingMessages::ManagedAccounts, bytes);

    let info = handler
        .parse_account_info(&message, &empty_ctx())
        .expect("protobuf ManagedAccounts must parse");
    assert_eq!(info.next_order_id, None);
    assert_eq!(info.managed_accounts, Some("DU111,DU222".to_string()));
}

#[test]
fn parse_account_info_next_valid_id_protobuf_decode_error() {
    // Garbage bytes for the NextValidId proto envelope.
    let handler = ConnectionHandler::default();
    let message = proto_response(IncomingMessages::NextValidId, vec![0xff, 0xff, 0xff]);

    let err = handler
        .parse_account_info(&message, &empty_ctx())
        .expect_err("garbage protobuf must error");
    assert!(matches!(err, Error::ProtobufDecode(_)), "got {err:?}");
}

#[test]
fn parse_account_info_managed_accounts_protobuf_decode_error() {
    let handler = ConnectionHandler::default();
    let message = proto_response(IncomingMessages::ManagedAccounts, vec![0xff, 0xff, 0xff]);

    let err = handler
        .parse_account_info(&message, &empty_ctx())
        .expect_err("garbage protobuf must error");
    assert!(matches!(err, Error::ProtobufDecode(_)), "got {err:?}");
}

#[test]
fn dispatch_unsolicited_open_order_no_callback_is_noop() {
    let message = ResponseMessage::from("5\0123\0AAPL\0STK\0");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    assert_eq!(sink.count(), 0, "OpenOrder must not deliver to notice sink");
}

#[test]
fn dispatch_unsolicited_order_status_no_callback_is_noop() {
    let message = ResponseMessage::from("3\0456\0Filled\0100\0");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    assert_eq!(sink.count(), 0);
}

#[test]
fn dispatch_unsolicited_open_order_end_no_callback_is_noop() {
    let message = ResponseMessage::from("53\01\0");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    assert_eq!(sink.count(), 0);
}

#[test]
fn dispatch_unsolicited_account_update_no_callback_is_noop() {
    use crate::messages::IncomingMessages;
    use crate::testdata::builders::accounts::account_value;
    use crate::testdata::builders::ResponseProtoEncoder;

    let message = proto_response(
        IncomingMessages::AccountValue,
        account_value()
            .key("NetLiquidation")
            .value("123.45")
            .currency("USD")
            .account("DU1")
            .encode_proto(),
    );

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    assert_eq!(sink.count(), 0);
}

#[test]
fn dispatch_unsolicited_account_value_decode_failure_emits_notice() {
    use crate::messages::IncomingMessages;

    // Garbage proto bytes for AccountValue: Error::ProtobufDecode surfaces as a synthesized notice.
    let message = proto_response(IncomingMessages::AccountValue, vec![0xFF, 0xFF, 0xFF]);

    let cb_fired = Arc::new(Mutex::new(false));
    let cb_fired_clone = cb_fired.clone();
    let cb = move |_msg: StartupMessage| *cb_fired_clone.lock().unwrap() = true;
    let sink = CapturingSink::default();

    dispatch_unsolicited_message(&message, &full_ctx(&cb, &sink));

    assert!(!*cb_fired.lock().unwrap(), "callback must not fire on decode failure");
    let notice = sink.last().expect("notice sink should receive decode-failure notice");
    assert_eq!(notice.code, HANDSHAKE_DECODE_FAILURE_CODE);
    assert!(notice.message.contains("AccountValue"));
}

// =============================================================================
// New typed-variant dispatch tests (PR 2 of retire-response-message-public-surface)
// =============================================================================

/// Build a proto-frame ResponseMessage for the given testdata builder.
fn proto_frame<B: crate::testdata::builders::ResponseProtoEncoder>(kind: IncomingMessages, builder: &B) -> ResponseMessage {
    proto_response(kind, builder.encode_proto())
}

#[test]
fn dispatch_unsolicited_execution_typed() {
    use crate::testdata::builders::orders::ExecutionDataResponse;

    let builder = ExecutionDataResponse::default().symbol("TSLA").shares(100.0).price(196.52);
    let message = proto_frame(IncomingMessages::ExecutionData, &builder);

    let captured: Arc<Mutex<Option<crate::orders::ExecutionData>>> = Arc::new(Mutex::new(None));
    let captured_clone = captured.clone();
    let cb = move |msg: StartupMessage| match msg {
        StartupMessage::Execution(e) => *captured.lock().unwrap() = Some(e),
        other => panic!("expected Execution, got {other:?}"),
    };

    dispatch_unsolicited_message(&message, &startup_ctx(&cb));
    let got = captured_clone.lock().unwrap().take().expect("callback didn't fire");
    assert_eq!(got.contract.symbol.to_string(), "TSLA");
    assert_eq!(got.execution.shares, 100.0);
    assert_eq!(got.execution.price, 196.52);
}

#[test]
fn dispatch_unsolicited_commission_report_typed() {
    use crate::testdata::builders::orders::CommissionReportResponse;

    let builder = CommissionReportResponse::default().commission(2.5).currency("USD");
    let message = proto_frame(IncomingMessages::CommissionsReport, &builder);

    let captured: Arc<Mutex<Option<crate::orders::CommissionReport>>> = Arc::new(Mutex::new(None));
    let captured_clone = captured.clone();
    let cb = move |msg: StartupMessage| match msg {
        StartupMessage::CommissionReport(c) => *captured.lock().unwrap() = Some(c),
        other => panic!("expected CommissionReport, got {other:?}"),
    };

    dispatch_unsolicited_message(&message, &startup_ctx(&cb));
    let got = captured_clone.lock().unwrap().take().expect("callback didn't fire");
    assert_eq!(got.commission, 2.5);
    assert_eq!(got.currency, "USD");
}

#[test]
fn dispatch_unsolicited_completed_order_typed() {
    use crate::testdata::builders::orders::CompletedOrderResponse;

    let builder = CompletedOrderResponse::default();
    let message = proto_frame(IncomingMessages::CompletedOrder, &builder);

    let captured: Arc<Mutex<Option<crate::orders::OrderData>>> = Arc::new(Mutex::new(None));
    let captured_clone = captured.clone();
    let cb = move |msg: StartupMessage| match msg {
        StartupMessage::CompletedOrder(o) => *captured.lock().unwrap() = Some(o),
        other => panic!("expected CompletedOrder, got {other:?}"),
    };

    dispatch_unsolicited_message(&message, &startup_ctx(&cb));
    let got = captured_clone.lock().unwrap().take().expect("callback didn't fire");
    // Completed orders carry the sentinel order_id = -1 per decode_completed_order_proto.
    assert_eq!(got.order_id, -1);
}

#[test]
fn dispatch_unsolicited_execution_data_end_typed() {
    // ExecutionDataEnd: msg_type=55, version=1, request_id=42
    let message = ResponseMessage::from("55\01\042\0");

    let fired = Arc::new(Mutex::new(false));
    let fired_clone = fired.clone();
    let cb = move |msg: StartupMessage| {
        if matches!(msg, StartupMessage::ExecutionDataEnd) {
            *fired.lock().unwrap() = true;
        } else {
            panic!("expected ExecutionDataEnd, got {msg:?}");
        }
    };

    dispatch_unsolicited_message(&message, &startup_ctx(&cb));
    assert!(*fired_clone.lock().unwrap(), "ExecutionDataEnd not delivered as typed variant");
}

#[test]
fn dispatch_unsolicited_completed_orders_end_typed() {
    // CompletedOrdersEnd: msg_type=102, no payload
    let message = ResponseMessage::from("102\0");

    let fired = Arc::new(Mutex::new(false));
    let fired_clone = fired.clone();
    let cb = move |msg: StartupMessage| {
        if matches!(msg, StartupMessage::CompletedOrdersEnd) {
            *fired.lock().unwrap() = true;
        } else {
            panic!("expected CompletedOrdersEnd, got {msg:?}");
        }
    };

    dispatch_unsolicited_message(&message, &startup_ctx(&cb));
    assert!(*fired_clone.lock().unwrap(), "CompletedOrdersEnd not delivered as typed variant");
}

#[test]
fn dispatch_unsolicited_execution_decode_failure_emits_notice() {
    // Text-framed ExecutionData — decoder calls require_proto() and rejects it.
    let message = ResponseMessage::from("11\042\013\0AAPL\0");

    let cb_fired = Arc::new(Mutex::new(false));
    let cb_fired_clone = cb_fired.clone();
    let cb = move |_msg: StartupMessage| *cb_fired_clone.lock().unwrap() = true;
    let sink = CapturingSink::default();

    dispatch_unsolicited_message(&message, &full_ctx(&cb, &sink));

    assert!(!*cb_fired.lock().unwrap(), "callback must not fire on decode failure");
    let notice = sink.last().expect("notice sink should receive decode-failure notice");
    assert_eq!(notice.code, HANDSHAKE_DECODE_FAILURE_CODE);
    assert!(notice.message.contains("ExecutionData"));
}

#[test]
fn dispatch_unsolicited_commission_report_decode_failure_emits_notice() {
    let message = ResponseMessage::from("59\01\0EXEC0001.01.01\0");

    let cb_fired = Arc::new(Mutex::new(false));
    let cb_fired_clone = cb_fired.clone();
    let cb = move |_msg: StartupMessage| *cb_fired_clone.lock().unwrap() = true;
    let sink = CapturingSink::default();

    dispatch_unsolicited_message(&message, &full_ctx(&cb, &sink));

    assert!(!*cb_fired.lock().unwrap(), "callback must not fire on decode failure");
    let notice = sink.last().expect("notice sink should receive decode-failure notice");
    assert_eq!(notice.code, HANDSHAKE_DECODE_FAILURE_CODE);
    assert!(notice.message.contains("CommissionsReport"));
}

#[test]
fn dispatch_unsolicited_completed_order_decode_failure_emits_notice() {
    let message = ResponseMessage::from("101\0\0");

    let cb_fired = Arc::new(Mutex::new(false));
    let cb_fired_clone = cb_fired.clone();
    let cb = move |_msg: StartupMessage| *cb_fired_clone.lock().unwrap() = true;
    let sink = CapturingSink::default();

    dispatch_unsolicited_message(&message, &full_ctx(&cb, &sink));

    assert!(!*cb_fired.lock().unwrap(), "callback must not fire on decode failure");
    let notice = sink.last().expect("notice sink should receive decode-failure notice");
    assert_eq!(notice.code, HANDSHAKE_DECODE_FAILURE_CODE);
    assert!(notice.message.contains("CompletedOrder"));
}

#[test]
fn dispatch_unsolicited_execution_no_callback_is_noop() {
    let message = ResponseMessage::from("11\042\013\0AAPL\0");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    assert_eq!(sink.count(), 0, "ExecutionData must not deliver to notice sink");
}

#[test]
fn dispatch_unsolicited_commission_report_no_callback_is_noop() {
    let message = ResponseMessage::from("59\01\0EXEC0001.01.01\0");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    assert_eq!(sink.count(), 0);
}

#[test]
fn dispatch_unsolicited_completed_order_no_callback_is_noop() {
    let message = ResponseMessage::from("101\0\0");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    assert_eq!(sink.count(), 0);
}

#[test]
fn dispatch_unsolicited_execution_data_end_no_callback_is_noop() {
    let message = ResponseMessage::from("55\01\042\0");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    assert_eq!(sink.count(), 0);
}

#[test]
fn dispatch_unsolicited_completed_orders_end_no_callback_is_noop() {
    let message = ResponseMessage::from("102\0");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    assert_eq!(sink.count(), 0);
}

/// An id that maps to no kind raises UNKNOWN_MESSAGE_TYPE_CODE during the
/// handshake too, not HANDSHAKE_UNKNOWN_FRAME_CODE: one condition, one code,
/// whether or not a reconnect is in flight.
#[test]
fn dispatch_unsolicited_unrecognized_id_raises_unknown_message_type() {
    let message = parse_raw_message(&helpers::unknown_message_frame()).expect("well-formed frame");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));

    let notice = sink.last().expect("an unrecognized id must raise a notice");
    assert_eq!(notice.code, UNKNOWN_MESSAGE_TYPE_CODE);
    assert!(!notice.is_handshake_synthetic());
    assert!(
        notice.message.contains(&helpers::UNKNOWN_MESSAGE_ID.to_string()),
        "notice must name the offending id, got {:?}",
        notice.message
    );
}

#[test]
fn dispatch_unsolicited_unknown_no_callback_still_notices() {
    // NewsBulletins (14). Unknown handshake kind always fires the notice
    // sink regardless of callback presence — synthesized notices target
    // `Client::notice_stream()`, decoupled from the typed startup callback.
    let message = ResponseMessage::from("14\0\0");

    let sink = CapturingSink::default();
    dispatch_unsolicited_message(&message, &notice_sink_ctx(&sink));
    let notice = sink.last().expect("notice sink should receive unknown-frame notice");
    assert_eq!(notice.code, HANDSHAKE_UNKNOWN_FRAME_CODE);
    assert!(notice.is_handshake_synthetic());
}

#[test]
fn account_info_merge() {
    let mut info = AccountInfo::default();

    let complete = info.merge(AccountInfo {
        next_order_id: Some(90),
        managed_accounts: None,
    });
    assert!(!complete, "only next_order_id is set");

    let complete = info.merge(AccountInfo::default());
    assert!(!complete);
    assert_eq!(info.next_order_id, Some(90), "absent field must not clear a set one");

    let complete = info.merge(AccountInfo {
        next_order_id: None,
        managed_accounts: Some("DU1234567".to_owned()),
    });
    assert!(complete, "both fields set");
    assert_eq!(info.next_order_id, Some(90));
    assert_eq!(info.managed_accounts.as_deref(), Some("DU1234567"));
}

#[test]
fn connection_metadata_apply_account_info() {
    let mut metadata = ConnectionMetadata {
        next_order_id: 1,
        managed_accounts: "OLD".to_owned(),
        ..Default::default()
    };

    metadata.apply_account_info(AccountInfo::default());
    assert_eq!(metadata.next_order_id, 1);
    assert_eq!(metadata.managed_accounts, "OLD");

    metadata.apply_account_info(AccountInfo {
        next_order_id: Some(90),
        managed_accounts: Some("DU1234567".to_owned()),
    });
    assert_eq!(metadata.next_order_id, 90);
    assert_eq!(metadata.managed_accounts, "DU1234567");
}

#[test]
fn parse_handshake_ack_works() {
    use std::io::ErrorKind;

    enum Expect {
        Ok(i32),
        Rejected,
        Io(ErrorKind),
        Parse,
    }

    let valid = format!("{}\020230405 22:20:39 PST\0", server_versions::PROTOBUF_REST_MESSAGES_3);
    let io = |kind| -> Result<Vec<u8>, Error> { Err(Error::Io(std::io::Error::from(kind))) };

    let cases = [
        (
            "valid reply",
            Ok(valid.into_bytes()),
            Expect::Ok(server_versions::PROTOBUF_REST_MESSAGES_3),
        ),
        ("unexpected eof", io(ErrorKind::UnexpectedEof), Expect::Rejected),
        ("other io error", io(ErrorKind::ConnectionReset), Expect::Io(ErrorKind::ConnectionReset)),
        ("empty reply", Ok(Vec::new()), Expect::Parse),
        ("non-numeric version", Ok(b"abc\020230405 22:20:39 PST\0".to_vec()), Expect::Parse),
        ("missing time", Ok(b"213\0".to_vec()), Expect::Parse),
    ];

    let handler = ConnectionHandler::default();
    for (name, input, expect) in cases {
        let result = parse_handshake_ack(&handler, input);
        match (expect, result) {
            (Expect::Ok(version), Ok(ack)) => {
                assert_eq!(ack.server_version, version, "{name}");
                assert!(ack.connection_time.is_some(), "{name}");
                assert!(ack.time_zone.is_some(), "{name}");
            }
            (Expect::Rejected, Err(Error::ConnectionRejected(msg))) => assert!(msg.contains("server may be rejecting"), "{name}: {msg}"),
            (Expect::Io(kind), Err(Error::Io(err))) => assert_eq!(err.kind(), kind, "{name}"),
            (Expect::Parse, Err(Error::Parse(..))) => {}
            (_, other) => panic!("{name}: unexpected result {other:?}"),
        }
    }
}