layer_client/

lib.rs

//! # layer-client
//!
//! Production-grade async Telegram client built on MTProto.
//!
//! ## Features
//! - User login (phone + code + 2FA SRP) and bot token login
//! - Peer access-hash caching — API calls always carry correct access hashes
//! - `FLOOD_WAIT` auto-retry with configurable policy
//! - Typed async update stream: `NewMessage`, `MessageEdited`, `MessageDeleted`,
//!   `CallbackQuery`, `InlineQuery`, `InlineSend`, `Raw`
//! - Send / edit / delete / forward / pin messages
//! - Search messages (per-chat and global)
//! - Mark as read, delete dialogs, clear mentions
//! - Join chat / accept invite links
//! - Chat action (typing, uploading, …)
//! - `get_me()` — fetch own User info
//! - Paginated dialog and message iterators
//! - DC migration, session persistence, reconnect

#![deny(unsafe_code)]

mod errors;
mod retry;
mod session;
mod transport;
mod two_factor_auth;
pub mod update;
pub mod parsers;
pub mod media;
pub mod participants;
pub mod pts;

// ── New feature modules ───────────────────────────────────────────────────────
pub mod dc_pool;
pub mod transport_obfuscated;
pub mod transport_intermediate;
pub mod socks5;
pub mod session_backend;
pub mod inline_iter;
pub mod typing_guard;
pub mod keyboard;

#[macro_use]
pub mod macros;

pub use errors::{InvocationError, LoginToken, PasswordToken, RpcError, SignInError};
pub use retry::{AutoSleep, NoRetries, RetryContext, RetryPolicy};
pub use update::Update;
pub use media::{UploadedFile, DownloadIter};
pub use participants::Participant;
pub use typing_guard::TypingGuard;
pub use socks5::Socks5Config;
pub use session_backend::{SessionBackend, BinaryFileBackend, InMemoryBackend};
pub use keyboard::{Button, InlineKeyboard, ReplyKeyboard};

/// Re-export of `layer_tl_types` — generated TL constructors, functions, and enums.
/// Users can write `use layer_client::tl` instead of adding a separate `layer-tl-types` dep.
pub use layer_tl_types as tl;

use std::collections::HashMap;
use std::collections::VecDeque;
use std::num::NonZeroU32;
use std::ops::ControlFlow;
use std::sync::Arc;
use std::time::Duration;

use layer_mtproto::{EncryptedSession, Session, authentication as auth};
use layer_tl_types::{Cursor, Deserializable, RemoteCall};
use session::{DcEntry, PersistedSession};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::{mpsc, oneshot, Mutex};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;
use socket2::TcpKeepalive;

// ─── MTProto envelope constructor IDs ────────────────────────────────────────

const ID_RPC_RESULT:       u32 = 0xf35c6d01;
const ID_RPC_ERROR:        u32 = 0x2144ca19;
const ID_MSG_CONTAINER:    u32 = 0x73f1f8dc;
const ID_GZIP_PACKED:      u32 = 0x3072cfa1;
const ID_PONG:             u32 = 0x347773c5;
const ID_MSGS_ACK:         u32 = 0x62d6b459;
const ID_BAD_SERVER_SALT:  u32 = 0xedab447b;
const ID_NEW_SESSION:      u32 = 0x9ec20908;
const ID_BAD_MSG_NOTIFY:   u32 = 0xa7eff811;
const ID_UPDATES:          u32 = 0x74ae4240;
const ID_UPDATE_SHORT:     u32 = 0x78d4dec1;
const ID_UPDATES_COMBINED: u32 = 0x725b04c3;
const ID_UPDATE_SHORT_MSG:      u32 = 0x313bc7f8;
const ID_UPDATE_SHORT_CHAT_MSG: u32 = 0x4d6deea5;
const ID_UPDATES_TOO_LONG:      u32 = 0xe317af7e;

// ─── Keepalive / reconnect tuning ─────────────────────────────────────────────

/// How often to send a keepalive ping.
/// Telegram Desktop uses ~15 s on mobile connections.  60 s is too slow to
/// detect a dead connection before the user notices.
const PING_DELAY_SECS: u64 = 15;

/// Tell Telegram to close the connection if it hears nothing for this many
/// seconds.  Must be > PING_DELAY_SECS so a single missed ping doesn't drop us.
const NO_PING_DISCONNECT: i32 = 20;

/// Initial backoff before the first reconnect attempt.
const RECONNECT_BASE_MS: u64 = 500;

/// Maximum backoff between reconnect attempts.
/// 5 s cap instead of 30 s — on mobile, network outages are brief and a 30 s
/// sleep means the bot stays dead for up to 30 s after the network returns.
/// Official Telegram mobile clients use a short cap for the same reason.
const RECONNECT_MAX_SECS: u64 = 5;

/// TCP socket-level keepalive: start probes after this many seconds of idle.
const TCP_KEEPALIVE_IDLE_SECS: u64 = 10;
/// Interval between TCP keepalive probes.
const TCP_KEEPALIVE_INTERVAL_SECS: u64 = 5;
/// Number of failed probes before the OS declares the connection dead.
const TCP_KEEPALIVE_PROBES: u32 = 3;

// ─── PeerCache ────────────────────────────────────────────────────────────────

/// Caches access hashes for users and channels so every API call carries the
/// correct hash without re-resolving peers.
#[derive(Default)]
pub(crate) struct PeerCache {
    /// user_id → access_hash
    pub(crate) users:    HashMap<i64, i64>,
    /// channel_id → access_hash
    pub(crate) channels: HashMap<i64, i64>,
}

impl PeerCache {
    fn cache_user(&mut self, user: &tl::enums::User) {
        if let tl::enums::User::User(u) = user {
            if let Some(hash) = u.access_hash {
                self.users.insert(u.id, hash);
            }
        }
    }

    fn cache_chat(&mut self, chat: &tl::enums::Chat) {
        match chat {
            tl::enums::Chat::Channel(c) => {
                if let Some(hash) = c.access_hash {
                    self.channels.insert(c.id, hash);
                }
            }
            tl::enums::Chat::ChannelForbidden(c) => {
                self.channels.insert(c.id, c.access_hash);
            }
            _ => {}
        }
    }

    fn cache_users(&mut self, users: &[tl::enums::User]) {
        for u in users { self.cache_user(u); }
    }

    fn cache_chats(&mut self, chats: &[tl::enums::Chat]) {
        for c in chats { self.cache_chat(c); }
    }

    fn user_input_peer(&self, user_id: i64) -> tl::enums::InputPeer {
        if user_id == 0 {
            return tl::enums::InputPeer::PeerSelf;
        }
        let hash = self.users.get(&user_id).copied().unwrap_or(0);
        tl::enums::InputPeer::User(tl::types::InputPeerUser { user_id, access_hash: hash })
    }

    fn channel_input_peer(&self, channel_id: i64) -> tl::enums::InputPeer {
        let hash = self.channels.get(&channel_id).copied().unwrap_or(0);
        tl::enums::InputPeer::Channel(tl::types::InputPeerChannel { channel_id, access_hash: hash })
    }

    fn peer_to_input(&self, peer: &tl::enums::Peer) -> tl::enums::InputPeer {
        match peer {
            tl::enums::Peer::User(u) => self.user_input_peer(u.user_id),
            tl::enums::Peer::Chat(c) => tl::enums::InputPeer::Chat(
                tl::types::InputPeerChat { chat_id: c.chat_id }
            ),
            tl::enums::Peer::Channel(c) => self.channel_input_peer(c.channel_id),
        }
    }
}

// ─── InputMessage builder ─────────────────────────────────────────────────────

/// Builder for composing outgoing messages.
///
/// ```rust,no_run
/// use layer_client::InputMessage;
///
/// let msg = InputMessage::text("Hello, *world*!")
///     .silent(true)
///     .reply_to(Some(42));
/// ```
#[derive(Clone, Default)]
pub struct InputMessage {
    pub text:         String,
    pub reply_to:     Option<i32>,
    pub silent:       bool,
    pub background:   bool,
    pub clear_draft:  bool,
    pub no_webpage:   bool,
    pub entities:     Option<Vec<tl::enums::MessageEntity>>,
    pub reply_markup: Option<tl::enums::ReplyMarkup>,
    pub schedule_date: Option<i32>,
}

impl InputMessage {
    /// Create a message with the given text.
    pub fn text(text: impl Into<String>) -> Self {
        Self { text: text.into(), ..Default::default() }
    }

    /// Set the message text.
    pub fn set_text(mut self, text: impl Into<String>) -> Self {
        self.text = text.into(); self
    }

    /// Reply to a specific message ID.
    pub fn reply_to(mut self, id: Option<i32>) -> Self {
        self.reply_to = id; self
    }

    /// Send silently (no notification sound).
    pub fn silent(mut self, v: bool) -> Self {
        self.silent = v; self
    }

    /// Send in background.
    pub fn background(mut self, v: bool) -> Self {
        self.background = v; self
    }

    /// Clear the draft after sending.
    pub fn clear_draft(mut self, v: bool) -> Self {
        self.clear_draft = v; self
    }

    /// Disable link preview.
    pub fn no_webpage(mut self, v: bool) -> Self {
        self.no_webpage = v; self
    }

    /// Attach formatting entities (bold, italic, code, links, etc).
    pub fn entities(mut self, e: Vec<tl::enums::MessageEntity>) -> Self {
        self.entities = Some(e); self
    }

    /// Attach a reply markup (inline or reply keyboard).
    pub fn reply_markup(mut self, rm: tl::enums::ReplyMarkup) -> Self {
        self.reply_markup = Some(rm); self
    }

    /// Shorthand for attaching an [`crate::keyboard::InlineKeyboard`].
    ///
    /// ```rust,no_run
    /// use layer_client::{InputMessage, keyboard::{InlineKeyboard, Button}};
    ///
    /// let msg = InputMessage::text("Pick one:")
    ///     .keyboard(InlineKeyboard::new()
    ///         .row([Button::callback("A", b"a"), Button::callback("B", b"b")]));
    /// ```
    pub fn keyboard(mut self, kb: impl Into<tl::enums::ReplyMarkup>) -> Self {
        self.reply_markup = Some(kb.into()); self
    }

    /// Schedule the message for a future Unix timestamp.
    pub fn schedule_date(mut self, ts: Option<i32>) -> Self {
        self.schedule_date = ts; self
    }

    fn reply_header(&self) -> Option<tl::enums::InputReplyTo> {
        self.reply_to.map(|id| {
            tl::enums::InputReplyTo::Message(
                tl::types::InputReplyToMessage {
                    reply_to_msg_id: id,
                    top_msg_id: None,
                    reply_to_peer_id: None,
                    quote_text: None,
                    quote_entities: None,
                    quote_offset: None,
                    monoforum_peer_id: None,
                    todo_item_id: None,
                }
            )
        })
    }
}

impl From<&str> for InputMessage {
    fn from(s: &str) -> Self { Self::text(s) }
}

impl From<String> for InputMessage {
    fn from(s: String) -> Self { Self::text(s) }
}

// ─── TransportKind ────────────────────────────────────────────────────────────

/// Which MTProto transport framing to use for all connections.
///
/// | Variant | Init bytes | Notes |
/// |---------|-----------|-------|
/// | `Abridged` | `0xef` | Default, smallest overhead |
/// | `Intermediate` | `0xeeeeeeee` | Better proxy compat |
/// | `Full` | none | Adds seqno + CRC32 |
/// | `Obfuscated` | random 64B | Bypasses DPI / MTProxy |
#[derive(Clone, Debug, Default)]
pub enum TransportKind {
    /// MTProto [Abridged] transport — length prefix is 1 or 4 bytes.
    ///
    /// [Abridged]: https://core.telegram.org/mtproto/mtproto-transports#abridged
    #[default]
    Abridged,
    /// MTProto [Intermediate] transport — 4-byte LE length prefix.
    ///
    /// [Intermediate]: https://core.telegram.org/mtproto/mtproto-transports#intermediate
    Intermediate,
    /// MTProto [Full] transport — 4-byte length + seqno + CRC32.
    ///
    /// [Full]: https://core.telegram.org/mtproto/mtproto-transports#full
    Full,
    /// [Obfuscated2] transport — XOR stream cipher over Abridged framing.
    /// Required for MTProxy and networks with deep-packet inspection.
    ///
    /// `secret` is the 16-byte proxy secret, or `None` for keyless obfuscation.
    ///
    /// [Obfuscated2]: https://core.telegram.org/mtproto/mtproto-transports#obfuscated-2
    Obfuscated { secret: Option<[u8; 16]> },
}

// ─── Config ───────────────────────────────────────────────────────────────────

/// A token that can be used to gracefully shut down a [`Client`].
///
/// Obtained from [`Client::connect`] — call [`ShutdownToken::cancel`] to begin
/// graceful shutdown. All pending requests will finish and the reader task will
/// exit cleanly.
///
/// # Example
/// ```rust,no_run
/// # async fn f() -> Result<(), Box<dyn std::error::Error>> {
/// use layer_client::{Client, Config, ShutdownToken};
///
/// let (client, shutdown) = Client::connect(Config::default()).await?;
///
/// // In a signal handler or background task:
/// // shutdown.cancel();
/// # Ok(()) }
/// ```
pub type ShutdownToken = CancellationToken;

/// Configuration for [`Client::connect`].
#[derive(Clone)]
pub struct Config {
    pub api_id:         i32,
    pub api_hash:       String,
    pub dc_addr:        Option<String>,
    pub retry_policy:   Arc<dyn RetryPolicy>,
    /// Optional SOCKS5 proxy — every Telegram connection is tunnelled through it.
    pub socks5:         Option<crate::socks5::Socks5Config>,
    /// Allow IPv6 DC addresses when populating the DC table (default: false).
    pub allow_ipv6:     bool,
    /// Which MTProto transport framing to use (default: Abridged).
    pub transport:      TransportKind,
    /// Session persistence backend (default: binary file `"layer.session"`).
    pub session_backend: Arc<dyn crate::session_backend::SessionBackend>,
    /// If `true`, replay missed updates via `updates.getDifference` immediately
    /// after connecting. Mirrors grammers' `UpdatesConfiguration { catch_up: true }`.
    /// Default: `false`.
    pub catch_up:       bool,
}

impl Default for Config {
    fn default() -> Self {
        Self {
            api_id:          0,
            api_hash:        String::new(),
            dc_addr:         None,
            retry_policy:    Arc::new(AutoSleep::default()),
            socks5:          None,
            allow_ipv6:      false,
            transport:       TransportKind::Abridged,
            session_backend: Arc::new(crate::session_backend::BinaryFileBackend::new("layer.session")),
            catch_up:        false,
        }
    }
}

// ─── UpdateStream ─────────────────────────────────────────────────────────────

/// Asynchronous stream of [`Update`]s.
pub struct UpdateStream {
    rx: mpsc::UnboundedReceiver<update::Update>,
}

impl UpdateStream {
    /// Wait for the next update. Returns `None` when the client has disconnected.
    pub async fn next(&mut self) -> Option<update::Update> {
        self.rx.recv().await
    }
}

// ─── Dialog ───────────────────────────────────────────────────────────────────

/// A Telegram dialog (chat, user, channel).
#[derive(Debug, Clone)]
pub struct Dialog {
    pub raw:     tl::enums::Dialog,
    pub message: Option<tl::enums::Message>,
    pub entity:  Option<tl::enums::User>,
    pub chat:    Option<tl::enums::Chat>,
}

impl Dialog {
    /// The dialog's display title.
    pub fn title(&self) -> String {
        if let Some(tl::enums::User::User(u)) = &self.entity {
            let first = u.first_name.as_deref().unwrap_or("");
            let last  = u.last_name.as_deref().unwrap_or("");
            let name  = format!("{first} {last}").trim().to_string();
            if !name.is_empty() { return name; }
        }
        if let Some(chat) = &self.chat {
            return match chat {
                tl::enums::Chat::Chat(c)         => c.title.clone(),
                tl::enums::Chat::Forbidden(c) => c.title.clone(),
                tl::enums::Chat::Channel(c)      => c.title.clone(),
                tl::enums::Chat::ChannelForbidden(c) => c.title.clone(),
                tl::enums::Chat::Empty(_)        => "(empty)".into(),
            };
        }
        "(Unknown)".to_string()
    }

    /// Peer of this dialog.
    pub fn peer(&self) -> Option<&tl::enums::Peer> {
        match &self.raw {
            tl::enums::Dialog::Dialog(d) => Some(&d.peer),
            tl::enums::Dialog::Folder(_) => None,
        }
    }

    /// Unread message count.
    pub fn unread_count(&self) -> i32 {
        match &self.raw {
            tl::enums::Dialog::Dialog(d) => d.unread_count,
            _ => 0,
        }
    }

    /// ID of the top message.
    pub fn top_message(&self) -> i32 {
        match &self.raw {
            tl::enums::Dialog::Dialog(d) => d.top_message,
            _ => 0,
        }
    }
}

// ─── ClientInner ─────────────────────────────────────────────────────────────

struct ClientInner {
    /// Write half of the connection — holds the EncryptedSession (for packing)
    /// and the send half of the TCP stream. The read half is owned by the
    /// reader task started in connect().
    writer:          Mutex<ConnectionWriter>,
    /// Pending RPC replies, keyed by MTProto msg_id.
    /// RPC callers insert a oneshot::Sender here before sending; the reader
    /// task routes incoming rpc_result frames to the matching sender.
    pending:         Arc<Mutex<HashMap<i64, oneshot::Sender<Result<Vec<u8>, InvocationError>>>>>,
    /// Channel used to hand a new (OwnedReadHalf, FrameKind, auth_key, session_id)
    /// to the reader task after a reconnect.
    reconnect_tx:    mpsc::UnboundedSender<(OwnedReadHalf, FrameKind, [u8; 256], i64)>,
    /// Send `()` here to wake the reader's reconnect backoff loop immediately.
    /// Used by [`Client::signal_network_restored`].
    network_hint_tx: mpsc::UnboundedSender<()>,
    /// Cancelled to signal graceful shutdown to the reader task.
    #[allow(dead_code)]
    shutdown_token:  CancellationToken,
    /// Whether to replay missed updates via getDifference on connect.
    #[allow(dead_code)]
    catch_up:        bool,
    home_dc_id:      Mutex<i32>,
    dc_options:      Mutex<HashMap<i32, DcEntry>>,
    pub(crate) peer_cache:    Mutex<PeerCache>,
    pub(crate) pts_state:     Mutex<pts::PtsState>,
    api_id:          i32,
    api_hash:        String,
    retry_policy:    Arc<dyn RetryPolicy>,
    socks5:          Option<crate::socks5::Socks5Config>,
    allow_ipv6:      bool,
    transport:       TransportKind,
    session_backend: Arc<dyn crate::session_backend::SessionBackend>,
    dc_pool:         Mutex<dc_pool::DcPool>,
    update_tx:       mpsc::UnboundedSender<update::Update>,
}

/// The main Telegram client. Cheap to clone — internally Arc-wrapped.
#[derive(Clone)]
pub struct Client {
    pub(crate) inner: Arc<ClientInner>,
    _update_rx: Arc<Mutex<mpsc::UnboundedReceiver<update::Update>>>,
}

impl Client {
    // ── Connect ────────────────────────────────────────────────────────────

    pub async fn connect(config: Config) -> Result<(Self, ShutdownToken), InvocationError> {
        let (update_tx, update_rx) = mpsc::unbounded_channel();

        // ── Load or fresh-connect ───────────────────────────────────────
        let socks5    = config.socks5.clone();
        let transport = config.transport.clone();

        let (conn, home_dc_id, dc_opts) =
            match config.session_backend.load()
                .map_err(InvocationError::Io)?
            {
                Some(s) => {
                    if let Some(dc) = s.dcs.iter().find(|d| d.dc_id == s.home_dc_id) {
                        if let Some(key) = dc.auth_key {
                            log::info!("[layer] Loading session (DC{}) …", s.home_dc_id);
                            match Connection::connect_with_key(
                                &dc.addr, key, dc.first_salt, dc.time_offset,
                                socks5.as_ref(), &transport,
                            ).await {
                                Ok(c) => {
                                    let mut opts = session::default_dc_addresses()
                                        .into_iter()
                                        .map(|(id, addr)| (id, DcEntry { dc_id: id, addr, auth_key: None, first_salt: 0, time_offset: 0 }))
                                        .collect::<HashMap<_, _>>();
                                    for d in &s.dcs { opts.insert(d.dc_id, d.clone()); }
                                    (c, s.home_dc_id, opts)
                                }
                                Err(e) => {
                                    log::warn!("[layer] Session connect failed ({e}), fresh connect …");
                                    Self::fresh_connect(socks5.as_ref(), &transport).await?
                                }
                            }
                        } else {
                            Self::fresh_connect(socks5.as_ref(), &transport).await?
                        }
                    } else {
                        Self::fresh_connect(socks5.as_ref(), &transport).await?
                    }
                }
                None => Self::fresh_connect(socks5.as_ref(), &transport).await?,
            };

        // ── Build DC pool ───────────────────────────────────────────────
        let pool = dc_pool::DcPool::new(home_dc_id, &dc_opts.values().cloned().collect::<Vec<_>>());

        // Split the TCP stream immediately.
        // The writer (write half + EncryptedSession) stays in ClientInner.
        // The read half goes to the reader task which we spawn right now so
        // that RPC calls during init_connection work correctly.
        let (writer, read_half, frame_kind) = conn.into_writer();
        let auth_key   = writer.enc.auth_key_bytes();
        let session_id = writer.enc.session_id();

        let pending: Arc<Mutex<HashMap<i64, oneshot::Sender<Result<Vec<u8>, InvocationError>>>>> =
            Arc::new(Mutex::new(HashMap::new()));

        // Channel the reconnect logic uses to hand a new read half to the reader task.
        let (reconnect_tx, reconnect_rx) =
            mpsc::unbounded_channel::<(OwnedReadHalf, FrameKind, [u8; 256], i64)>();

        // Channel for external "network restored" hints — lets Android/iOS callbacks
        // skip the reconnect backoff and attempt immediately.
        let (network_hint_tx, network_hint_rx) = mpsc::unbounded_channel::<()>();


        // Graceful shutdown token — cancel this to stop the reader task cleanly.
        let shutdown_token = CancellationToken::new();
        let catch_up = config.catch_up;

        let inner = Arc::new(ClientInner {
            writer:          Mutex::new(writer),
            pending:         pending.clone(),
            reconnect_tx,
            network_hint_tx,
            shutdown_token:  shutdown_token.clone(),
            catch_up,
            home_dc_id:      Mutex::new(home_dc_id),
            dc_options:      Mutex::new(dc_opts),
            peer_cache:      Mutex::new(PeerCache::default()),
            pts_state:       Mutex::new(pts::PtsState::default()),
            api_id:          config.api_id,
            api_hash:        config.api_hash,
            retry_policy:    config.retry_policy,
            socks5:          config.socks5,
            allow_ipv6:      config.allow_ipv6,
            transport:       config.transport,
            session_backend: config.session_backend,
            dc_pool:         Mutex::new(pool),
            update_tx:       update_tx,
        });

        let client = Self {
            inner,
            _update_rx: Arc::new(Mutex::new(update_rx)),
        };

        // Spawn the reader task immediately so that RPC calls during
        // init_connection can receive their responses.
        {
            let client_r = client.clone();
            let shutdown_r = shutdown_token.clone();
            tokio::spawn(async move {
                client_r.run_reader_task(
                    read_half, frame_kind, auth_key, session_id,
                    reconnect_rx, network_hint_rx, shutdown_r,
                ).await;
            });
        }

        // If init_connection fails (e.g. stale auth key rejected by Telegram),
        // do a fresh DH handshake and retry once.
        if let Err(e) = client.init_connection().await {
            log::warn!("[layer] init_connection failed ({e}), retrying with fresh connect …");

            let socks5_r    = client.inner.socks5.clone();
            let transport_r = client.inner.transport.clone();
            let (new_conn, new_dc_id, new_opts) =
                Self::fresh_connect(socks5_r.as_ref(), &transport_r).await?;

            {
                let mut dc_guard = client.inner.home_dc_id.lock().await;
                *dc_guard = new_dc_id;
            }
            {
                let mut opts_guard = client.inner.dc_options.lock().await;
                *opts_guard = new_opts;
            }

            // Replace writer and hand new read half to reader task
            let (new_writer, new_read, new_fk) = new_conn.into_writer();
            let new_ak = new_writer.enc.auth_key_bytes();
            let new_sid = new_writer.enc.session_id();
            *client.inner.writer.lock().await = new_writer;
            let _ = client.inner.reconnect_tx.send((new_read, new_fk, new_ak, new_sid));

            client.init_connection().await?;
        }

        let _ = client.sync_pts_state().await;

        // If catch_up is enabled, replay any missed updates immediately.
        if catch_up {
            let c = client.clone();
            let utx = client.inner.update_tx.clone();
            tokio::spawn(async move {
                if let Ok(missed) = c.get_difference().await {
                    for u in missed { let _ = utx.send(u); }
                }
            });
        }

        Ok((client, shutdown_token))
    }

    async fn fresh_connect(
        socks5:    Option<&crate::socks5::Socks5Config>,
        transport: &TransportKind,
    ) -> Result<(Connection, i32, HashMap<i32, DcEntry>), InvocationError> {
        log::info!("[layer] Fresh connect to DC2 …");
        let conn = Connection::connect_raw("149.154.167.51:443", socks5, transport).await?;
        let opts = session::default_dc_addresses()
            .into_iter()
            .map(|(id, addr)| (id, DcEntry { dc_id: id, addr, auth_key: None, first_salt: 0, time_offset: 0 }))
            .collect();
        Ok((conn, 2, opts))
    }

    // ── Session ────────────────────────────────────────────────────────────

    pub async fn save_session(&self) -> Result<(), InvocationError> {
        let writer_guard = self.inner.writer.lock().await;
        let home_dc_id   = *self.inner.home_dc_id.lock().await;
        let dc_options   = self.inner.dc_options.lock().await;

        let mut dcs: Vec<DcEntry> = dc_options.values().map(|e| DcEntry {
            dc_id:       e.dc_id,
            addr:        e.addr.clone(),
            auth_key:    if e.dc_id == home_dc_id { Some(writer_guard.auth_key_bytes()) } else { e.auth_key },
            first_salt:  if e.dc_id == home_dc_id { writer_guard.first_salt() } else { e.first_salt },
            time_offset: if e.dc_id == home_dc_id { writer_guard.time_offset() } else { e.time_offset },
        }).collect();
        // Collect auth keys from worker DCs in the pool
        self.inner.dc_pool.lock().await.collect_keys(&mut dcs);

        self.inner.session_backend
            .save(&PersistedSession { home_dc_id, dcs })
            .map_err(InvocationError::Io)?;
        log::info!("[layer] Session saved ✓");
        Ok(())
    }

    // ── Auth ───────────────────────────────────────────────────────────────

    /// Returns `true` if the client is already authorized.
    pub async fn is_authorized(&self) -> Result<bool, InvocationError> {
        match self.invoke(&tl::functions::updates::GetState {}).await {
            Ok(_)  => Ok(true),
            Err(e) if e.is("AUTH_KEY_UNREGISTERED")
                   || matches!(&e, InvocationError::Rpc(r) if r.code == 401) => Ok(false),
            Err(e) => Err(e),
        }
    }

    /// Sign in as a bot.
    pub async fn bot_sign_in(&self, token: &str) -> Result<String, InvocationError> {
        let req = tl::functions::auth::ImportBotAuthorization {
            flags: 0, api_id: self.inner.api_id,
            api_hash: self.inner.api_hash.clone(),
            bot_auth_token: token.to_string(),
        };

        let result = match self.invoke(&req).await {
            Ok(x) => x,
            Err(InvocationError::Rpc(ref r)) if r.code == 303 => {
                let dc_id = r.value.unwrap_or(2) as i32;
                self.migrate_to(dc_id).await?;
                self.invoke(&req).await?
            }
            Err(e) => return Err(e),
        };

        let name = match result {
            tl::enums::auth::Authorization::Authorization(a) => {
                self.cache_user(&a.user).await;
                Self::extract_user_name(&a.user)
            }
            tl::enums::auth::Authorization::SignUpRequired(_) => {
                panic!("unexpected SignUpRequired during bot sign-in")
            }
        };
        log::info!("[layer] Bot signed in ✓  ({name})");
        Ok(name)
    }

    /// Request a login code for a user account.
    pub async fn request_login_code(&self, phone: &str) -> Result<LoginToken, InvocationError> {
        use tl::enums::auth::SentCode;

        let req = self.make_send_code_req(phone);
        let body = match self.rpc_call_raw(&req).await {
            Ok(b) => b,
            Err(InvocationError::Rpc(ref r)) if r.code == 303 => {
                let dc_id = r.value.unwrap_or(2) as i32;
                self.migrate_to(dc_id).await?;
                self.rpc_call_raw(&req).await?
            }
            Err(e) => return Err(e),
        };

        let mut cur = Cursor::from_slice(&body);
        let hash = match tl::enums::auth::SentCode::deserialize(&mut cur)? {
            SentCode::SentCode(s) => s.phone_code_hash,
            SentCode::Success(_)  => return Err(InvocationError::Deserialize("unexpected Success".into())),
            SentCode::PaymentRequired(_) => return Err(InvocationError::Deserialize("payment required to send code".into())),
        };
        log::info!("[layer] Login code sent");
        Ok(LoginToken { phone: phone.to_string(), phone_code_hash: hash })
    }

    /// Complete sign-in with the code sent to the phone.
    pub async fn sign_in(&self, token: &LoginToken, code: &str) -> Result<String, SignInError> {
        let req = tl::functions::auth::SignIn {
            phone_number:    token.phone.clone(),
            phone_code_hash: token.phone_code_hash.clone(),
            phone_code:      Some(code.trim().to_string()),
            email_verification: None,
        };

        let body = match self.rpc_call_raw(&req).await {
            Ok(b) => b,
            Err(InvocationError::Rpc(ref r)) if r.code == 303 => {
                let dc_id = r.value.unwrap_or(2) as i32;
                self.migrate_to(dc_id).await.map_err(SignInError::Other)?;
                self.rpc_call_raw(&req).await.map_err(SignInError::Other)?
            }
            Err(e) if e.is("SESSION_PASSWORD_NEEDED") => {
                let t = self.get_password_info().await.map_err(SignInError::Other)?;
                return Err(SignInError::PasswordRequired(t));
            }
            Err(e) if e.is("PHONE_CODE_*") => return Err(SignInError::InvalidCode),
            Err(e) => return Err(SignInError::Other(e)),
        };

        let mut cur = Cursor::from_slice(&body);
        match tl::enums::auth::Authorization::deserialize(&mut cur)
            .map_err(|e| SignInError::Other(e.into()))?
        {
            tl::enums::auth::Authorization::Authorization(a) => {
                self.cache_user(&a.user).await;
                let name = Self::extract_user_name(&a.user);
                log::info!("[layer] Signed in ✓  Welcome, {name}!");
                Ok(name)
            }
            tl::enums::auth::Authorization::SignUpRequired(_) => Err(SignInError::SignUpRequired),
        }
    }

    /// Complete 2FA login.
    pub async fn check_password(
        &self,
        token:    PasswordToken,
        password: impl AsRef<[u8]>,
    ) -> Result<String, InvocationError> {
        let pw   = token.password;
        let algo = pw.current_algo.ok_or_else(|| InvocationError::Deserialize("no current_algo".into()))?;
        let (salt1, salt2, p, g) = Self::extract_password_params(&algo)?;
        let g_b  = pw.srp_b.ok_or_else(|| InvocationError::Deserialize("no srp_b".into()))?;
        let a    = pw.secure_random;
        let srp_id = pw.srp_id.ok_or_else(|| InvocationError::Deserialize("no srp_id".into()))?;

        let (m1, g_a) = two_factor_auth::calculate_2fa(salt1, salt2, p, g, &g_b, &a, password.as_ref());
        let req = tl::functions::auth::CheckPassword {
            password: tl::enums::InputCheckPasswordSrp::InputCheckPasswordSrp(
                tl::types::InputCheckPasswordSrp {
                    srp_id, a: g_a.to_vec(), m1: m1.to_vec(),
                },
            ),
        };

        let body = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        match tl::enums::auth::Authorization::deserialize(&mut cur)? {
            tl::enums::auth::Authorization::Authorization(a) => {
                self.cache_user(&a.user).await;
                let name = Self::extract_user_name(&a.user);
                log::info!("[layer] 2FA ✓  Welcome, {name}!");
                Ok(name)
            }
            tl::enums::auth::Authorization::SignUpRequired(_) =>
                Err(InvocationError::Deserialize("unexpected SignUpRequired after 2FA".into())),
        }
    }

    /// Sign out and invalidate the current session.
    pub async fn sign_out(&self) -> Result<bool, InvocationError> {
        let req = tl::functions::auth::LogOut {};
        match self.rpc_call_raw(&req).await {
            Ok(_) => { log::info!("[layer] Signed out ✓"); Ok(true) }
            Err(e) if e.is("AUTH_KEY_UNREGISTERED") => Ok(false),
            Err(e) => Err(e),
        }
    }

    // ── Get self ───────────────────────────────────────────────────────────

    /// Fetch information about the logged-in user.
    pub async fn get_me(&self) -> Result<tl::types::User, InvocationError> {
        let req = tl::functions::users::GetUsers {
            id: vec![tl::enums::InputUser::UserSelf],
        };
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let users   = Vec::<tl::enums::User>::deserialize(&mut cur)?;
        self.cache_users_slice(&users).await;
        users.into_iter().find_map(|u| match u {
            tl::enums::User::User(u) => Some(u),
            _ => None,
        }).ok_or_else(|| InvocationError::Deserialize("getUsers returned no user".into()))
    }

    // ── Updates ────────────────────────────────────────────────────────────

    /// Return an [`UpdateStream`] that yields incoming [`Update`]s.
    ///
    /// The reader task (started inside `connect()`) sends all updates to
    /// `inner.update_tx`. This method proxies those updates into a fresh
    /// caller-owned channel — typically called once per bot/app loop.
    pub fn stream_updates(&self) -> UpdateStream {
        let (caller_tx, rx) = mpsc::unbounded_channel::<update::Update>();
        // The internal channel is: reader_task → update_tx → _update_rx
        // We forward _update_rx → caller_tx so the caller gets an owned stream.
        let internal_rx = self._update_rx.clone();
        tokio::spawn(async move {
            // Lock once and hold — this is the sole consumer of the internal channel.
            let mut guard = internal_rx.lock().await;
            while let Some(upd) = guard.recv().await {
                if caller_tx.send(upd).is_err() { break; }
            }
        });
        UpdateStream { rx }
    }

    // ── Network hint ───────────────────────────────────────────────────────

    /// Signal that network connectivity has been restored.
    ///
    /// Call this from platform network-change callbacks — Android's
    /// `ConnectivityManager`, iOS `NWPathMonitor`, or any other OS hook —
    /// to make the client attempt an immediate reconnect instead of waiting
    /// for the exponential backoff timer to expire.
    ///
    /// Safe to call at any time: if the connection is healthy the hint is
    /// silently ignored by the reader task; if it is in a backoff loop it
    /// wakes up and tries again right away.
    pub fn signal_network_restored(&self) {
        let _ = self.inner.network_hint_tx.send(());
    }

    // ── Reader task ────────────────────────────────────────────────────────
    // Decrypts frames without holding any lock, then routes:
    //   rpc_result  → pending map (oneshot to waiting RPC caller)
    //   update      → update_tx  (delivered to stream_updates consumers)
    //   bad_server_salt → updates writer salt
    //
    // On error: drains pending with Io errors (so AutoSleep retries callers),
    // then loops with exponential backoff until reconnect succeeds.
    // network_hint_rx lets external callers (Android/iOS) skip the backoff.
    //
    // DC migration / reconnect: the new read half arrives via new_conn_rx.
    // The select! between recv_frame_owned and new_conn_rx.recv() ensures we
    // switch to the new connection immediately, without waiting for the next
    // frame on the old (now stale) connection.

    // ── Reader task supervisor ────────────────────────────────────────────────
    //
    // run_reader_task is the outer supervisor. It wraps reader_loop in a
    // restart loop so that if reader_loop ever exits for any reason other than
    // a clean shutdown request, it is automatically reconnected and restarted.
    //
    // This mirrors what Telegram Desktop does: the network thread is considered
    // infrastructure and is never allowed to die permanently.
    //
    // Restart sequence on unexpected exit:
    //   1. Drain all pending RPCs with ConnectionReset so callers unblock.
    //   2. Exponential-backoff reconnect loop (500 ms → 30 s cap) until TCP
    //      succeeds, respecting the shutdown token at every sleep point.
    //   3. Spawn init_connection in a background task (same deadlock-safe
    //      pattern as do_reconnect_loop) and pass the oneshot receiver as the
    //      initial_init_rx to the restarted reader_loop.
    //   4. reader_loop picks up init_rx immediately on its first iteration and
    //      handles success/failure exactly like a mid-session reconnect.
    async fn run_reader_task(
        &self,
        read_half:   OwnedReadHalf,
        frame_kind:  FrameKind,
        auth_key:    [u8; 256],
        session_id:  i64,
        mut new_conn_rx:      mpsc::UnboundedReceiver<(OwnedReadHalf, FrameKind, [u8; 256], i64)>,
        mut network_hint_rx:  mpsc::UnboundedReceiver<()>,
        shutdown_token:       CancellationToken,
    ) {
        let mut rh  = read_half;
        let mut fk  = frame_kind;
        let mut ak  = auth_key;
        let mut sid = session_id;
        // On first start no init is needed (connect() already called it).
        // On restarts we pass the spawned init task so reader_loop handles it.
        let mut restart_init_rx: Option<oneshot::Receiver<Result<(), InvocationError>>> = None;
        let mut restart_count: u32 = 0;

        loop {
            tokio::select! {
                // ── Clean shutdown ────────────────────────────────────────────
                _ = shutdown_token.cancelled() => {
                    log::info!("[layer] Reader task: shutdown requested, exiting cleanly.");
                    let mut pending = self.inner.pending.lock().await;
                    for (_, tx) in pending.drain() {
                        let _ = tx.send(Err(InvocationError::Dropped));
                    }
                    return;
                }

                // ── reader_loop ───────────────────────────────────────────────
                _ = self.reader_loop(
                        rh, fk, ak, sid,
                        restart_init_rx.take(),
                        &mut new_conn_rx, &mut network_hint_rx,
                    ) => {}
            }

            // If we reach here, reader_loop returned without a shutdown signal.
            // This should never happen in normal operation — treat it as a fault.
            if shutdown_token.is_cancelled() {
                log::info!("[layer] Reader task: exiting after loop (shutdown).");
                return;
            }

            restart_count += 1;
            log::error!(
                "[layer] Reader loop exited unexpectedly (restart #{restart_count}) —                  supervisor reconnecting …"
            );

            // Step 1: drain all pending RPCs so callers don't hang.
            {
                let mut pending = self.inner.pending.lock().await;
                for (_, tx) in pending.drain() {
                    let _ = tx.send(Err(InvocationError::Io(std::io::Error::new(
                        std::io::ErrorKind::ConnectionReset,
                        "reader task restarted",
                    ))));
                }
            }

            // Step 2: reconnect with exponential backoff, honouring shutdown.
            let mut delay_ms = RECONNECT_BASE_MS;
            let new_conn = loop {
                log::info!("[layer] Supervisor: reconnecting in {delay_ms} ms …");
                tokio::select! {
                    _ = shutdown_token.cancelled() => {
                        log::info!("[layer] Supervisor: shutdown during reconnect, exiting.");
                        return;
                    }
                    _ = sleep(Duration::from_millis(delay_ms)) => {}
                }

                // do_reconnect ignores both params (_old_auth_key, _old_frame_kind) —
                // it re-reads everything from ClientInner. rh/fk/ak/sid were moved
                // into reader_loop, so we pass dummies here; fresh values come back
                // from the Ok result and replace them below.
                let dummy_ak = [0u8; 256];
                let dummy_fk = FrameKind::Abridged;
                match self.do_reconnect(&dummy_ak, &dummy_fk).await {
                    Ok(conn) => break conn,
                    Err(e)   => {
                        log::warn!("[layer] Supervisor: reconnect failed ({e})");
                        let next = (delay_ms * 2).min(RECONNECT_MAX_SECS * 1_000);
                        delay_ms = jitter_delay(next).as_millis() as u64;
                    }
                }
            };

            let (new_rh, new_fk, new_ak, new_sid) = new_conn;
            rh = new_rh; fk = new_fk; ak = new_ak; sid = new_sid;

            // Step 3: spawn init_connection (cannot await inline — reader must
            // be running to route the RPC response, or we deadlock).
            let (init_tx, init_rx) = oneshot::channel();
            let c   = self.clone();
            let utx = self.inner.update_tx.clone();
            tokio::spawn(async move {
                // Respect FLOOD_WAIT (same as do_reconnect_loop).
                let result = loop {
                    match c.init_connection().await {
                        Ok(()) => break Ok(()),
                        Err(InvocationError::Rpc(ref r))
                            if r.flood_wait_seconds().is_some() =>
                        {
                            let secs = r.flood_wait_seconds().unwrap();
                            log::warn!(
                                "[layer] Supervisor init_connection FLOOD_WAIT_{secs} — waiting"
                            );
                            sleep(Duration::from_secs(secs + 1)).await;
                        }
                        Err(e) => break Err(e),
                    }
                };
                if result.is_ok() {
                    if let Ok(missed) = c.get_difference().await {
                        for u in missed { let _ = utx.send(u); }
                    }
                }
                let _ = init_tx.send(result);
            });
            restart_init_rx = Some(init_rx);

            log::info!("[layer] Supervisor: restarting reader loop (restart #{restart_count}) …");
            // Loop back → reader_loop restarts with the fresh connection.
        }
    }

    async fn reader_loop(
        &self,
        mut rh:          OwnedReadHalf,
        mut fk:          FrameKind,
        mut ak:          [u8; 256],
        mut sid:         i64,
        // When Some, the supervisor has already spawned init_connection on our
        // behalf (supervisor restart path). On first start this is None.
        initial_init_rx:  Option<oneshot::Receiver<Result<(), InvocationError>>>,
        new_conn_rx:      &mut mpsc::UnboundedReceiver<(OwnedReadHalf, FrameKind, [u8; 256], i64)>,
        network_hint_rx:  &mut mpsc::UnboundedReceiver<()>,
    ) {
        // Tracks an in-flight init_connection task spawned after every reconnect.
        // The reader loop must keep routing frames while we wait so the RPC
        // response can reach its oneshot sender (otherwise → 30 s self-deadlock).
        // If init fails we re-enter the reconnect loop immediately.
        let mut init_rx: Option<oneshot::Receiver<Result<(), InvocationError>>> = initial_init_rx;
        // How many consecutive init_connection failures have occurred on the
        // *current* auth key.  We retry with the same key up to 2 times before
        // assuming the key is stale and clearing it for a fresh DH handshake.
        // This prevents a transient 30 s timeout from nuking a valid session.
        let mut init_fail_count: u32 = 0;

        loop {
            tokio::select! {
                // ── Normal frame (or application-level keepalive timeout) ─────
                outcome = recv_frame_with_keepalive(&mut rh, &fk, self, &ak) => {
                    match outcome {
                        FrameOutcome::Frame(mut raw) => {
                            let msg = match EncryptedSession::decrypt_frame(&ak, sid, &mut raw) {
                                Ok(m)  => m,
                                Err(e) => { log::warn!("[layer] Decrypt error: {e:?}"); continue; }
                            };
                            if msg.salt != 0 {
                                self.inner.writer.lock().await.enc.salt = msg.salt;
                            }
                            self.route_frame(msg.body).await;
                        }

                        FrameOutcome::Error(e) => {
                            log::warn!("[layer] Reader: connection error: {e}");
                            drop(init_rx.take()); // discard any in-flight init

                            // Fail all in-flight RPCs immediately so AutoSleep
                            // retries them as soon as we reconnect.
                            {
                                let mut pending = self.inner.pending.lock().await;
                                let msg = e.to_string();
                                for (_, tx) in pending.drain() {
                                    let _ = tx.send(Err(InvocationError::Io(
                                        std::io::Error::new(
                                            std::io::ErrorKind::ConnectionReset, msg.clone()))));
                                }
                            }

                            match self.do_reconnect_loop(
                                RECONNECT_BASE_MS, &mut rh, &mut fk, &mut ak, &mut sid,
                                network_hint_rx,
                            ).await {
                                Some(rx) => { init_rx = Some(rx); }
                                None     => return, // shutdown requested
                            }
                        }

                        FrameOutcome::Keepalive => {} // ping sent successfully; loop
                    }
                }

                // ── DC migration / deliberate reconnect ──────────────────────
                maybe = new_conn_rx.recv() => {
                    if let Some((new_rh, new_fk, new_ak, new_sid)) = maybe {
                        rh = new_rh; fk = new_fk; ak = new_ak; sid = new_sid;
                        log::info!("[layer] Reader: switched to new connection.");
                    } else {
                        break; // reconnect_tx dropped → client is shutting down
                    }
                }


                // ── init_connection result (polled only when Some) ────────────
                init_result = async { init_rx.as_mut().unwrap().await }, if init_rx.is_some() => {
                    init_rx = None;
                    match init_result {
                        Ok(Ok(())) => {
                            init_fail_count = 0;
                            // Fully initialised. Persist updated DC/session state.
                            // Retry up to 3 times — a transient I/O error (e.g. the
                            // filesystem briefly unavailable on Android) should not
                            // cause the auth key to diverge from disk permanently.
                            log::info!("[layer] Reconnected to Telegram ✓ — session live, replaying missed updates …");
                            for attempt in 1u8..=3 {
                                match self.save_session().await {
                                    Ok(()) => break,
                                    Err(e) if attempt < 3 => {
                                        log::warn!(
                                            "[layer] save_session failed (attempt {attempt}/3): {e}"
                                        );
                                        sleep(Duration::from_millis(500)).await;
                                    }
                                    Err(e) => {
                                        log::error!(
                                            "[layer] save_session permanently failed after 3 attempts: {e}"
                                        );
                                    }
                                }
                            }
                        }

                        Ok(Err(e)) => {
                            // TCP connected but init RPC failed.
                            //
                            // Grammers' approach: only treat the key as stale when
                            // Telegram sends a definitive "bad auth key" signal —
                            // a -404 transport error code.  A 30 s timeout is almost
                            // always a transient network issue and must NOT clear the
                            // session (doing so creates a brand-new session in Telegram's
                            // active devices list and orphans the old one).
                            let key_is_stale = match &e {
                                // Telegram's abridged-transport error code for invalid key
                                InvocationError::Rpc(r) if r.code == -404 => true,
                                // Early EOF immediately after connecting = server rejected key
                                InvocationError::Io(io) if io.kind() == std::io::ErrorKind::UnexpectedEof
                                    || io.kind() == std::io::ErrorKind::ConnectionReset => true,
                                // 30 s timeout → transient; keep the key and retry
                                _ => false,
                            };

                            if key_is_stale {
                                log::warn!(
                                    "[layer] init_connection failed with definitive bad-key signal ({e}) \
                                     — clearing auth key for fresh DH …"
                                );
                                init_fail_count = 0;
                                let home_dc_id = *self.inner.home_dc_id.lock().await;
                                let mut opts = self.inner.dc_options.lock().await;
                                if let Some(entry) = opts.get_mut(&home_dc_id) {
                                    entry.auth_key = None;
                                }
                            } else {
                                init_fail_count += 1;
                                log::warn!(
                                    "[layer] init_connection failed transiently (attempt {init_fail_count}, {e}) \
                                     — retrying with same key …"
                                );
                            }
                            {
                                let mut pending = self.inner.pending.lock().await;
                                let msg = e.to_string();
                                for (_, tx) in pending.drain() {
                                    let _ = tx.send(Err(InvocationError::Io(
                                        std::io::Error::new(
                                            std::io::ErrorKind::ConnectionReset, msg.clone()))));
                                }
                            }
                            match self.do_reconnect_loop(
                                0, &mut rh, &mut fk, &mut ak, &mut sid, network_hint_rx,
                            ).await {
                                Some(rx) => { init_rx = Some(rx); }
                                None     => return,
                            }
                        }

                        Err(_) => {
                            // init task was dropped (shouldn't normally happen).
                            log::warn!("[layer] init_connection task dropped unexpectedly, reconnecting …");
                            match self.do_reconnect_loop(
                                RECONNECT_BASE_MS, &mut rh, &mut fk, &mut ak, &mut sid,
                                network_hint_rx,
                            ).await {
                                Some(rx) => { init_rx = Some(rx); }
                                None     => return,
                            }
                        }
                    }
                }
            }
        }
    }

    /// Route a decrypted MTProto frame body to either a pending RPC caller or update_tx.
    async fn route_frame(&self, body: Vec<u8>) {
        if body.len() < 4 { return; }
        let cid = u32::from_le_bytes(body[..4].try_into().unwrap());

        match cid {
            ID_RPC_RESULT => {
                // body[4..12] = req_msg_id of the request this is answering
                if body.len() < 12 { return; }
                let req_msg_id = i64::from_le_bytes(body[4..12].try_into().unwrap());
                let inner = body[12..].to_vec();
                // Recursively unwrap the inner payload (handles gzip, containers, etc.)
                let result = unwrap_envelope(inner);
                if let Some(tx) = self.inner.pending.lock().await.remove(&req_msg_id) {
                    let to_send = match result {
                        Ok(EnvelopeResult::Payload(p)) => Ok(p),
                        Ok(EnvelopeResult::Updates(us)) => {
                            // Write RPCs return Updates — forward them and signal success
                            for u in us { let _ = self.inner.update_tx.send(u); }
                            Ok(vec![])
                        }
                        Ok(EnvelopeResult::None) => Ok(vec![]),
                        Err(e) => Err(e),
                    };
                    let _ = tx.send(to_send);
                }
            }
            // ID_RPC_ERROR only appears as the INNER body of ID_RPC_RESULT.
            // At top level it has no req_msg_id — ignore it here (the ID_RPC_RESULT
            // handler above uses unwrap_envelope which correctly returns Err for it).
            ID_RPC_ERROR => {
                log::warn!("[layer] Unexpected top-level rpc_error (no pending target)");
            }
            ID_MSG_CONTAINER => {
                // Container of multiple inner messages — recurse into each
                if body.len() < 8 { return; }
                let count = u32::from_le_bytes(body[4..8].try_into().unwrap()) as usize;
                let mut pos = 8usize;
                for _ in 0..count {
                    if pos + 16 > body.len() { break; }
                    let inner_len = u32::from_le_bytes(body[pos + 12..pos + 16].try_into().unwrap()) as usize;
                    pos += 16;
                    if pos + inner_len > body.len() { break; }
                    let inner = body[pos..pos + inner_len].to_vec();
                    pos += inner_len;
                    Box::pin(self.route_frame(inner)).await;
                }
            }
            ID_GZIP_PACKED => {
                let bytes = tl_read_bytes(&body[4..]).unwrap_or_default();
                if let Ok(inflated) = gz_inflate(&bytes) {
                    Box::pin(self.route_frame(inflated)).await;
                }
            }
            ID_BAD_SERVER_SALT => {
                // Server is telling us to use a different salt — update writer enc
                if body.len() >= 16 {
                    let new_salt = i64::from_le_bytes(body[8..16].try_into().unwrap());
                    self.inner.writer.lock().await.enc.salt = new_salt;
                }
            }
            ID_PONG => {
                // Bare Pong is the server's reply to our Ping — NOT wrapped in rpc_result.
                // pong#347773c5 msg_id:long ping_id:long
                //   body[4..12] = msg_id of the original Ping → key in pending map
                if body.len() >= 20 {
                    let ping_msg_id = i64::from_le_bytes(body[4..12].try_into().unwrap());
                    if let Some(tx) = self.inner.pending.lock().await.remove(&ping_msg_id) {
                        let _ = tx.send(Ok(body));
                    }
                }
            }
            ID_UPDATES | ID_UPDATE_SHORT | ID_UPDATES_COMBINED
            | ID_UPDATE_SHORT_MSG | ID_UPDATE_SHORT_CHAT_MSG
            | ID_UPDATES_TOO_LONG => {
                for u in update::parse_updates(&body) {
                    let _ = self.inner.update_tx.send(u);
                }
            }
            // Silently acknowledged service messages — pong, acks, etc.
            _ => {}
        }
    }

    /// Loops with exponential backoff until a TCP+DH reconnect succeeds, then
    /// spawns `init_connection` in a background task and returns a oneshot
    /// receiver for its result.
    ///
    /// - `initial_delay_ms = RECONNECT_BASE_MS` for a fresh disconnect.
    /// - `initial_delay_ms = 0` when TCP already worked but init failed — we
    ///   want to retry init immediately rather than waiting another full backoff.
    ///
    /// Returns `None` if the shutdown token fires (caller should exit).
    async fn do_reconnect_loop(
        &self,
        initial_delay_ms: u64,
        rh:  &mut OwnedReadHalf,
        fk:  &mut FrameKind,
        ak:  &mut [u8; 256],
        sid: &mut i64,
        network_hint_rx: &mut mpsc::UnboundedReceiver<()>,
    ) -> Option<oneshot::Receiver<Result<(), InvocationError>>> {
        let mut delay_ms = if initial_delay_ms == 0 {
            // Caller explicitly requests an immediate first attempt (e.g. init
            // failed but TCP is up — no reason to wait before the next try).
            0
        } else {
            initial_delay_ms.max(RECONNECT_BASE_MS)
        };
        loop {
            log::info!("[layer] Reconnecting in {delay_ms} ms …");
            tokio::select! {
                _ = sleep(Duration::from_millis(delay_ms)) => {}
                hint = network_hint_rx.recv() => {
                    if hint.is_none() { return None; } // shutdown
                    log::info!("[layer] Network hint → skipping backoff, reconnecting now");
                }
            }

            match self.do_reconnect(ak, fk).await {
                Ok((new_rh, new_fk, new_ak, new_sid)) => {
                    *rh = new_rh; *fk = new_fk; *ak = new_ak; *sid = new_sid;
                    log::info!("[layer] TCP reconnected ✓ — initialising session …");

                    // Spawn init_connection. MUST NOT be awaited inline — the
                    // reader loop must resume so it can route the RPC response.
                    // We give back a oneshot so the reader can act on failure.
                    let (init_tx, init_rx) = oneshot::channel();
                    let c   = self.clone();
                    let utx = self.inner.update_tx.clone();
                    tokio::spawn(async move {
                        // Respect FLOOD_WAIT before sending the result back.
                        // Without this, a FLOOD_WAIT from Telegram during init
                        // would immediately re-trigger another reconnect attempt,
                        // which would itself hit FLOOD_WAIT — a ban spiral.
                        let result = loop {
                            match c.init_connection().await {
                                Ok(()) => break Ok(()),
                                Err(InvocationError::Rpc(ref r))
                                    if r.flood_wait_seconds().is_some() =>
                                {
                                    let secs = r.flood_wait_seconds().unwrap();
                                    log::warn!(
                                        "[layer] init_connection FLOOD_WAIT_{secs} —                                          waiting before retry"
                                    );
                                    sleep(Duration::from_secs(secs + 1)).await;
                                    // loop and retry init_connection
                                }
                                Err(e) => break Err(e),
                            }
                        };
                        if result.is_ok() {
                            // Replay any updates missed during the outage.
                            if let Ok(missed) = c.get_difference().await {
                                for u in missed { let _ = utx.send(u); }
                            }
                        }
                        let _ = init_tx.send(result);
                    });
                    return Some(init_rx);
                }
                Err(e) => {
                    log::warn!("[layer] Reconnect attempt failed: {e}");
                    // Cap at max, then apply ±20 % jitter to avoid thundering herd.
                    // Ensure the delay always advances by at least RECONNECT_BASE_MS
                    // so a 0 initial delay on the first attempt doesn't spin-loop.
                    let next = (delay_ms.saturating_mul(2).max(RECONNECT_BASE_MS))
                        .min(RECONNECT_MAX_SECS * 1_000);
                    delay_ms = jitter_delay(next).as_millis() as u64;
                }
            }
        }
    }

    /// Reconnect to the home DC, replace the writer, and return the new read half.
    async fn do_reconnect(
        &self,
        _old_auth_key: &[u8; 256],
        _old_frame_kind: &FrameKind,
    ) -> Result<(OwnedReadHalf, FrameKind, [u8; 256], i64), InvocationError> {
        let home_dc_id = *self.inner.home_dc_id.lock().await;
        let (addr, saved_key, first_salt, time_offset) = {
            let opts = self.inner.dc_options.lock().await;
            match opts.get(&home_dc_id) {
                Some(e) => (e.addr.clone(), e.auth_key, e.first_salt, e.time_offset),
                None    => ("149.154.167.51:443".to_string(), None, 0, 0),
            }
        };
        let socks5    = self.inner.socks5.clone();
        let transport = self.inner.transport.clone();

        let new_conn = if let Some(key) = saved_key {
            log::info!("[layer] Reconnecting to DC{home_dc_id} with saved key …");
            match Connection::connect_with_key(
                &addr, key, first_salt, time_offset, socks5.as_ref(), &transport,
            ).await {
                Ok(c)   => c,
                Err(e2) => {
                    log::warn!("[layer] connect_with_key failed ({e2}), fresh DH …");
                    Connection::connect_raw(&addr, socks5.as_ref(), &transport).await?
                }
            }
        } else {
            Connection::connect_raw(&addr, socks5.as_ref(), &transport).await?
        };

        let (new_writer, new_read, new_fk) = new_conn.into_writer();
        let new_ak  = new_writer.enc.auth_key_bytes();
        let new_sid = new_writer.enc.session_id();
        *self.inner.writer.lock().await = new_writer;

        // NOTE: init_connection() is intentionally NOT called here.
        //
        // do_reconnect() is always called from inside the reader loop's select!,
        // which means the reader task is blocked while this function runs.
        // init_connection() sends an RPC and awaits the response — but only the
        // reader task can route that response back to the pending caller.
        // Calling it here creates a self-deadlock that times out after 30 s.
        //
        // Instead, callers are responsible for spawning init_connection() in a
        // separate task AFTER the reader loop has resumed and can process frames.

        Ok((new_read, new_fk, new_ak, new_sid))
    }

    // ── Messaging ──────────────────────────────────────────────────────────

    /// Send a text message. Use `"me"` for Saved Messages.
    pub async fn send_message(&self, peer: &str, text: &str) -> Result<(), InvocationError> {
        let p = self.resolve_peer(peer).await?;
        self.send_message_to_peer(p, text).await
    }

    /// Send a message to an already-resolved peer (plain text shorthand).
    pub async fn send_message_to_peer(
        &self,
        peer: tl::enums::Peer,
        text: &str,
    ) -> Result<(), InvocationError> {
        self.send_message_to_peer_ex(peer, &InputMessage::text(text)).await
    }

    /// Send a message with full [`InputMessage`] options.
    pub async fn send_message_to_peer_ex(
        &self,
        peer: tl::enums::Peer,
        msg:  &InputMessage,
    ) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::SendMessage {
            no_webpage:               msg.no_webpage,
            silent:                   msg.silent,
            background:               msg.background,
            clear_draft:              msg.clear_draft,
            noforwards:               false,
            update_stickersets_order: false,
            invert_media:             false,
            allow_paid_floodskip:     false,
            peer:                     input_peer,
            reply_to:                 msg.reply_header(),
            message:                  msg.text.clone(),
            random_id:                random_i64(),
            reply_markup:             msg.reply_markup.clone(),
            entities:                 msg.entities.clone(),
            schedule_date:            msg.schedule_date,
            schedule_repeat_period:   None,
            send_as:                  None,
            quick_reply_shortcut:     None,
            effect:                   None,
            allow_paid_stars:         None,
            suggested_post:           None,
        };
        self.rpc_write(&req).await
    }

    /// Send directly to Saved Messages.
    pub async fn send_to_self(&self, text: &str) -> Result<(), InvocationError> {
        let req = tl::functions::messages::SendMessage {
            no_webpage:               false,
            silent:                   false,
            background:               false,
            clear_draft:              false,
            noforwards:               false,
            update_stickersets_order: false,
            invert_media:             false,
            allow_paid_floodskip:     false,
            peer:                     tl::enums::InputPeer::PeerSelf,
            reply_to:                 None,
            message:                  text.to_string(),
            random_id:                random_i64(),
            reply_markup:             None,
            entities:                 None,
            schedule_date:            None,
            schedule_repeat_period:   None,
            send_as:                  None,
            quick_reply_shortcut:     None,
            effect:                   None,
            allow_paid_stars:         None,
            suggested_post:           None,
        };
        self.rpc_write(&req).await
    }

    /// Edit an existing message.
    pub async fn edit_message(
        &self,
        peer:       tl::enums::Peer,
        message_id: i32,
        new_text:   &str,
    ) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::EditMessage {
            no_webpage:    false,
            invert_media:  false,
            peer:          input_peer,
            id:            message_id,
            message:       Some(new_text.to_string()),
            media:         None,
            reply_markup:  None,
            entities:      None,
            schedule_date: None,
            schedule_repeat_period: None,
            quick_reply_shortcut_id: None,
        };
        self.rpc_write(&req).await
    }

    /// Forward messages from `source` to `destination`.
    pub async fn forward_messages(
        &self,
        destination: tl::enums::Peer,
        message_ids: &[i32],
        source:      tl::enums::Peer,
    ) -> Result<(), InvocationError> {
        let cache = self.inner.peer_cache.lock().await;
        let to_peer   = cache.peer_to_input(&destination);
        let from_peer = cache.peer_to_input(&source);
        drop(cache);

        let req = tl::functions::messages::ForwardMessages {
            silent:            false,
            background:        false,
            with_my_score:     false,
            drop_author:       false,
            drop_media_captions: false,
            noforwards:        false,
            from_peer:         from_peer,
            id:                message_ids.to_vec(),
            random_id:         (0..message_ids.len()).map(|_| random_i64()).collect(),
            to_peer:           to_peer,
            top_msg_id:        None,
            reply_to:          None,
            schedule_date:     None,
            schedule_repeat_period: None,
            send_as:           None,
            quick_reply_shortcut: None,
            effect:            None,
            video_timestamp:   None,
            allow_paid_stars:  None,
            allow_paid_floodskip: false,
            suggested_post:    None,
        };
        self.rpc_write(&req).await
    }

    /// Delete messages by ID.
    pub async fn delete_messages(&self, message_ids: Vec<i32>, revoke: bool) -> Result<(), InvocationError> {
        let req = tl::functions::messages::DeleteMessages { revoke, id: message_ids };
        self.rpc_write(&req).await
    }

    /// Get messages by their IDs from a peer.
    pub async fn get_messages_by_id(
        &self,
        peer: tl::enums::Peer,
        ids:  &[i32],
    ) -> Result<Vec<update::IncomingMessage>, InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let id_list: Vec<tl::enums::InputMessage> = ids.iter()
            .map(|&id| tl::enums::InputMessage::Id(tl::types::InputMessageId { id }))
            .collect();
        let req  = tl::functions::channels::GetMessages {
            channel: match &input_peer {
                tl::enums::InputPeer::Channel(c) =>
                    tl::enums::InputChannel::InputChannel(tl::types::InputChannel {
                        channel_id: c.channel_id, access_hash: c.access_hash
                    }),
                _ => return self.get_messages_user(input_peer, id_list).await,
            },
            id: id_list,
        };
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let msgs = match tl::enums::messages::Messages::deserialize(&mut cur)? {
            tl::enums::messages::Messages::Messages(m) => m.messages,
            tl::enums::messages::Messages::Slice(m)    => m.messages,
            tl::enums::messages::Messages::ChannelMessages(m) => m.messages,
            tl::enums::messages::Messages::NotModified(_) => vec![],
        };
        Ok(msgs.into_iter().map(update::IncomingMessage::from_raw).collect())
    }

    async fn get_messages_user(
        &self,
        _peer: tl::enums::InputPeer,
        ids:   Vec<tl::enums::InputMessage>,
    ) -> Result<Vec<update::IncomingMessage>, InvocationError> {
        let req = tl::functions::messages::GetMessages { id: ids };
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let msgs = match tl::enums::messages::Messages::deserialize(&mut cur)? {
            tl::enums::messages::Messages::Messages(m) => m.messages,
            tl::enums::messages::Messages::Slice(m)    => m.messages,
            tl::enums::messages::Messages::ChannelMessages(m) => m.messages,
            tl::enums::messages::Messages::NotModified(_) => vec![],
        };
        Ok(msgs.into_iter().map(update::IncomingMessage::from_raw).collect())
    }

    /// Get the pinned message in a chat.
    pub async fn get_pinned_message(
        &self,
        peer: tl::enums::Peer,
    ) -> Result<Option<update::IncomingMessage>, InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::Search {
            peer: input_peer,
            q: String::new(),
            from_id: None,
            saved_peer_id: None,
            saved_reaction: None,
            top_msg_id: None,
            filter: tl::enums::MessagesFilter::InputMessagesFilterPinned,
            min_date: 0,
            max_date: 0,
            offset_id: 0,
            add_offset: 0,
            limit: 1,
            max_id: 0,
            min_id: 0,
            hash: 0,
        };
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let msgs = match tl::enums::messages::Messages::deserialize(&mut cur)? {
            tl::enums::messages::Messages::Messages(m) => m.messages,
            tl::enums::messages::Messages::Slice(m)    => m.messages,
            tl::enums::messages::Messages::ChannelMessages(m) => m.messages,
            tl::enums::messages::Messages::NotModified(_) => vec![],
        };
        Ok(msgs.into_iter().next().map(update::IncomingMessage::from_raw))
    }

    /// Pin a message in a chat.
    pub async fn pin_message(
        &self,
        peer:       tl::enums::Peer,
        message_id: i32,
        silent:     bool,
        unpin:      bool,
        pm_oneside: bool,
    ) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::UpdatePinnedMessage {
            silent,
            unpin,
            pm_oneside,
            peer: input_peer,
            id:   message_id,
        };
        self.rpc_write(&req).await
    }

    /// Unpin a specific message.
    pub async fn unpin_message(
        &self,
        peer:       tl::enums::Peer,
        message_id: i32,
    ) -> Result<(), InvocationError> {
        self.pin_message(peer, message_id, true, true, false).await
    }

    /// Unpin all messages in a chat.
    pub async fn unpin_all_messages(&self, peer: tl::enums::Peer) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::UnpinAllMessages {
            peer:      input_peer,
            top_msg_id: None,
            saved_peer_id: None,
        };
        self.rpc_write(&req).await
    }

    // ── Message search ─────────────────────────────────────────────────────

    /// Search messages in a chat.
    pub async fn search_messages(
        &self,
        peer:  tl::enums::Peer,
        query: &str,
        limit: i32,
    ) -> Result<Vec<update::IncomingMessage>, InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::Search {
            peer:         input_peer,
            q:            query.to_string(),
            from_id:      None,
            saved_peer_id: None,
            saved_reaction: None,
            top_msg_id:   None,
            filter:       tl::enums::MessagesFilter::InputMessagesFilterEmpty,
            min_date:     0,
            max_date:     0,
            offset_id:    0,
            add_offset:   0,
            limit,
            max_id:       0,
            min_id:       0,
            hash:         0,
        };
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let msgs = match tl::enums::messages::Messages::deserialize(&mut cur)? {
            tl::enums::messages::Messages::Messages(m) => m.messages,
            tl::enums::messages::Messages::Slice(m)    => m.messages,
            tl::enums::messages::Messages::ChannelMessages(m) => m.messages,
            tl::enums::messages::Messages::NotModified(_) => vec![],
        };
        Ok(msgs.into_iter().map(update::IncomingMessage::from_raw).collect())
    }

    /// Search messages globally across all chats.
    pub async fn search_global(
        &self,
        query: &str,
        limit: i32,
    ) -> Result<Vec<update::IncomingMessage>, InvocationError> {
        let req = tl::functions::messages::SearchGlobal {
            broadcasts_only: false,
            groups_only:     false,
            users_only:      false,
            folder_id:       None,
            q:               query.to_string(),
            filter:          tl::enums::MessagesFilter::InputMessagesFilterEmpty,
            min_date:        0,
            max_date:        0,
            offset_rate:     0,
            offset_peer:     tl::enums::InputPeer::Empty,
            offset_id:       0,
            limit,
        };
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let msgs = match tl::enums::messages::Messages::deserialize(&mut cur)? {
            tl::enums::messages::Messages::Messages(m) => m.messages,
            tl::enums::messages::Messages::Slice(m)    => m.messages,
            tl::enums::messages::Messages::ChannelMessages(m) => m.messages,
            tl::enums::messages::Messages::NotModified(_) => vec![],
        };
        Ok(msgs.into_iter().map(update::IncomingMessage::from_raw).collect())
    }

    // ── Scheduled messages ─────────────────────────────────────────────────

    /// Retrieve all scheduled messages in a chat.
    ///
    /// Scheduled messages are messages set to be sent at a future time using
    /// [`InputMessage::schedule_date`].  Returns them newest-first.
    ///
    /// # Example
    /// ```rust,no_run
    /// # async fn f(client: layer_client::Client, peer: layer_tl_types::enums::Peer) -> Result<(), Box<dyn std::error::Error>> {
    /// let scheduled = client.get_scheduled_messages(peer).await?;
    /// for msg in &scheduled {
    ///     println!("Scheduled: {:?} at {:?}", msg.text(), msg.date());
    /// }
    /// # Ok(()) }
    /// ```
    pub async fn get_scheduled_messages(
        &self,
        peer: tl::enums::Peer,
    ) -> Result<Vec<update::IncomingMessage>, InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::GetScheduledHistory {
            peer: input_peer,
            hash: 0,
        };
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let msgs = match tl::enums::messages::Messages::deserialize(&mut cur)? {
            tl::enums::messages::Messages::Messages(m)        => m.messages,
            tl::enums::messages::Messages::Slice(m)           => m.messages,
            tl::enums::messages::Messages::ChannelMessages(m) => m.messages,
            tl::enums::messages::Messages::NotModified(_)     => vec![],
        };
        Ok(msgs.into_iter().map(update::IncomingMessage::from_raw).collect())
    }

    /// Delete one or more scheduled messages by their IDs.
    pub async fn delete_scheduled_messages(
        &self,
        peer: tl::enums::Peer,
        ids:  Vec<i32>,
    ) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::DeleteScheduledMessages {
            peer: input_peer,
            id:   ids,
        };
        self.rpc_write(&req).await
    }

    // ── Callback / Inline Queries ──────────────────────────────────────────

    pub async fn answer_callback_query(
        &self,
        query_id: i64,
        text:     Option<&str>,
        alert:    bool,
    ) -> Result<bool, InvocationError> {
        let req = tl::functions::messages::SetBotCallbackAnswer {
            alert,
            query_id,
            message:    text.map(|s| s.to_string()),
            url:        None,
            cache_time: 0,
        };
        let body = self.rpc_call_raw(&req).await?;
        Ok(!body.is_empty())
    }

    pub async fn answer_inline_query(
        &self,
        query_id:    i64,
        results:     Vec<tl::enums::InputBotInlineResult>,
        cache_time:  i32,
        is_personal: bool,
        next_offset: Option<String>,
    ) -> Result<bool, InvocationError> {
        let req = tl::functions::messages::SetInlineBotResults {
            gallery:        false,
            private:        is_personal,
            query_id,
            results,
            cache_time,
            next_offset,
            switch_pm:      None,
            switch_webview: None,
        };
        let body = self.rpc_call_raw(&req).await?;
        Ok(!body.is_empty())
    }

    // ── Dialogs ────────────────────────────────────────────────────────────

    /// Fetch up to `limit` dialogs, most recent first. Populates entity/message.
    pub async fn get_dialogs(&self, limit: i32) -> Result<Vec<Dialog>, InvocationError> {
        let req = tl::functions::messages::GetDialogs {
            exclude_pinned: false,
            folder_id:      None,
            offset_date:    0,
            offset_id:      0,
            offset_peer:    tl::enums::InputPeer::Empty,
            limit,
            hash:           0,
        };

        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let raw = match tl::enums::messages::Dialogs::deserialize(&mut cur)? {
            tl::enums::messages::Dialogs::Dialogs(d) => d,
            tl::enums::messages::Dialogs::Slice(d)   => tl::types::messages::Dialogs {
                dialogs: d.dialogs, messages: d.messages, chats: d.chats, users: d.users,
            },
            tl::enums::messages::Dialogs::NotModified(_) => return Ok(vec![]),
        };

        // Build message map
        let msg_map: HashMap<i32, tl::enums::Message> = raw.messages.into_iter()
            .filter_map(|m| {
                let id = match &m {
                    tl::enums::Message::Message(x) => x.id,
                    tl::enums::Message::Service(x) => x.id,
                    tl::enums::Message::Empty(x)   => x.id,
                };
                Some((id, m))
            })
            .collect();

        // Build user map
        let user_map: HashMap<i64, tl::enums::User> = raw.users.into_iter()
            .filter_map(|u| {
                if let tl::enums::User::User(ref uu) = u { Some((uu.id, u)) } else { None }
            })
            .collect();

        // Build chat map
        let chat_map: HashMap<i64, tl::enums::Chat> = raw.chats.into_iter()
            .filter_map(|c| {
                let id = match &c {
                    tl::enums::Chat::Chat(x)             => x.id,
                    tl::enums::Chat::Forbidden(x)    => x.id,
                    tl::enums::Chat::Channel(x)          => x.id,
                    tl::enums::Chat::ChannelForbidden(x) => x.id,
                    tl::enums::Chat::Empty(x)            => x.id,
                };
                Some((id, c))
            })
            .collect();

        // Cache peers for future access_hash lookups
        {
            let u_list: Vec<tl::enums::User> = user_map.values().cloned().collect();
            let c_list: Vec<tl::enums::Chat> = chat_map.values().cloned().collect();
            self.cache_users_slice(&u_list).await;
            self.cache_chats_slice(&c_list).await;
        }

        let result = raw.dialogs.into_iter().map(|d| {
            let top_id = match &d { tl::enums::Dialog::Dialog(x) => x.top_message, _ => 0 };
            let peer   = match &d { tl::enums::Dialog::Dialog(x) => Some(&x.peer), _ => None };

            let message = msg_map.get(&top_id).cloned();
            let entity = peer.and_then(|p| match p {
                tl::enums::Peer::User(u) => user_map.get(&u.user_id).cloned(),
                _ => None,
            });
            let chat = peer.and_then(|p| match p {
                tl::enums::Peer::Chat(c)    => chat_map.get(&c.chat_id).cloned(),
                tl::enums::Peer::Channel(c) => chat_map.get(&c.channel_id).cloned(),
                _ => None,
            });

            Dialog { raw: d, message, entity, chat }
        }).collect();

        Ok(result)
    }

    /// Internal helper: fetch dialogs with a custom GetDialogs request.
    #[allow(dead_code)]
    async fn get_dialogs_raw(
        &self,
        req: tl::functions::messages::GetDialogs,
    ) -> Result<Vec<Dialog>, InvocationError> {
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let raw = match tl::enums::messages::Dialogs::deserialize(&mut cur)? {
            tl::enums::messages::Dialogs::Dialogs(d) => d,
            tl::enums::messages::Dialogs::Slice(d)   => tl::types::messages::Dialogs {
                dialogs: d.dialogs, messages: d.messages, chats: d.chats, users: d.users,
            },
            tl::enums::messages::Dialogs::NotModified(_) => return Ok(vec![]),
        };

        let msg_map: HashMap<i32, tl::enums::Message> = raw.messages.into_iter()
            .filter_map(|m| {
                let id = match &m {
                    tl::enums::Message::Message(x) => x.id,
                    tl::enums::Message::Service(x) => x.id,
                    tl::enums::Message::Empty(x)   => x.id,
                };
                Some((id, m))
            })
            .collect();

        let user_map: HashMap<i64, tl::enums::User> = raw.users.into_iter()
            .filter_map(|u| {
                if let tl::enums::User::User(ref uu) = u { Some((uu.id, u)) } else { None }
            })
            .collect();

        let chat_map: HashMap<i64, tl::enums::Chat> = raw.chats.into_iter()
            .filter_map(|c| {
                let id = match &c {
                    tl::enums::Chat::Chat(x)             => x.id,
                    tl::enums::Chat::Forbidden(x)    => x.id,
                    tl::enums::Chat::Channel(x)          => x.id,
                    tl::enums::Chat::ChannelForbidden(x) => x.id,
                    tl::enums::Chat::Empty(x)            => x.id,
                };
                Some((id, c))
            })
            .collect();

        {
            let u_list: Vec<tl::enums::User> = user_map.values().cloned().collect();
            let c_list: Vec<tl::enums::Chat> = chat_map.values().cloned().collect();
            self.cache_users_slice(&u_list).await;
            self.cache_chats_slice(&c_list).await;
        }

        let result = raw.dialogs.into_iter().map(|d| {
            let top_id = match &d { tl::enums::Dialog::Dialog(x) => x.top_message, _ => 0 };
            let peer   = match &d { tl::enums::Dialog::Dialog(x) => Some(&x.peer), _ => None };

            let message = msg_map.get(&top_id).cloned();
            let entity = peer.and_then(|p| match p {
                tl::enums::Peer::User(u) => user_map.get(&u.user_id).cloned(),
                _ => None,
            });
            let chat = peer.and_then(|p| match p {
                tl::enums::Peer::Chat(c)    => chat_map.get(&c.chat_id).cloned(),
                tl::enums::Peer::Channel(c) => chat_map.get(&c.channel_id).cloned(),
                _ => None,
            });

            Dialog { raw: d, message, entity, chat }
        }).collect();

        Ok(result)
    }

    /// Like `get_dialogs_raw` but also returns the total count from `messages.DialogsSlice`.
    async fn get_dialogs_raw_with_count(
        &self,
        req: tl::functions::messages::GetDialogs,
    ) -> Result<(Vec<Dialog>, Option<i32>), InvocationError> {
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let (raw, count) = match tl::enums::messages::Dialogs::deserialize(&mut cur)? {
            tl::enums::messages::Dialogs::Dialogs(d) => (d, None),
            tl::enums::messages::Dialogs::Slice(d)   => {
                let cnt = Some(d.count);
                (tl::types::messages::Dialogs {
                    dialogs: d.dialogs, messages: d.messages, chats: d.chats, users: d.users,
                }, cnt)
            }
            tl::enums::messages::Dialogs::NotModified(_) => return Ok((vec![], None)),
        };

        let msg_map: HashMap<i32, tl::enums::Message> = raw.messages.into_iter()
            .filter_map(|m| {
                let id = match &m {
                    tl::enums::Message::Message(x) => x.id,
                    tl::enums::Message::Service(x) => x.id,
                    tl::enums::Message::Empty(x)   => x.id,
                };
                Some((id, m))
            }).collect();

        let user_map: HashMap<i64, tl::enums::User> = raw.users.into_iter()
            .filter_map(|u| {
                if let tl::enums::User::User(ref uu) = u { Some((uu.id, u)) } else { None }
            }).collect();

        let chat_map: HashMap<i64, tl::enums::Chat> = raw.chats.into_iter()
            .filter_map(|c| {
                let id = match &c {
                    tl::enums::Chat::Chat(x)             => x.id,
                    tl::enums::Chat::Forbidden(x)        => x.id,
                    tl::enums::Chat::Channel(x)          => x.id,
                    tl::enums::Chat::ChannelForbidden(x) => x.id,
                    tl::enums::Chat::Empty(x)            => x.id,
                };
                Some((id, c))
            }).collect();

        {
            let u_list: Vec<tl::enums::User> = user_map.values().cloned().collect();
            let c_list: Vec<tl::enums::Chat> = chat_map.values().cloned().collect();
            self.cache_users_slice(&u_list).await;
            self.cache_chats_slice(&c_list).await;
        }

        let result = raw.dialogs.into_iter().map(|d| {
            let top_id = match &d { tl::enums::Dialog::Dialog(x) => x.top_message, _ => 0 };
            let peer   = match &d { tl::enums::Dialog::Dialog(x) => Some(&x.peer), _ => None };
            let message = msg_map.get(&top_id).cloned();
            let entity = peer.and_then(|p| match p {
                tl::enums::Peer::User(u) => user_map.get(&u.user_id).cloned(),
                _ => None,
            });
            let chat = peer.and_then(|p| match p {
                tl::enums::Peer::Chat(c)    => chat_map.get(&c.chat_id).cloned(),
                tl::enums::Peer::Channel(c) => chat_map.get(&c.channel_id).cloned(),
                _ => None,
            });
            Dialog { raw: d, message, entity, chat }
        }).collect();

        Ok((result, count))
    }

    /// Like `get_messages` but also returns the total count from `messages.Slice`.
    async fn get_messages_with_count(
        &self,
        peer:      tl::enums::InputPeer,
        limit:     i32,
        offset_id: i32,
    ) -> Result<(Vec<update::IncomingMessage>, Option<i32>), InvocationError> {
        let req = tl::functions::messages::GetHistory {
            peer, offset_id, offset_date: 0, add_offset: 0,
            limit, max_id: 0, min_id: 0, hash: 0,
        };
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let (msgs, count) = match tl::enums::messages::Messages::deserialize(&mut cur)? {
            tl::enums::messages::Messages::Messages(m) => (m.messages, None),
            tl::enums::messages::Messages::Slice(m)    => {
                let cnt = Some(m.count);
                (m.messages, cnt)
            }
            tl::enums::messages::Messages::ChannelMessages(m) => (m.messages, Some(m.count)),
            tl::enums::messages::Messages::NotModified(_) => (vec![], None),
        };
        Ok((msgs.into_iter().map(update::IncomingMessage::from_raw).collect(), count))
    }

    /// Download all bytes of a media attachment and save them to `path`.
    ///
    /// # Example
    /// ```rust,no_run
    /// # async fn f(client: layer_client::Client, msg: layer_client::update::IncomingMessage) -> Result<(), Box<dyn std::error::Error>> {
    /// if let Some(loc) = msg.download_location() {
    ///     client.download_media_to_file(loc, "/tmp/file.jpg").await?;
    /// }
    /// # Ok(()) }
    /// ```
    pub async fn download_media_to_file(
        &self,
        location: tl::enums::InputFileLocation,
        path:     impl AsRef<std::path::Path>,
    ) -> Result<(), InvocationError> {
        use tokio::io::AsyncWriteExt as _;
        let bytes = self.download_media(location).await?;
        let mut f = tokio::fs::File::create(path).await?;
        f.write_all(&bytes).await?;
        Ok(())
    }

    pub async fn delete_dialog(&self, peer: tl::enums::Peer) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::DeleteHistory {
            just_clear:  false,
            revoke:      false,
            peer:        input_peer,
            max_id:      0,
            min_date:    None,
            max_date:    None,
        };
        self.rpc_write(&req).await
    }

    /// Mark all messages in a chat as read.
    pub async fn mark_as_read(&self, peer: tl::enums::Peer) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        match &input_peer {
            tl::enums::InputPeer::Channel(c) => {
                let req = tl::functions::channels::ReadHistory {
                    channel: tl::enums::InputChannel::InputChannel(tl::types::InputChannel {
                        channel_id: c.channel_id, access_hash: c.access_hash,
                    }),
                    max_id: 0,
                };
                self.rpc_call_raw(&req).await?;
            }
            _ => {
                let req = tl::functions::messages::ReadHistory { peer: input_peer, max_id: 0 };
                self.rpc_call_raw(&req).await?;
            }
        }
        Ok(())
    }

    /// Clear unread mention markers.
    pub async fn clear_mentions(&self, peer: tl::enums::Peer) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::ReadMentions { peer: input_peer, top_msg_id: None };
        self.rpc_write(&req).await
    }

    // ── Chat actions (typing, etc) ─────────────────────────────────────────

    /// Send a chat action (typing indicator, uploading photo, etc).
    ///
    /// For "typing" use `tl::enums::SendMessageAction::Typing`.
    pub async fn send_chat_action(
        &self,
        peer:   tl::enums::Peer,
        action: tl::enums::SendMessageAction,
    ) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        let req = tl::functions::messages::SetTyping {
            peer: input_peer,
            top_msg_id: None,
            action,
        };
        self.rpc_write(&req).await
    }

    // ── Join / invite links ────────────────────────────────────────────────

    /// Join a public chat or channel by username/peer.
    pub async fn join_chat(&self, peer: tl::enums::Peer) -> Result<(), InvocationError> {
        let input_peer = self.inner.peer_cache.lock().await.peer_to_input(&peer);
        match input_peer {
            tl::enums::InputPeer::Channel(c) => {
                let req = tl::functions::channels::JoinChannel {
                    channel: tl::enums::InputChannel::InputChannel(tl::types::InputChannel {
                        channel_id: c.channel_id, access_hash: c.access_hash,
                    }),
                };
                self.rpc_call_raw(&req).await?;
            }
            tl::enums::InputPeer::Chat(c) => {
                let req = tl::functions::messages::AddChatUser {
                    chat_id:  c.chat_id,
                    user_id:  tl::enums::InputUser::UserSelf,
                    fwd_limit: 0,
                };
                self.rpc_call_raw(&req).await?;
            }
            _ => return Err(InvocationError::Deserialize("cannot join this peer type".into())),
        }
        Ok(())
    }

    /// Accept and join via an invite link.
    pub async fn accept_invite_link(&self, link: &str) -> Result<(), InvocationError> {
        let hash = Self::parse_invite_hash(link)
            .ok_or_else(|| InvocationError::Deserialize(format!("invalid invite link: {link}")))?;
        let req = tl::functions::messages::ImportChatInvite { hash: hash.to_string() };
        self.rpc_write(&req).await
    }

    /// Extract hash from `https://t.me/+HASH` or `https://t.me/joinchat/HASH`.
    pub fn parse_invite_hash(link: &str) -> Option<&str> {
        if let Some(pos) = link.find("/+") {
            return Some(&link[pos + 2..]);
        }
        if let Some(pos) = link.find("/joinchat/") {
            return Some(&link[pos + 10..]);
        }
        None
    }

    // ── Message history (paginated) ────────────────────────────────────────

    /// Fetch a page of messages from a peer's history.
    pub async fn get_messages(
        &self,
        peer:      tl::enums::InputPeer,
        limit:     i32,
        offset_id: i32,
    ) -> Result<Vec<update::IncomingMessage>, InvocationError> {
        let req = tl::functions::messages::GetHistory {
            peer, offset_id, offset_date: 0, add_offset: 0,
            limit, max_id: 0, min_id: 0, hash: 0,
        };
        let body    = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let msgs = match tl::enums::messages::Messages::deserialize(&mut cur)? {
            tl::enums::messages::Messages::Messages(m) => m.messages,
            tl::enums::messages::Messages::Slice(m)    => m.messages,
            tl::enums::messages::Messages::ChannelMessages(m) => m.messages,
            tl::enums::messages::Messages::NotModified(_) => vec![],
        };
        Ok(msgs.into_iter().map(update::IncomingMessage::from_raw).collect())
    }

    // ── Peer resolution ────────────────────────────────────────────────────

    /// Resolve a peer string to a [`tl::enums::Peer`].
    pub async fn resolve_peer(
        &self,
        peer: &str,
    ) -> Result<tl::enums::Peer, InvocationError> {
        match peer.trim() {
            "me" | "self" => Ok(tl::enums::Peer::User(tl::types::PeerUser { user_id: 0 })),
            username if username.starts_with('@') => {
                self.resolve_username(&username[1..]).await
            }
            id_str => {
                if let Ok(id) = id_str.parse::<i64>() {
                    Ok(tl::enums::Peer::User(tl::types::PeerUser { user_id: id }))
                } else {
                    Err(InvocationError::Deserialize(format!("cannot resolve peer: {peer}")))
                }
            }
        }
    }

    async fn resolve_username(&self, username: &str) -> Result<tl::enums::Peer, InvocationError> {
        let req  = tl::functions::contacts::ResolveUsername {
            username: username.to_string(), referer: None,
        };
        let body = self.rpc_call_raw(&req).await?;
        let mut cur = Cursor::from_slice(&body);
        let resolved = match tl::enums::contacts::ResolvedPeer::deserialize(&mut cur)? {
            tl::enums::contacts::ResolvedPeer::ResolvedPeer(r) => r,
        };
        // Cache users and chats from the resolution
        self.cache_users_slice(&resolved.users).await;
        self.cache_chats_slice(&resolved.chats).await;
        Ok(resolved.peer)
    }

    // ── Raw invoke ─────────────────────────────────────────────────────────

    /// Invoke any TL function directly, handling flood-wait retries.
    pub async fn invoke<R: RemoteCall>(&self, req: &R) -> Result<R::Return, InvocationError> {
        let body = self.rpc_call_raw(req).await?;
        let mut cur = Cursor::from_slice(&body);
        R::Return::deserialize(&mut cur).map_err(Into::into)
    }

    async fn rpc_call_raw<R: RemoteCall>(&self, req: &R) -> Result<Vec<u8>, InvocationError> {
        let mut fail_count   = NonZeroU32::new(1).unwrap();
        let mut slept_so_far = Duration::default();
        loop {
            match self.do_rpc_call(req).await {
                Ok(body) => return Ok(body),
                Err(e) => {
                    let ctx = RetryContext { fail_count, slept_so_far, error: e };
                    match self.inner.retry_policy.should_retry(&ctx) {
                        ControlFlow::Continue(delay) => {
                            sleep(delay).await;
                            slept_so_far += delay;
                            fail_count = fail_count.saturating_add(1);
                        }
                        ControlFlow::Break(()) => return Err(ctx.error),
                    }
                }
            }
        }
    }

    /// Send an RPC call and await the response via a oneshot channel.
    ///
    /// This is the core of the split-stream design:
    ///  1. Pack the request and get its msg_id.
    ///  2. Register a oneshot Sender in the pending map (BEFORE sending).
    ///  3. Send the frame while holding the writer lock.
    ///  4. Release the writer lock immediately — the reader task now runs freely.
    ///  5. Await the oneshot Receiver; the reader task will fulfill it when
    ///     the matching rpc_result frame arrives.
    async fn do_rpc_call<R: RemoteCall>(&self, req: &R) -> Result<Vec<u8>, InvocationError> {
        let (tx, rx) = oneshot::channel();
        {
            let mut w = self.inner.writer.lock().await;
            let (wire, msg_id) = w.enc.pack_with_msg_id(req);
            let fk = w.frame_kind.clone();
            // Insert BEFORE sending — response cannot arrive before we send, but
            // inserting first is the safe contract.
            self.inner.pending.lock().await.insert(msg_id, tx);
            send_frame_write(&mut w.write_half, &wire, &fk).await?;
        }
        // Writer lock is released. Reader task can now freely read from the socket
        // without competing with us.
        match tokio::time::timeout(Duration::from_secs(30), rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_))     => Err(InvocationError::Deserialize("RPC channel closed (reader died?)".into())),
            Err(_)         => Err(InvocationError::Deserialize("RPC timed out after 30 s".into())),
        }
    }

    /// Like `rpc_call_raw` but for write RPCs (Serializable, return type is Updates).
    /// Uses the same oneshot mechanism — the reader task signals success/failure.
    async fn rpc_write<S: tl::Serializable>(&self, req: &S) -> Result<(), InvocationError> {
        let mut fail_count   = NonZeroU32::new(1).unwrap();
        let mut slept_so_far = Duration::default();
        loop {
            let result = self.do_rpc_write(req).await;
            match result {
                Ok(()) => return Ok(()),
                Err(e) => {
                    let ctx = RetryContext { fail_count, slept_so_far, error: e };
                    match self.inner.retry_policy.should_retry(&ctx) {
                        ControlFlow::Continue(delay) => {
                            sleep(delay).await;
                            slept_so_far += delay;
                            fail_count = fail_count.saturating_add(1);
                        }
                        ControlFlow::Break(()) => return Err(ctx.error),
                    }
                }
            }
        }
    }

    async fn do_rpc_write<S: tl::Serializable>(&self, req: &S) -> Result<(), InvocationError> {
        let (tx, rx) = oneshot::channel();
        {
            let mut w = self.inner.writer.lock().await;
            let (wire, msg_id) = w.enc.pack_serializable_with_msg_id(req);
            let fk = w.frame_kind.clone();
            self.inner.pending.lock().await.insert(msg_id, tx);
            send_frame_write(&mut w.write_half, &wire, &fk).await?;
        }
        match tokio::time::timeout(Duration::from_secs(30), rx).await {
            Ok(Ok(result)) => result.map(|_| ()),  // ignore body for write RPCs
            Ok(Err(_))     => Err(InvocationError::Deserialize("rpc_write channel closed".into())),
            Err(_)         => Err(InvocationError::Deserialize("rpc_write timed out after 30 s".into())),
        }
    }

    // ── initConnection ─────────────────────────────────────────────────────

    async fn init_connection(&self) -> Result<(), InvocationError> {
        use tl::functions::{InvokeWithLayer, InitConnection, help::GetConfig};
        let req = InvokeWithLayer {
            layer: tl::LAYER,
            query: InitConnection {
                api_id:           self.inner.api_id,
                device_model:     "Linux".to_string(),
                system_version:   "1.0".to_string(),
                app_version:      env!("CARGO_PKG_VERSION").to_string(),
                system_lang_code: "en".to_string(),
                lang_pack:        "".to_string(),
                lang_code:        "en".to_string(),
                proxy:            None,
                params:           None,
                query:            GetConfig {},
            },
        };

        // Use the split-writer oneshot path (reader task routes the response).
        let body = self.rpc_call_raw_serializable(&req).await?;

        let mut cur = Cursor::from_slice(&body);
        if let Ok(tl::enums::Config::Config(cfg)) = tl::enums::Config::deserialize(&mut cur) {
            let allow_ipv6 = self.inner.allow_ipv6;
            let mut opts = self.inner.dc_options.lock().await;
            for opt in &cfg.dc_options {
                let tl::enums::DcOption::DcOption(o) = opt;
                if o.media_only || o.cdn || o.tcpo_only { continue; }
                if o.ipv6 && !allow_ipv6 { continue; }
                let addr = format!("{}:{}", o.ip_address, o.port);
                let entry = opts.entry(o.id).or_insert_with(|| DcEntry {
                    dc_id: o.id, addr: addr.clone(),
                    auth_key: None, first_salt: 0, time_offset: 0,
                });
                entry.addr = addr;
            }
            log::info!("[layer] initConnection ✓  ({} DCs, ipv6={})", cfg.dc_options.len(), allow_ipv6);
        }
        Ok(())
    }

    // ── DC migration ───────────────────────────────────────────────────────

    async fn migrate_to(&self, new_dc_id: i32) -> Result<(), InvocationError> {
        let addr = {
            let opts = self.inner.dc_options.lock().await;
            opts.get(&new_dc_id).map(|e| e.addr.clone())
                .unwrap_or_else(|| "149.154.167.51:443".to_string())
        };
        log::info!("[layer] Migrating to DC{new_dc_id} ({addr}) …");

        let saved_key = {
            let opts = self.inner.dc_options.lock().await;
            opts.get(&new_dc_id).and_then(|e| e.auth_key)
        };

        let socks5    = self.inner.socks5.clone();
        let transport = self.inner.transport.clone();
        let conn = if let Some(key) = saved_key {
            Connection::connect_with_key(&addr, key, 0, 0, socks5.as_ref(), &transport).await?
        } else {
            Connection::connect_raw(&addr, socks5.as_ref(), &transport).await?
        };

        let new_key = conn.auth_key_bytes();
        {
            let mut opts = self.inner.dc_options.lock().await;
            let entry = opts.entry(new_dc_id).or_insert_with(|| DcEntry {
                dc_id: new_dc_id, addr: addr.clone(),
                auth_key: None, first_salt: 0, time_offset: 0,
            });
            entry.auth_key = Some(new_key);
        }

        // Split the new connection and replace writer + reader.
        let (new_writer, new_read, new_fk) = conn.into_writer();
        let new_ak  = new_writer.enc.auth_key_bytes();
        let new_sid = new_writer.enc.session_id();
        *self.inner.writer.lock().await = new_writer;
        *self.inner.home_dc_id.lock().await = new_dc_id;

        // Hand the new read half to the reader task FIRST so it can route
        // the upcoming init_connection RPC response.
        let _ = self.inner.reconnect_tx.send((new_read, new_fk, new_ak, new_sid));

        // migrate_to() is called from user-facing methods (bot_sign_in,
        // request_login_code, sign_in) — NOT from inside the reader loop.
        // The reader task is a separate tokio task running concurrently, so
        // awaiting init_connection() here is safe: the reader is free to route
        // the RPC response while we wait. We must await before returning so
        // the caller can safely retry the original request on the new DC.
        //
        // Respect FLOOD_WAIT: if Telegram rate-limits init, wait and retry
        // rather than returning an error that would abort the whole auth flow.
        loop {
            match self.init_connection().await {
                Ok(()) => break,
                Err(InvocationError::Rpc(ref r))
                    if r.flood_wait_seconds().is_some() =>
                {
                    let secs = r.flood_wait_seconds().unwrap();
                    log::warn!(
                        "[layer] migrate_to DC{new_dc_id}: init FLOOD_WAIT_{secs} — waiting"
                    );
                    sleep(Duration::from_secs(secs + 1)).await;
                }
                Err(e) => return Err(e),
            }
        }

        self.save_session().await.ok();
        log::info!("[layer] Now on DC{new_dc_id} ✓");
        Ok(())
    }

    // ── Cache helpers ──────────────────────────────────────────────────────

    async fn cache_user(&self, user: &tl::enums::User) {
        self.inner.peer_cache.lock().await.cache_user(user);
    }

    async fn cache_users_slice(&self, users: &[tl::enums::User]) {
        let mut cache = self.inner.peer_cache.lock().await;
        cache.cache_users(users);
    }

    async fn cache_chats_slice(&self, chats: &[tl::enums::Chat]) {
        let mut cache = self.inner.peer_cache.lock().await;
        cache.cache_chats(chats);
    }

    // Public versions used by sub-modules (media.rs, participants.rs, pts.rs)
    #[doc(hidden)]
    pub async fn cache_users_slice_pub(&self, users: &[tl::enums::User]) {
        self.cache_users_slice(users).await;
    }

    #[doc(hidden)]
    pub async fn cache_chats_slice_pub(&self, chats: &[tl::enums::Chat]) {
        self.cache_chats_slice(chats).await;
    }

    /// Public RPC call for use by sub-modules.
    #[doc(hidden)]
    pub async fn rpc_call_raw_pub<R: layer_tl_types::RemoteCall>(&self, req: &R) -> Result<Vec<u8>, InvocationError> {
        self.rpc_call_raw(req).await
    }

    /// Like rpc_call_raw but takes a Serializable (for InvokeWithLayer wrappers).
    async fn rpc_call_raw_serializable<S: tl::Serializable>(&self, req: &S) -> Result<Vec<u8>, InvocationError> {
        let mut fail_count   = NonZeroU32::new(1).unwrap();
        let mut slept_so_far = Duration::default();
        loop {
            match self.do_rpc_write_returning_body(req).await {
                Ok(body) => return Ok(body),
                Err(e) => {
                    let ctx = RetryContext { fail_count, slept_so_far, error: e };
                    match self.inner.retry_policy.should_retry(&ctx) {
                        ControlFlow::Continue(delay) => {
                            sleep(delay).await;
                            slept_so_far += delay;
                            fail_count = fail_count.saturating_add(1);
                        }
                        ControlFlow::Break(()) => return Err(ctx.error),
                    }
                }
            }
        }
    }

    async fn do_rpc_write_returning_body<S: tl::Serializable>(&self, req: &S) -> Result<Vec<u8>, InvocationError> {
        let (tx, rx) = oneshot::channel();
        {
            let mut w = self.inner.writer.lock().await;
            let (wire, msg_id) = w.enc.pack_serializable_with_msg_id(req);
            let fk = w.frame_kind.clone();
            self.inner.pending.lock().await.insert(msg_id, tx);
            send_frame_write(&mut w.write_half, &wire, &fk).await?;
        }
        match tokio::time::timeout(Duration::from_secs(30), rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_))     => Err(InvocationError::Deserialize("rpc channel closed".into())),
            Err(_)         => Err(InvocationError::Deserialize("rpc timed out after 30 s".into())),
        }
    }

    // ── Paginated dialog iterator ──────────────────────────────────────────

    /// Fetch dialogs page by page.
    ///
    /// Returns a [`DialogIter`] that can be advanced with [`DialogIter::next`].
    /// This lets you page through all dialogs without loading them all at once.
    ///
    /// # Example
    /// ```rust,no_run
    /// # async fn f(client: layer_client::Client) -> Result<(), Box<dyn std::error::Error>> {
    /// let mut iter = client.iter_dialogs();
    /// while let Some(dialog) = iter.next(&client).await? {
    ///     println!("{}", dialog.title());
    /// }
    /// # Ok(()) }
    /// ```
    pub fn iter_dialogs(&self) -> DialogIter {
        DialogIter {
            offset_date: 0,
            offset_id:   0,
            offset_peer: tl::enums::InputPeer::Empty,
            done:        false,
            buffer:      VecDeque::new(),
            total:       None,
        }
    }

    /// Fetch messages from a peer, page by page.
    ///
    /// Returns a [`MessageIter`] that can be advanced with [`MessageIter::next`].
    ///
    /// # Example
    /// ```rust,no_run
    /// # async fn f(client: layer_client::Client, peer: layer_tl_types::enums::Peer) -> Result<(), Box<dyn std::error::Error>> {
    /// let mut iter = client.iter_messages(peer);
    /// while let Some(msg) = iter.next(&client).await? {
    ///     println!("{:?}", msg.text());
    /// }
    /// # Ok(()) }
    /// ```
    pub fn iter_messages(&self, peer: tl::enums::Peer) -> MessageIter {
        MessageIter {
            peer,
            offset_id: 0,
            done:      false,
            buffer:    VecDeque::new(),
            total:     None,
        }
    }

    // ── resolve_peer helper returning Result on unknown hash ───────────────

    /// Try to resolve a peer to InputPeer, returning an error if the access_hash
    /// is unknown (i.e. the peer has not been seen in any prior API call).
    pub async fn resolve_to_input_peer(
        &self,
        peer: &tl::enums::Peer,
    ) -> Result<tl::enums::InputPeer, InvocationError> {
        let cache = self.inner.peer_cache.lock().await;
        match peer {
            tl::enums::Peer::User(u) => {
                if u.user_id == 0 {
                    return Ok(tl::enums::InputPeer::PeerSelf);
                }
                match cache.users.get(&u.user_id) {
                    Some(&hash) => Ok(tl::enums::InputPeer::User(tl::types::InputPeerUser {
                        user_id: u.user_id, access_hash: hash,
                    })),
                    None => Err(InvocationError::Deserialize(format!(
                        "access_hash unknown for user {}; resolve via username first", u.user_id
                    ))),
                }
            }
            tl::enums::Peer::Chat(c) => {
                Ok(tl::enums::InputPeer::Chat(tl::types::InputPeerChat { chat_id: c.chat_id }))
            }
            tl::enums::Peer::Channel(c) => {
                match cache.channels.get(&c.channel_id) {
                    Some(&hash) => Ok(tl::enums::InputPeer::Channel(tl::types::InputPeerChannel {
                        channel_id: c.channel_id, access_hash: hash,
                    })),
                    None => Err(InvocationError::Deserialize(format!(
                        "access_hash unknown for channel {}; resolve via username first", c.channel_id
                    ))),
                }
            }
        }
    }

    // ── Multi-DC pool ──────────────────────────────────────────────────────

    /// Invoke a request on a specific DC, using the pool.
    ///
    /// If the target DC has no auth key yet, one is acquired via DH and then
    /// authorized via `auth.exportAuthorization` / `auth.importAuthorization`
    /// so the worker DC can serve user-account requests too.
    pub async fn invoke_on_dc<R: RemoteCall>(
        &self,
        dc_id: i32,
        req:   &R,
    ) -> Result<R::Return, InvocationError> {
        let body = self.rpc_on_dc_raw(dc_id, req).await?;
        let mut cur = Cursor::from_slice(&body);
        R::Return::deserialize(&mut cur).map_err(Into::into)
    }

    /// Raw RPC call routed to `dc_id`, exporting auth if needed.
    async fn rpc_on_dc_raw<R: RemoteCall>(
        &self,
        dc_id: i32,
        req:   &R,
    ) -> Result<Vec<u8>, InvocationError> {
        // Check if we need to open a new connection for this DC
        let needs_new = {
            let pool = self.inner.dc_pool.lock().await;
            !pool.has_connection(dc_id)
        };

        if needs_new {
            let addr = {
                let opts = self.inner.dc_options.lock().await;
                opts.get(&dc_id).map(|e| e.addr.clone())
                    .ok_or_else(|| InvocationError::Deserialize(format!("unknown DC{dc_id}")))?
            };

            let socks5    = self.inner.socks5.clone();
            let transport = self.inner.transport.clone();
            let saved_key = {
                let opts = self.inner.dc_options.lock().await;
                opts.get(&dc_id).and_then(|e| e.auth_key)
            };

            let dc_conn = if let Some(key) = saved_key {
                dc_pool::DcConnection::connect_with_key(&addr, key, 0, 0, socks5.as_ref(), &transport).await?
            } else {
                let conn = dc_pool::DcConnection::connect_raw(&addr, socks5.as_ref(), &transport).await?;
                // Export auth from home DC and import into worker DC
                let home_dc_id = *self.inner.home_dc_id.lock().await;
                if dc_id != home_dc_id {
                    if let Err(e) = self.export_import_auth(dc_id, &conn).await {
                        log::warn!("[layer] Auth export/import for DC{dc_id} failed: {e}");
                    }
                }
                conn
            };

            let key = dc_conn.auth_key_bytes();
            {
                let mut opts = self.inner.dc_options.lock().await;
                if let Some(e) = opts.get_mut(&dc_id) {
                    e.auth_key = Some(key);
                }
            }
            self.inner.dc_pool.lock().await.insert(dc_id, dc_conn);
        }

        let dc_entries: Vec<DcEntry> = self.inner.dc_options.lock().await.values().cloned().collect();
        self.inner.dc_pool.lock().await.invoke_on_dc(dc_id, &dc_entries, req).await
    }

    /// Export authorization from the home DC and import it into `dc_id`.
    async fn export_import_auth(
        &self,
        dc_id:   i32,
        _dc_conn: &dc_pool::DcConnection, // reserved for future direct import
    ) -> Result<(), InvocationError> {
        // Export from home DC
        let export_req = tl::functions::auth::ExportAuthorization { dc_id };
        let body    = self.rpc_call_raw(&export_req).await?;
        let mut cur = Cursor::from_slice(&body);
        let exported = match tl::enums::auth::ExportedAuthorization::deserialize(&mut cur)? {
            tl::enums::auth::ExportedAuthorization::ExportedAuthorization(e) => e,
        };

        // Import into the target DC via the pool
        let import_req = tl::functions::auth::ImportAuthorization {
            id:    exported.id,
            bytes: exported.bytes,
        };
        let dc_entries: Vec<DcEntry> = self.inner.dc_options.lock().await.values().cloned().collect();
        self.inner.dc_pool.lock().await.invoke_on_dc(dc_id, &dc_entries, &import_req).await?;
        log::info!("[layer] Auth exported+imported to DC{dc_id} ✓");
        Ok(())
    }

    // ── Private helpers ────────────────────────────────────────────────────

    async fn get_password_info(&self) -> Result<PasswordToken, InvocationError> {
        let body    = self.rpc_call_raw(&tl::functions::account::GetPassword {}).await?;
        let mut cur = Cursor::from_slice(&body);
        let pw = match tl::enums::account::Password::deserialize(&mut cur)? {
            tl::enums::account::Password::Password(p) => p,
        };
        Ok(PasswordToken { password: pw })
    }

    fn make_send_code_req(&self, phone: &str) -> tl::functions::auth::SendCode {
        tl::functions::auth::SendCode {
            phone_number: phone.to_string(),
            api_id:       self.inner.api_id,
            api_hash:     self.inner.api_hash.clone(),
            settings:     tl::enums::CodeSettings::CodeSettings(
                tl::types::CodeSettings {
                    allow_flashcall: false, current_number: false, allow_app_hash: false,
                    allow_missed_call: false, allow_firebase: false, unknown_number: false,
                    logout_tokens: None, token: None, app_sandbox: None,
                },
            ),
        }
    }

    fn extract_user_name(user: &tl::enums::User) -> String {
        match user {
            tl::enums::User::User(u) => {
                format!("{} {}",
                    u.first_name.as_deref().unwrap_or(""),
                    u.last_name.as_deref().unwrap_or(""))
                    .trim().to_string()
            }
            tl::enums::User::Empty(_) => "(unknown)".into(),
        }
    }

    fn extract_password_params(
        algo: &tl::enums::PasswordKdfAlgo,
    ) -> Result<(&[u8], &[u8], &[u8], i32), InvocationError> {
        match algo {
            tl::enums::PasswordKdfAlgo::Sha256Sha256Pbkdf2Hmacsha512iter100000Sha256ModPow(a) => {
                Ok((&a.salt1, &a.salt2, &a.p, a.g))
            }
            _ => Err(InvocationError::Deserialize("unsupported password KDF algo".into())),
        }
    }
}

// ─── Paginated iterators ──────────────────────────────────────────────────────

/// Cursor-based iterator over dialogs. Created by [`Client::iter_dialogs`].
pub struct DialogIter {
    offset_date: i32,
    offset_id:   i32,
    offset_peer: tl::enums::InputPeer,
    done:        bool,
    buffer:      VecDeque<Dialog>,
    /// Total dialog count as reported by the first server response.
    /// `None` until the first page is fetched.
    pub total: Option<i32>,
}

impl DialogIter {
    const PAGE_SIZE: i32 = 100;

    /// Total number of dialogs as reported by the server on the first page fetch.
    ///
    /// Returns `None` before the first [`next`](Self::next) call, and `None` for
    /// accounts with fewer dialogs than `PAGE_SIZE` (where the server returns
    /// `messages.Dialogs` instead of `messages.DialogsSlice`).
    pub fn total(&self) -> Option<i32> { self.total }

    /// Fetch the next dialog. Returns `None` when all dialogs have been yielded.
    pub async fn next(&mut self, client: &Client) -> Result<Option<Dialog>, InvocationError> {
        if let Some(d) = self.buffer.pop_front() { return Ok(Some(d)); }
        if self.done { return Ok(None); }

        let req = tl::functions::messages::GetDialogs {
            exclude_pinned: false,
            folder_id:      None,
            offset_date:    self.offset_date,
            offset_id:      self.offset_id,
            offset_peer:    self.offset_peer.clone(),
            limit:          Self::PAGE_SIZE,
            hash:           0,
        };

        let (dialogs, count) = client.get_dialogs_raw_with_count(req).await?;
        // Populate total from the first response (messages.DialogsSlice carries a count).
        if self.total.is_none() {
            self.total = count;
        }
        if dialogs.is_empty() || dialogs.len() < Self::PAGE_SIZE as usize {
            self.done = true;
        }

        // Prepare cursor for next page
        if let Some(last) = dialogs.last() {
            self.offset_date = last.message.as_ref().map(|m| match m {
                tl::enums::Message::Message(x) => x.date,
                tl::enums::Message::Service(x) => x.date,
                _ => 0,
            }).unwrap_or(0);
            self.offset_id = last.top_message();
            if let Some(peer) = last.peer() {
                self.offset_peer = client.inner.peer_cache.lock().await.peer_to_input(peer);
            }
        }

        self.buffer.extend(dialogs);
        Ok(self.buffer.pop_front())
    }
}

/// Cursor-based iterator over message history. Created by [`Client::iter_messages`].
pub struct MessageIter {
    peer:      tl::enums::Peer,
    offset_id: i32,
    done:      bool,
    buffer:    VecDeque<update::IncomingMessage>,
    /// Total message count from the first server response (messages.Slice).
    /// `None` until the first page is fetched, `None` for `messages.Messages`
    /// (which returns an exact slice with no separate count).
    pub total: Option<i32>,
}

impl MessageIter {
    const PAGE_SIZE: i32 = 100;

    /// Total message count from the first server response.
    ///
    /// Returns `None` before the first [`next`](Self::next) call, or for chats
    /// where the server returns an exact (non-slice) response.
    pub fn total(&self) -> Option<i32> { self.total }

    /// Fetch the next message (newest first). Returns `None` when all messages have been yielded.
    pub async fn next(&mut self, client: &Client) -> Result<Option<update::IncomingMessage>, InvocationError> {
        if let Some(m) = self.buffer.pop_front() { return Ok(Some(m)); }
        if self.done { return Ok(None); }

        let input_peer = client.inner.peer_cache.lock().await.peer_to_input(&self.peer);
        let (page, count) = client.get_messages_with_count(input_peer, Self::PAGE_SIZE, self.offset_id).await?;

        if self.total.is_none() { self.total = count; }

        if page.is_empty() || page.len() < Self::PAGE_SIZE as usize {
            self.done = true;
        }
        if let Some(last) = page.last() {
            self.offset_id = last.id();
        }

        self.buffer.extend(page);
        Ok(self.buffer.pop_front())
    }
}

// ─── Public random helper (used by media.rs) ──────────────────────────────────

/// Public wrapper for `random_i64` used by sub-modules.
#[doc(hidden)]
pub fn random_i64_pub() -> i64 { random_i64() }

// ─── Connection ───────────────────────────────────────────────────────────────

/// How framing bytes are sent/received on a connection.
#[derive(Clone)]
enum FrameKind {
    Abridged,
    Intermediate,
    #[allow(dead_code)]
    Full { send_seqno: u32, recv_seqno: u32 },
}


// ─── Split connection types ───────────────────────────────────────────────────

/// Write half of a split connection.  Held under `Mutex` in `ClientInner`.
/// Owns the EncryptedSession (for packing) and the pending-RPC map.
struct ConnectionWriter {
    write_half: OwnedWriteHalf,
    enc:        EncryptedSession,
    frame_kind: FrameKind,
}

impl ConnectionWriter {
    fn auth_key_bytes(&self) -> [u8; 256] { self.enc.auth_key_bytes() }
    fn first_salt(&self)    -> i64         { self.enc.salt }
    fn time_offset(&self)   -> i32         { self.enc.time_offset }
}

struct Connection {
    stream:     TcpStream,
    enc:        EncryptedSession,
    frame_kind: FrameKind,
}

impl Connection {
    /// Open a TCP stream, optionally via SOCKS5, and apply transport init bytes.
    async fn open_stream(
        addr:      &str,
        socks5:    Option<&crate::socks5::Socks5Config>,
        transport: &TransportKind,
    ) -> Result<(TcpStream, FrameKind), InvocationError> {
        let stream = match socks5 {
            Some(proxy) => proxy.connect(addr).await?,
            None => {
                // Let tokio do the TCP handshake properly (await until connected),
                // then apply socket2 keepalive options to the live socket.
                let stream = TcpStream::connect(addr).await
                    .map_err(InvocationError::Io)?;

                // TCP-level keepalive: OS sends probes independently of our
                // application-level pings. Catches cases where the network
                // disappears without a TCP RST (e.g. mobile data switching,
                // NAT table expiry) faster than a 15 s application ping would.
                // We use socket2 only for the setsockopt call, not for connect.
                {
                    let sock = socket2::SockRef::from(&stream);
                    let keepalive = TcpKeepalive::new()
                        .with_time(Duration::from_secs(TCP_KEEPALIVE_IDLE_SECS))
                        .with_interval(Duration::from_secs(TCP_KEEPALIVE_INTERVAL_SECS));
                    #[cfg(not(target_os = "windows"))]
                    let keepalive = keepalive.with_retries(TCP_KEEPALIVE_PROBES);
                    sock.set_tcp_keepalive(&keepalive).ok();
                }
                stream
            }
        };
        Self::apply_transport_init(stream, transport).await
    }

    /// Send the transport init bytes and return the stream + FrameKind.
    async fn apply_transport_init(
        mut stream: TcpStream,
        transport:  &TransportKind,
    ) -> Result<(TcpStream, FrameKind), InvocationError> {
        match transport {
            TransportKind::Abridged => {
                stream.write_all(&[0xef]).await?;
                Ok((stream, FrameKind::Abridged))
            }
            TransportKind::Intermediate => {
                stream.write_all(&[0xee, 0xee, 0xee, 0xee]).await?;
                Ok((stream, FrameKind::Intermediate))
            }
            TransportKind::Full => {
                // Full transport has no init byte
                Ok((stream, FrameKind::Full { send_seqno: 0, recv_seqno: 0 }))
            }
            TransportKind::Obfuscated { secret } => {
                // For obfuscated we do the full handshake inside open_obfuscated,
                // then wrap back in a plain TcpStream via into_inner.
                // Since ObfuscatedStream is a different type we reuse the Abridged
                // frame logic internally — the encryption layer handles everything.
                //
                // Implementation note: We convert to Abridged after the handshake
                // because ObfuscatedStream internally already uses Abridged framing
                // with XOR applied on top.  The outer Connection just sends raw bytes.
                let mut nonce = [0u8; 64];
                getrandom::getrandom(&mut nonce).map_err(|_| InvocationError::Deserialize("getrandom".into()))?;
                // Write obfuscated handshake header
                let (enc_key, enc_iv, _dec_key, _dec_iv) = crate::transport_obfuscated::derive_keys(&nonce, secret.as_ref());
                let mut enc_cipher = crate::transport_obfuscated::ObfCipher::new(enc_key, enc_iv);
                // Stamp protocol tag into nonce[56..60]
                let mut handshake = nonce;
                handshake[56] = 0xef; handshake[57] = 0xef;
                handshake[58] = 0xef; handshake[59] = 0xef;
                enc_cipher.apply(&mut handshake[56..]);
                stream.write_all(&handshake).await?;
                Ok((stream, FrameKind::Abridged))
            }
        }
    }

    async fn connect_raw(
        addr:      &str,
        socks5:    Option<&crate::socks5::Socks5Config>,
        transport: &TransportKind,
    ) -> Result<Self, InvocationError> {
        log::info!("[layer] Connecting to {addr} (DH) …");

        // Wrap the entire DH handshake in a timeout so a silent server
        // response (e.g. a mis-framed transport error) never causes an
        // infinite hang.
        let addr2      = addr.to_string();
        let socks5_c   = socks5.cloned();
        let transport_c = transport.clone();

        let fut = async move {
            let (mut stream, frame_kind) =
                Self::open_stream(&addr2, socks5_c.as_ref(), &transport_c).await?;

            let mut plain = Session::new();

            let (req1, s1) = auth::step1().map_err(|e| InvocationError::Deserialize(e.to_string()))?;
            send_frame(&mut stream, &plain.pack(&req1).to_plaintext_bytes(), &frame_kind).await?;
            let res_pq: tl::enums::ResPq = recv_frame_plain(&mut stream, &frame_kind).await?;

            let (req2, s2) = auth::step2(s1, res_pq).map_err(|e| InvocationError::Deserialize(e.to_string()))?;
            send_frame(&mut stream, &plain.pack(&req2).to_plaintext_bytes(), &frame_kind).await?;
            let dh: tl::enums::ServerDhParams = recv_frame_plain(&mut stream, &frame_kind).await?;

            let (req3, s3) = auth::step3(s2, dh).map_err(|e| InvocationError::Deserialize(e.to_string()))?;
            send_frame(&mut stream, &plain.pack(&req3).to_plaintext_bytes(), &frame_kind).await?;
            let ans: tl::enums::SetClientDhParamsAnswer = recv_frame_plain(&mut stream, &frame_kind).await?;

            let done = auth::finish(s3, ans).map_err(|e| InvocationError::Deserialize(e.to_string()))?;
            log::info!("[layer] DH complete ✓");

            Ok::<Self, InvocationError>(Self {
                stream,
                enc: EncryptedSession::new(done.auth_key, done.first_salt, done.time_offset),
                frame_kind,
            })
        };

        tokio::time::timeout(Duration::from_secs(15), fut)
            .await
            .map_err(|_| InvocationError::Deserialize(
                format!("DH handshake with {addr} timed out after 15 s")
            ))?
    }

    async fn connect_with_key(
        addr:        &str,
        auth_key:    [u8; 256],
        first_salt:  i64,
        time_offset: i32,
        socks5:      Option<&crate::socks5::Socks5Config>,
        transport:   &TransportKind,
    ) -> Result<Self, InvocationError> {
        let addr2       = addr.to_string();
        let socks5_c    = socks5.cloned();
        let transport_c = transport.clone();

        let fut = async move {
            let (stream, frame_kind) =
                Self::open_stream(&addr2, socks5_c.as_ref(), &transport_c).await?;
            Ok::<Self, InvocationError>(Self {
                stream,
                enc: EncryptedSession::new(auth_key, first_salt, time_offset),
                frame_kind,
            })
        };

        tokio::time::timeout(Duration::from_secs(15), fut)
            .await
            .map_err(|_| InvocationError::Deserialize(
                format!("connect_with_key to {addr} timed out after 15 s")
            ))?
    }

    fn auth_key_bytes(&self) -> [u8; 256] { self.enc.auth_key_bytes() }

    /// Split into a write-only `ConnectionWriter` and the TCP read half.
    fn into_writer(self) -> (ConnectionWriter, OwnedReadHalf, FrameKind) {
        let (read_half, write_half) = self.stream.into_split();
        let writer = ConnectionWriter {
            write_half,
            enc:        self.enc,
            frame_kind: self.frame_kind.clone(),
        };
        (writer, read_half, self.frame_kind)
    }
}

// ─── Transport framing (multi-kind) ──────────────────────────────────────────

/// Send a framed message using the active transport kind.
async fn send_frame(
    stream: &mut TcpStream,
    data:   &[u8],
    kind:   &FrameKind,
) -> Result<(), InvocationError> {
    match kind {
        FrameKind::Abridged => send_abridged(stream, data).await,
        FrameKind::Intermediate => {
            stream.write_all(&(data.len() as u32).to_le_bytes()).await?;
            stream.write_all(data).await?;
            Ok(())
        }
        FrameKind::Full { .. } => {
            // seqno and CRC handled inside Connection; here we just prefix length
            // Full framing: [total_len 4B][seqno 4B][payload][crc32 4B]
            // But send_frame is called with already-encrypted payload.
            // We use a simplified approach: emit the same as Intermediate for now
            // and note that Full's seqno/CRC are transport-level, not app-level.
            stream.write_all(&(data.len() as u32).to_le_bytes()).await?;
            stream.write_all(data).await?;
            Ok(())
        }
    }
}

// ─── Split-reader helpers ─────────────────────────────────────────────────────

/// Outcome of a timed frame read attempt.
enum FrameOutcome {
    Frame(Vec<u8>),
    Error(InvocationError),
    Keepalive,  // timeout elapsed but ping was sent; caller should loop
}

/// Read one frame with a 60-second keepalive timeout (PING_DELAY_SECS).
///
/// If the timeout fires we send a `PingDelayDisconnect` — this tells Telegram
/// to forcibly close the connection after `NO_PING_DISCONNECT` seconds of
/// silence, giving us a clean EOF to detect rather than a silently stale socket.
/// That mirrors what both grammers and the official Telegram clients do.
async fn recv_frame_with_keepalive(
    rh:      &mut OwnedReadHalf,
    fk:      &FrameKind,
    client:  &Client,
    _ak:     &[u8; 256],
) -> FrameOutcome {
    match tokio::time::timeout(Duration::from_secs(PING_DELAY_SECS), recv_frame_read(rh, fk)).await {
        Ok(Ok(raw)) => FrameOutcome::Frame(raw),
        Ok(Err(e))  => FrameOutcome::Error(e),
        Err(_) => {
            // Keepalive timeout: send PingDelayDisconnect so Telegram closes the
            // connection cleanly (EOF) if it hears nothing for NO_PING_DISCONNECT
            // seconds, rather than leaving a silently stale socket.
            let ping_req = tl::functions::PingDelayDisconnect {
                ping_id:          random_i64(),
                disconnect_delay: NO_PING_DISCONNECT,
            };
            let mut w = client.inner.writer.lock().await;
            let wire = w.enc.pack(&ping_req);
            let fk = w.frame_kind.clone();
            match send_frame_write(&mut w.write_half, &wire, &fk).await {
                Ok(()) => FrameOutcome::Keepalive,
                // If the write itself fails the connection is already dead.
                // Return Error so the reader immediately enters the reconnect loop
                // instead of sitting silent for another PING_DELAY_SECS.
                Err(e) => FrameOutcome::Error(e),
            }
        }
    }
}

/// Send a framed message via an OwnedWriteHalf (split connection).
async fn send_frame_write(
    stream: &mut OwnedWriteHalf,
    data:   &[u8],
    kind:   &FrameKind,
) -> Result<(), InvocationError> {
    match kind {
        FrameKind::Abridged => {
            let words = data.len() / 4;
            if words < 0x7f {
                stream.write_all(&[words as u8]).await?;
            } else {
                let b = [0x7f, (words & 0xff) as u8, ((words >> 8) & 0xff) as u8, ((words >> 16) & 0xff) as u8];
                stream.write_all(&b).await?;
            }
            stream.write_all(data).await?;
            Ok(())
        }
        FrameKind::Intermediate | FrameKind::Full { .. } => {
            stream.write_all(&(data.len() as u32).to_le_bytes()).await?;
            stream.write_all(data).await?;
            Ok(())
        }
    }
}

/// Receive a framed message via an OwnedReadHalf (split connection).
async fn recv_frame_read(
    stream: &mut OwnedReadHalf,
    kind:   &FrameKind,
) -> Result<Vec<u8>, InvocationError> {
    match kind {
        FrameKind::Abridged => {
            let mut h = [0u8; 1];
            stream.read_exact(&mut h).await?;
            let words = if h[0] < 0x7f {
                h[0] as usize
            } else {
                let mut b = [0u8; 3];
                stream.read_exact(&mut b).await?;
                b[0] as usize | (b[1] as usize) << 8 | (b[2] as usize) << 16
            };
            let len = words * 4;
            let mut buf = vec![0u8; len];
            stream.read_exact(&mut buf).await?;
            Ok(buf)
        }
        FrameKind::Intermediate | FrameKind::Full { .. } => {
            let mut len_buf = [0u8; 4];
            stream.read_exact(&mut len_buf).await?;
            let len = u32::from_le_bytes(len_buf) as usize;
            let mut buf = vec![0u8; len];
            stream.read_exact(&mut buf).await?;
            Ok(buf)
        }
    }
}


/// Send using Abridged framing (used for DH plaintext during connect).
async fn send_abridged(stream: &mut TcpStream, data: &[u8]) -> Result<(), InvocationError> {
    let words = data.len() / 4;
    if words < 0x7f {
        stream.write_all(&[words as u8]).await?;
    } else {
        let b = [0x7f, (words & 0xff) as u8, ((words >> 8) & 0xff) as u8, ((words >> 16) & 0xff) as u8];
        stream.write_all(&b).await?;
    }
    stream.write_all(data).await?;
    Ok(())
}

async fn recv_abridged(stream: &mut TcpStream) -> Result<Vec<u8>, InvocationError> {
    let mut h = [0u8; 1];
    stream.read_exact(&mut h).await?;
    let words = if h[0] < 0x7f {
        h[0] as usize
    } else {
        let mut b = [0u8; 3];
        stream.read_exact(&mut b).await?;
        let w = b[0] as usize | (b[1] as usize) << 8 | (b[2] as usize) << 16;
        // word count of 1 after 0xFF = Telegram 4-byte transport error code
        if w == 1 {
            let mut code_buf = [0u8; 4];
            stream.read_exact(&mut code_buf).await?;
            let code = i32::from_le_bytes(code_buf);
            return Err(InvocationError::Rpc(RpcError::from_telegram(code, "transport error")));
        }
        w
    };
    // Guard against implausibly large reads — a raw 4-byte transport error
    // whose first byte was mis-read as a word count causes a hang otherwise.
    if words == 0 || words > 0x8000 {
        return Err(InvocationError::Deserialize(
            format!("abridged: implausible word count {words} (possible transport error or framing mismatch)")
        ));
    }
    let mut buf = vec![0u8; words * 4];
    stream.read_exact(&mut buf).await?;
    Ok(buf)
}

/// Receive a plaintext (pre-auth) frame and deserialize it.
async fn recv_frame_plain<T: Deserializable>(
    stream: &mut TcpStream,
    _kind:  &FrameKind,
) -> Result<T, InvocationError> {
    let raw = recv_abridged(stream).await?; // DH always uses abridged for plaintext
    if raw.len() < 20 {
        return Err(InvocationError::Deserialize("plaintext frame too short".into()));
    }
    if u64::from_le_bytes(raw[..8].try_into().unwrap()) != 0 {
        return Err(InvocationError::Deserialize("expected auth_key_id=0 in plaintext".into()));
    }
    let body_len = u32::from_le_bytes(raw[16..20].try_into().unwrap()) as usize;
    let mut cur  = Cursor::from_slice(&raw[20..20 + body_len]);
    T::deserialize(&mut cur).map_err(Into::into)
}

// ─── MTProto envelope ─────────────────────────────────────────────────────────

enum EnvelopeResult {
    Payload(Vec<u8>),
    Updates(Vec<update::Update>),
    None,
}

fn unwrap_envelope(body: Vec<u8>) -> Result<EnvelopeResult, InvocationError> {
    if body.len() < 4 {
        return Err(InvocationError::Deserialize("body < 4 bytes".into()));
    }
    let cid = u32::from_le_bytes(body[..4].try_into().unwrap());

    match cid {
        ID_RPC_RESULT => {
            if body.len() < 12 {
                return Err(InvocationError::Deserialize("rpc_result too short".into()));
            }
            unwrap_envelope(body[12..].to_vec())
        }
        ID_RPC_ERROR => {
            if body.len() < 8 {
                return Err(InvocationError::Deserialize("rpc_error too short".into()));
            }
            let code    = i32::from_le_bytes(body[4..8].try_into().unwrap());
            let message = tl_read_string(&body[8..]).unwrap_or_default();
            Err(InvocationError::Rpc(RpcError::from_telegram(code, &message)))
        }
        ID_MSG_CONTAINER => {
            if body.len() < 8 {
                return Err(InvocationError::Deserialize("container too short".into()));
            }
            let count = u32::from_le_bytes(body[4..8].try_into().unwrap()) as usize;
            let mut pos = 8usize;
            let mut payload: Option<Vec<u8>> = None;
            let mut updates_buf: Vec<update::Update> = Vec::new();

            for _ in 0..count {
                if pos + 16 > body.len() { break; }
                let inner_len = u32::from_le_bytes(body[pos + 12..pos + 16].try_into().unwrap()) as usize;
                pos += 16;
                if pos + inner_len > body.len() { break; }
                let inner = body[pos..pos + inner_len].to_vec();
                pos += inner_len;
                match unwrap_envelope(inner)? {
                    EnvelopeResult::Payload(p)  => { payload = Some(p); }
                    EnvelopeResult::Updates(us) => { updates_buf.extend(us); }
                    EnvelopeResult::None        => {}
                }
            }
            if let Some(p) = payload {
                Ok(EnvelopeResult::Payload(p))
            } else if !updates_buf.is_empty() {
                Ok(EnvelopeResult::Updates(updates_buf))
            } else {
                Ok(EnvelopeResult::None)
            }
        }
        ID_GZIP_PACKED => {
            let bytes = tl_read_bytes(&body[4..]).unwrap_or_default();
            unwrap_envelope(gz_inflate(&bytes)?)
        }
        // MTProto service messages — silently acknowledged, no payload extracted.
        // NOTE: ID_PONG is intentionally NOT listed here. Pong arrives as a bare
        // top-level frame (never inside rpc_result), so it is handled in route_frame
        // directly. Silencing it here would drop it before invoke() can resolve it.
        ID_MSGS_ACK | ID_NEW_SESSION | ID_BAD_SERVER_SALT | ID_BAD_MSG_NOTIFY
        // These are correctly silenced (grammers silences these too)
        | 0xd33b5459  // MsgsStateReq
        | 0x04deb57d  // MsgsStateInfo
        | 0x8cc0d131  // MsgsAllInfo
        | 0x276d3ec6  // MsgDetailedInfo
        | 0x809db6df  // MsgNewDetailedInfo
        | 0x7d861a08  // MsgResendReq / MsgResendAnsReq
        | 0x0949d9dc  // FutureSalt
        | 0xae500895  // FutureSalts
        | 0x9299359f  // HttpWait
        | 0xe22045fc  // DestroySessionOk
        | 0x62d350c9  // DestroySessionNone
        => {
            Ok(EnvelopeResult::None)
        }
        ID_UPDATES | ID_UPDATE_SHORT | ID_UPDATES_COMBINED
        | ID_UPDATE_SHORT_MSG | ID_UPDATE_SHORT_CHAT_MSG
        | ID_UPDATES_TOO_LONG => {
            Ok(EnvelopeResult::Updates(update::parse_updates(&body)))
        }
        _ => Ok(EnvelopeResult::Payload(body)),
    }
}

// ─── Utilities ────────────────────────────────────────────────────────────────

fn random_i64() -> i64 {
    let mut b = [0u8; 8];
    getrandom::getrandom(&mut b).expect("getrandom");
    i64::from_le_bytes(b)
}

/// Apply ±20 % random jitter to a backoff delay.
/// Prevents thundering-herd when many clients reconnect simultaneously
/// (e.g. after a server restart or a shared network outage).
fn jitter_delay(base_ms: u64) -> Duration {
    // Use two random bytes for the jitter factor (0..=65535 → 0.80 … 1.20).
    let mut b = [0u8; 2];
    getrandom::getrandom(&mut b).unwrap_or(());
    let rand_frac = u16::from_le_bytes(b) as f64 / 65535.0; // 0.0 … 1.0
    let factor    = 0.80 + rand_frac * 0.40;                 // 0.80 … 1.20
    Duration::from_millis((base_ms as f64 * factor) as u64)
}

fn tl_read_bytes(data: &[u8]) -> Option<Vec<u8>> {
    if data.is_empty() { return Some(vec![]); }
    let (len, start) = if data[0] < 254 { (data[0] as usize, 1) }
    else if data.len() >= 4 {
        (data[1] as usize | (data[2] as usize) << 8 | (data[3] as usize) << 16, 4)
    } else { return None; };
    if data.len() < start + len { return None; }
    Some(data[start..start + len].to_vec())
}

fn tl_read_string(data: &[u8]) -> Option<String> {
    tl_read_bytes(data).map(|b| String::from_utf8_lossy(&b).into_owned())
}

fn gz_inflate(data: &[u8]) -> Result<Vec<u8>, InvocationError> {
    use std::io::Read;
    let mut out = Vec::new();
    if flate2::read::GzDecoder::new(data).read_to_end(&mut out).is_ok() && !out.is_empty() {
        return Ok(out);
    }
    out.clear();
    flate2::read::ZlibDecoder::new(data)
        .read_to_end(&mut out)
        .map_err(|_| InvocationError::Deserialize("decompression failed".into()))?;
    Ok(out)
}