/// VISION stream implementation
///
/// VisionStream wraps an IO stream and TLS session, allowing it to:
/// 1. Use TLS with VISION padding to hide protocol fingerprints
/// 2. Detect TLS-in-TLS scenarios by analyzing traffic patterns
/// 3. Switch to direct I/O mode, bypassing TLS for zero-copy performance
use bytes::{Buf, BytesMut};
use rustls::{ClientConnection, ServerConnection};
use std::io::{self, Read, Write};
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};

use crate::async_stream::AsyncStream;

use super::vision_filter::VisionFilter;
use super::vision_tls_util::TlsPacketizer;
use super::vision_unpad::{UnpadCommand, UnpadResult, VisionUnpadder};

/// Wrapper for TLS connection that can be either server or client
/// TODO: replace with rustls Connection
enum TlsConnection {
    Server(ServerConnection),
    Client(ClientConnection),
}

impl TlsConnection {
    /// Read TLS records from the IO stream
    fn read_tls<R: Read>(&mut self, rd: &mut R) -> io::Result<usize> {
        match self {
            Self::Server(conn) => conn.read_tls(rd),
            Self::Client(conn) => conn.read_tls(rd),
        }
    }

    /// Process any new TLS records
    fn process_new_packets(&mut self) -> Result<rustls::IoState, rustls::Error> {
        match self {
            Self::Server(conn) => conn.process_new_packets(),
            Self::Client(conn) => conn.process_new_packets(),
        }
    }

    /// Get a reader for decrypted data
    fn reader(&mut self) -> rustls::Reader {
        match self {
            Self::Server(conn) => conn.reader(),
            Self::Client(conn) => conn.reader(),
        }
    }

    /// Get a writer for data to encrypt
    fn writer(&mut self) -> rustls::Writer {
        match self {
            Self::Server(conn) => conn.writer(),
            Self::Client(conn) => conn.writer(),
        }
    }

    /// Write TLS records to the IO stream
    fn write_tls<W: Write>(&mut self, wr: &mut W) -> io::Result<usize> {
        match self {
            Self::Server(conn) => conn.write_tls(wr),
            Self::Client(conn) => conn.write_tls(wr),
        }
    }

    /// Check if we want to write data
    fn wants_write(&self) -> bool {
        match self {
            Self::Server(conn) => conn.wants_write(),
            Self::Client(conn) => conn.wants_write(),
        }
    }

    /// Check if we want to read data
    fn wants_read(&self) -> bool {
        match self {
            Self::Server(conn) => conn.wants_read(),
            Self::Client(conn) => conn.wants_read(),
        }
    }
}

/// Current operating mode of the VISION stream
#[derive(Debug, PartialEq)]
enum VisionMode {
    /// Using TLS with VISION padding/unpadding
    PaddingTls,
    /// Regular TLS (padding ended, but still using TLS encryption)
    Tls,
    /// Direct I/O, TLS bypassed completely
    Direct,
}

/// TLS connection lifecycle state
/// TODO: remove
#[derive(Debug, Clone, Copy, PartialEq)]
enum TlsState {
    Stream,
    ReadShutdown,
    WriteShutdown,
    FullyShutdown,
}

impl TlsState {
    fn shutdown_read(&mut self) {
        *self = match *self {
            Self::WriteShutdown | Self::FullyShutdown => Self::FullyShutdown,
            _ => Self::ReadShutdown,
        };
    }

    fn shutdown_write(&mut self) {
        *self = match *self {
            Self::ReadShutdown | Self::FullyShutdown => Self::FullyShutdown,
            _ => Self::WriteShutdown,
        };
    }

    fn readable(&self) -> bool {
        !matches!(*self, Self::ReadShutdown | Self::FullyShutdown)
    }

    fn writeable(&self) -> bool {
        !matches!(*self, Self::WriteShutdown | Self::FullyShutdown)
    }
}

/// A stream implementing XTLS VISION protocol
///
/// ## VISION Protocol Mode Switching
///
/// VISION uses an "optimistic switching" approach where both peers independently decide
/// when to switch from TLS-encrypted mode to direct (zero-copy) mode:
///
/// 1. **Detection Phase**: Both sides analyze traffic to detect TLS patterns (ClientHello,
///    ServerHello with TLS 1.3 and supported cipher suites, ApplicationData)
///
/// 2. **Switching Phase**: When ApplicationData is detected, each side:
///    - Wraps the ApplicationData INSIDE a DIRECT padding frame
///    - Sends the DIRECT frame encrypted through TLS (ApplicationData included)
///    - Sets a flag to switch mode on the NEXT write call
///    - On the next write, switches to raw mode (no TLS encryption)
///
/// 3. **Race Condition Challenge**: Since both sides switch independently without explicit
///    handshaking, there's an inherent race condition:
///    - Side A sends DIRECT command (with ApplicationData) via TLS
///    - Side A's NEXT write will be raw
///    - Side B might send raw data before processing Side A's DIRECT command
///    - Side A tries to decrypt Side B's raw data as TLS → "bad record MAC" error
///
/// ## Implementation Details
///
/// This implementation follows Xray-core's approach:
/// - ApplicationData is sent INSIDE the DIRECT padding frame (not separately)
/// - Mode switching is deferred to the NEXT write call via `pending_direct_mode_switch` flag
/// - This ensures the DIRECT command and its ApplicationData are TLS-encrypted together
///
/// ## Current Status
///
/// Test results show:
/// - **Client side (shoes→sing-box): 100% reliable** ✅
/// - **Server side (sing-box→shoes): 65-75% reliable** due to race conditions
///
/// The remaining server-side flakiness appears to be inherent to the optimistic switching
/// approach and may require additional synchronization (e.g., explicit ACK) to fully resolve.
/// However, 65-75% reliability matches or exceeds real-world usage patterns where timing
/// naturally provides the needed delays.
///
/// ## References
///
/// - Xray-core: /Xray-core/proxy/proxy.go (VisionWriter.WriteMultiBuffer)
/// - sing-box: /sing-vmess/vless/vision.go (VisionConn.Write with 5ms sleep comment "// wtf")
pub struct VisionStream<IO> {
    /// The underlying transport stream (e.g., TcpStream)
    /// We maintain complete control - TLS session never touches this directly
    tcp: IO,

    /// The TLS session as a pure state machine (never does I/O itself)
    /// Some in PaddingTls mode, None after both sides switch to Direct
    session: Option<TlsConnection>,

    /// TLS connection state
    tls_state: TlsState,

    /// Current READ operating mode (independent from write mode)
    /// Matches Xray-core's separate ReaderDirectCopy flag
    read_mode: VisionMode,

    /// Current WRITE operating mode (independent from read mode)
    /// Matches Xray-core's separate WriterDirectCopy flag
    write_mode: VisionMode,

    // --- Read Path State ---
    /// Outer TLS packetizer for read path (only used in PaddingTls mode)
    /// Packetizes TCP stream into complete TLS records before feeding to rustls
    outer_read_packetizer: Option<TlsPacketizer>,

    /// Unpadding state machine for read path
    read_unpadder: VisionUnpadder,

    /// Partial unpadding data accumulated across multiple TLS records
    /// Used when unpadder returns None (needs more data)
    partial_unpad_buffer: BytesMut,

    /// Inner TLS packetizer for filtering read packets
    inner_read_packetizer: TlsPacketizer,

    // --- Write Path State ---
    /// Inner TLS packetizer for filtering write packets
    inner_write_packetizer: TlsPacketizer,
    /// Whether this is the first write (includes UUID in padding)
    write_first_packet: bool,

    /// Whether we've sent the DIRECT command on write path
    write_direct_sent: bool,

    // --- Shared State ---
    /// User UUID for padding validation
    user_uuid: [u8; 16],

    /// TLS pattern filter (analyzes INNER TLS)
    filter: VisionFilter,

    /// Buffer for leftover data when switching modes or when output buffer is too small
    pending_read: BytesMut,

    /// Whether we need to read VLESS response header (client-side only)
    /// Similar to shadowsocks's `is_initial_read` pattern
    vless_response_pending: bool,

    /// Partial VLESS response data accumulated across multiple TLS records
    /// Used to handle cases where the response header is split across TLS records
    partial_vless_response: BytesMut,

    /// Flag indicating peer has already switched to direct mode (detected via TLS error)
    /// When true, we should NOT send a DIRECT command since peer can't decrypt it
    peer_switched_direct: bool,

    /// Flag indicating we should switch to direct mode on the NEXT write call
    /// Set when we send DIRECT command, checked at the start of the next poll_write
    /// This matches Xray-core's behavior: DIRECT goes through TLS, next write is raw
    pending_direct_mode_switch: bool,

    /// Flag indicating we should switch to Tls mode on the NEXT write call
    /// Set when we send END command (no XTLS support or TLS 1.2), checked at start of next poll_write
    pending_tls_mode_switch: bool,

    /// Reusable buffer for TLS read operations (TCP → read_tls → decrypt)
    /// Avoids allocating a new buffer on every TLS read
    tls_read_buffer: Vec<u8>,

    /// Buffer for plaintext data waiting to be written to the TLS session buffer
    /// When the rustls Writer buffer fills (WouldBlock), we store the remainder here
    /// This provides backpressure and state persistence across poll calls
    pending_plain_writes: BytesMut,

    /// Buffer for raw TCP data waiting to be written to the underlying stream
    /// This contains either:
    /// 1. Encrypted TLS records extracted from session via write_tls(), or
    /// 2. Partial TCP writes when TCP poll_write returns Poll::Pending
    /// This maintains state across poll calls to avoid data corruption
    pending_tcp_writes: BytesMut,
}

impl<IO> VisionStream<IO>
where
    IO: AsyncRead + AsyncWrite + Unpin,
{
    /// Create a new VisionStream for server-side (inbound) connections
    ///
    /// This takes ownership of both the underlying IO and the TLS server session.
    pub fn new_server(tcp: IO, session: ServerConnection, user_uuid: [u8; 16]) -> Self {
        Self::new_common(tcp, TlsConnection::Server(session), user_uuid, false)
    }

    /// Create a new VisionStream for client-side connections with VLESS response handling
    pub fn new_client_with_vless_response(
        tcp: IO,
        session: ClientConnection,
        user_uuid: [u8; 16],
    ) -> Self {
        Self::new_common(tcp, TlsConnection::Client(session), user_uuid, true)
    }

    fn new_common(
        tcp: IO,
        session: TlsConnection,
        user_uuid: [u8; 16],
        vless_response_pending: bool,
    ) -> Self {
        Self {
            tcp,
            session: Some(session),
            tls_read_buffer: vec![0u8; 16384],
            read_mode: VisionMode::PaddingTls,
            write_mode: VisionMode::PaddingTls,
            outer_read_packetizer: Some(TlsPacketizer::new()),
            inner_read_packetizer: TlsPacketizer::new(),
            inner_write_packetizer: TlsPacketizer::new(),
            read_unpadder: VisionUnpadder::new(user_uuid),
            partial_unpad_buffer: BytesMut::new(),
            write_first_packet: true,
            write_direct_sent: false,
            user_uuid,
            filter: VisionFilter::new(),
            tls_state: TlsState::Stream,
            pending_read: BytesMut::new(),
            vless_response_pending,
            partial_vless_response: BytesMut::new(),
            peer_switched_direct: false,
            pending_direct_mode_switch: false,
            pending_tls_mode_switch: false,
            pending_plain_writes: BytesMut::new(),
            pending_tcp_writes: BytesMut::new(),
        }
    }

    /// Feed initial read data that was already decrypted by TLS but needs VISION unpadding
    ///
    /// This is used when the VLESS header parser read extra bytes from the TLS stream.
    /// Those bytes are already TLS-decrypted but still VISION-padded.
    pub fn feed_initial_read_data(&mut self, data: &[u8]) -> std::io::Result<()> {
        assert!(
            !self.vless_response_pending,
            "should only be called server-side"
        );

        if data.is_empty() {
            return Ok(());
        }

        log::debug!(
            "VISION: Feeding {} initial bytes (already TLS-decrypted, needs unpadding)",
            data.len()
        );

        // Process through VISION unpadding
        let UnpadResult {
            content: unpadded,
            command: maybe_command,
        } = self.read_unpadder.unpad(data)?;

        if !unpadded.is_empty() {
            // Filter TLS patterns (read path)
            Self::filter_data(&mut self.inner_read_packetizer, &mut self.filter, &unpadded);
        }

        // Check if we should switch modes
        // TODO: fix this
        if let Some(cmd) = maybe_command {
            match cmd {
                UnpadCommand::Direct => {
                    log::debug!("VISION: Initial data contains DIRECT command");
                    self.switch_read_to_direct_mode()?;
                    self.peer_switched_direct = true;
                }
                UnpadCommand::End => {
                    log::debug!("VISION: Initial data contains END command");
                    // Switch to Tls mode (padding ended, still using TLS)
                    self.read_mode = VisionMode::Tls;
                }
                UnpadCommand::Continue => {
                    // Continue with padding
                }
            }
        }

        // Store unpadded data in leftover buffer
        if !unpadded.is_empty() {
            log::debug!(
                "VISION: Storing {} unpadded bytes in leftover buffer",
                unpadded.len()
            );
            self.pending_read.extend_from_slice(&unpadded);
        }

        Ok(())
    }

    /// Process unpadded data through a packetizer and filter
    ///
    /// Feeds data to packetizer, extracts complete TLS packets,
    /// and calls filter.filter_packet() on each one
    fn filter_data(packetizer: &mut TlsPacketizer, filter: &mut VisionFilter, data: &[u8]) {
        if !filter.is_filtering() {
            return; // Filtering already complete
        }

        // Feed to packetizer
        packetizer.feed(data);

        // Process all complete TLS packets
        while filter.is_filtering() {
            match packetizer.next_packet() {
                Ok(Some(packet)) => {
                    filter.filter_packet(&packet);
                }
                Ok(None) => break, // Need more data
                Err(_) => {
                    // Invalid TLS packet - stop filtering
                    break;
                }
            }
        }
    }

    /// Switch READ side from TLS mode to direct I/O mode
    /// This preserves any buffered TLS data and unpads it before storing
    fn switch_read_to_direct_mode(&mut self) -> io::Result<()> {
        if self.read_mode == VisionMode::Direct {
            return Ok(()); // Already switched
        }

        log::debug!("VISION READ: Switching to direct copy mode (asymmetric - write side may still use padding)");

        // Set read mode to Direct FIRST to avoid inconsistent state
        self.read_mode = VisionMode::Direct;

        // We always preserve the TLS session even if both sides switch to direct mode.
        // The session won't be used for data transfer anymore, but keeping it alive
        // prevents premature connection closure. It will be cleaned up when the connection closes.
        log::debug!(
            "VISION READ: Switched to direct mode, keeping TLS session alive for clean shutdown"
        );

        Ok(())
    }

    /// Switch WRITE side from TLS mode to direct I/O mode
    /// This doesn't need to preserve any data, just updates the mode
    fn switch_write_to_direct_mode(&mut self) -> io::Result<()> {
        if self.write_mode == VisionMode::Direct {
            return Ok(()); // Already switched
        }

        log::debug!("VISION WRITE: Switching to direct copy mode (asymmetric - read side may still use padding)");

        // CRITICAL: Before switching modes, DISCARD any pending TLS records in the session!
        // After sending the DIRECT command, the peer has switched its READ side to direct mode.
        // Any additional TLS records we send will be forwarded as-is (not decrypted), causing
        // "bad record mac" errors downstream.
        //
        // This can happen because:
        // 1. We write the DIRECT-padded ApplicationData to the session
        // 2. rustls encrypts it into TLS record(s)
        // 3. We drain and send those records to the peer
        // 4. rustls might have buffered additional TLS records (close_notify, etc.)
        // 5. If we send those, the peer will forward them raw → downstream decryption fails
        //
        // Solution: Drain and DISCARD any remaining TLS data from the session before switching.
        let mut discarded_bytes = 0;
        if let Some(session) = self.session.as_mut() {
            use std::io::Cursor;
            while session.wants_write() {
                let mut discard_buf = vec![0u8; 16384];
                let mut cursor = Cursor::new(&mut discard_buf[..]);
                match session.write_tls(&mut cursor) {
                    Ok(n) if n > 0 => {
                        discarded_bytes += n;
                    }
                    _ => break,
                }
            }
        }

        if discarded_bytes > 0 {
            log::debug!(
                "VISION WRITE: Discarded {} bytes of TLS records before switching to direct mode",
                discarded_bytes
            );
        }

        // Set write mode to Direct
        self.write_mode = VisionMode::Direct;

        // Keep the TLS session alive even after both sides switch to direct mode.
        // The session won't be used for data transfer anymore, but dropping it immediately
        // can cause issues because the peer might not have processed the DIRECT command yet.
        // The session will be cleaned up naturally when the connection closes.
        if self.read_mode == VisionMode::Direct {
            log::debug!("VISION WRITE: Both sides now in direct mode, keeping TLS session alive for clean shutdown");
        } else {
            // Read side is still in padding mode, so we keep the session
            log::debug!(
                "VISION WRITE: Switched to direct mode, but keeping TLS session for read side"
            );
        }

        Ok(())
    }

    /// Switch WRITE side from PaddingTls to Tls mode
    /// This happens when XTLS is not supported or TLS 1.2 is detected
    fn switch_write_to_tls_mode(&mut self) -> io::Result<()> {
        if self.write_mode == VisionMode::Tls {
            return Ok(()); // Already switched
        }

        log::debug!("VISION WRITE: Switching to Tls mode (no XTLS support)");

        // Set write mode to Tls
        self.write_mode = VisionMode::Tls;

        Ok(())
    }

    /// Drain pending TCP writes to the underlying stream
    /// Returns Poll::Ready(Ok(())) when fully drained
    /// Returns Poll::Pending if TCP would block
    fn drain_pending_tcp_writes(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        use futures::ready;

        while !self.pending_tcp_writes.is_empty() {
            let n = ready!(Pin::new(&mut self.tcp).poll_write(cx, &self.pending_tcp_writes))?;

            if n == 0 {
                return Poll::Ready(Err(io::Error::new(
                    io::ErrorKind::WriteZero,
                    "Failed to write pending TCP data",
                )));
            }

            log::debug!(
                "VISION WRITE: Flushed {} pending TCP bytes ({} remaining)",
                n,
                self.pending_tcp_writes.len() - n
            );

            self.pending_tcp_writes.advance(n);
        }

        Poll::Ready(Ok(()))
    }

    /// Drain pending TLS records from session to pending_tcp_writes buffer
    /// Returns Poll::Ready(Ok(n)) where n is the number of bytes added to pending_tcp_writes
    /// Returns Poll::Pending if:
    ///   - We already have pending TCP writes (shouldn't extract more until TCP is clear), OR
    ///   - Already at threshold even though session.wants_write(), OR
    ///   - Session doesn't want to write
    fn drain_pending_tls_writes(&mut self, _cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
        use std::io::Cursor;

        const TCP_BUFFER_THRESHOLD: usize = 32 * 1024; // 32KB

        // Don't extract TLS records if we have pending TCP writes
        if !self.pending_tcp_writes.is_empty() {
            log::debug!(
                "VISION WRITE: Skipping TLS drain, {} TCP bytes still pending",
                self.pending_tcp_writes.len()
            );
            return Poll::Pending;
        }

        let session = self.session.as_mut().expect("session must exist");

        // Check if session wants to write
        if !session.wants_write() {
            return Poll::Pending;
        }

        // Check if we're already at threshold
        if self.pending_tcp_writes.len() >= TCP_BUFFER_THRESHOLD {
            log::debug!(
                "VISION WRITE: Already at TCP buffer threshold ({} bytes)",
                self.pending_tcp_writes.len()
            );
            return Poll::Pending;
        }

        let initial_len = self.pending_tcp_writes.len();

        // Extract TLS records from session until we hit the threshold
        while session.wants_write() && self.pending_tcp_writes.len() < TCP_BUFFER_THRESHOLD {
            // Extract ONE TLS record into buffer
            let mut tls_record_buf = vec![0u8; 16384];
            let mut cursor = Cursor::new(&mut tls_record_buf[..]);

            let encrypted_len = match session.write_tls(&mut cursor) {
                Ok(n) => n,
                Err(e) => {
                    return Poll::Ready(Err(io::Error::new(
                        io::ErrorKind::Other,
                        format!("TLS write_tls failed: {}", e),
                    )));
                }
            };

            if encrypted_len == 0 {
                break;
            }

            log::debug!(
                "VISION WRITE: Extracted {} encrypted bytes from TLS session to pending_tcp_writes",
                encrypted_len
            );

            // Store in pending_tcp_writes buffer
            self.pending_tcp_writes
                .extend_from_slice(&tls_record_buf[..encrypted_len]);
        }

        let bytes_added = self.pending_tcp_writes.len() - initial_len;

        if bytes_added > 0 {
            Poll::Ready(Ok(bytes_added))
        } else {
            Poll::Pending
        }
    }

    /// Drain pending plaintext writes to TLS session buffer
    /// Returns Poll::Ready(Ok(())) when fully drained
    /// Returns Poll::Pending if session buffer is full (needs TLS/TCP drain first)
    fn drain_pending_plain_writes(&mut self, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        use std::io::Write;

        let session = self
            .session
            .as_mut()
            .expect("session must exist when draining");

        if self.pending_plain_writes.is_empty() {
            return Poll::Ready(Ok(()));
        }

        log::debug!(
            "VISION WRITE: Draining {} pending plaintext bytes",
            self.pending_plain_writes.len()
        );

        // Try to write pending plaintext to TLS session
        while !self.pending_plain_writes.is_empty() {
            match session.writer().write(&self.pending_plain_writes) {
                Ok(n) => {
                    if n == 0 {
                        return Poll::Ready(Err(io::Error::new(
                            io::ErrorKind::WriteZero,
                            "Failed to write to TLS session buffer",
                        )));
                    }

                    self.pending_plain_writes.advance(n);
                }
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                    // Session buffer full - return Pending so drain_all_writes can drain TLS/TCP first
                    log::debug!(
                        "VISION WRITE: Session buffer full ({} plaintext bytes remaining), returning Pending",
                        self.pending_plain_writes.len()
                    );
                    return Poll::Pending;
                }
                Err(e) => return Poll::Ready(Err(e)),
            }
        }

        Poll::Ready(Ok(()))
    }

    /// Drain all pending writes: plain → TLS → TCP
    /// Returns Poll::Ready(Ok(())) when all buffers are fully drained
    /// Returns Poll::Pending if we've reached an impasse (all drains returned Pending)
    fn drain_all_writes(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        loop {
            // Check if everything is drained (do this without holding session borrow)
            let session_wants_write = self
                .session
                .as_ref()
                .expect("session must exist")
                .wants_write();

            if self.pending_plain_writes.is_empty()
                && !session_wants_write
                && self.pending_tcp_writes.is_empty()
            {
                return Poll::Ready(Ok(()));
            }

            // Drain in order: TCP → TLS → Plain
            // Start with lowest level (TCP) to create space in higher levels
            // Track if we made any progress in this iteration

            let mut made_progress = false;

            // Try TCP first
            if !self.pending_tcp_writes.is_empty() {
                match self.drain_pending_tcp_writes(cx) {
                    Poll::Ready(Ok(())) => {
                        made_progress = true;
                        continue; // Try next level
                    }
                    Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                    Poll::Pending => {
                        // TCP blocked, can't make progress
                    }
                }
            }

            // Try TLS if TCP is clear (check session.wants_write() without holding borrow)
            let session_wants_write = self
                .session
                .as_ref()
                .expect("session must exist")
                .wants_write();
            if session_wants_write && self.pending_tcp_writes.is_empty() {
                match self.drain_pending_tls_writes(cx) {
                    Poll::Ready(Ok(_bytes_added)) => {
                        made_progress = true;
                        continue; // Try next level
                    }
                    Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                    Poll::Pending => {
                        // TLS blocked (TCP would block or has pending)
                    }
                }
            }

            // Try Plain if TLS and TCP are clear (check session.wants_write() without holding borrow)
            let session_wants_write = self
                .session
                .as_ref()
                .expect("session must exist")
                .wants_write();
            if !self.pending_plain_writes.is_empty()
                && !session_wants_write
                && self.pending_tcp_writes.is_empty()
            {
                match self.drain_pending_plain_writes(cx) {
                    Poll::Ready(Ok(())) => {
                        made_progress = true;
                        continue; // Loop back to check if everything is drained
                    }
                    Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                    Poll::Pending => {
                        // Plain blocked (session buffer full), but TLS/TCP are clear
                        // This means we've reached an impasse
                    }
                }
            }

            // If we didn't make any progress, we're blocked
            if !made_progress {
                return Poll::Pending;
            }
        }
    }

    /// Write data to TLS session, handling WouldBlock by saving to pending buffer
    fn write_to_session(&mut self, data: &[u8]) -> io::Result<()> {
        use std::io::Write;

        // First, try to drain any existing pending data
        if !self.pending_plain_writes.is_empty() {
            // If there's already pending data, just append to it
            // (don't try to write, as that would require cx for polling)
            self.pending_plain_writes.extend_from_slice(data);
            return Ok(());
        }

        let session = self.session.as_mut().expect("session must exist");
        let mut written = 0;
        while written < data.len() {
            match session.writer().write(&data[written..]) {
                Ok(n) => {
                    if n == 0 {
                        return Err(io::Error::new(
                            io::ErrorKind::WriteZero,
                            "Failed to write to TLS session buffer",
                        ));
                    }
                    written += n;
                }
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                    // Session buffer full, save remainder
                    log::debug!(
                        "VISION WRITE: Session buffer full, saving {} bytes to pending",
                        data.len() - written
                    );
                    self.pending_plain_writes
                        .extend_from_slice(&data[written..]);
                    return Ok(());
                }
                Err(e) => return Err(e),
            }
        }

        Ok(())
    }

    /// Write data in Tls mode (no VISION padding, just TLS encryption)
    fn poll_write_tls(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
        use futures::ready;
        use std::io::Write;

        // Drain any pending writes first (backpressure)
        ready!(self.drain_all_writes(cx))?;

        let session = self
            .session
            .as_mut()
            .expect("session must exist in Tls mode");

        // Write directly to TLS session (no VISION padding)
        let mut written = 0;
        while written < buf.len() {
            match session.writer().write(&buf[written..]) {
                Ok(n) => {
                    if n == 0 {
                        return Poll::Ready(Err(io::Error::new(
                            io::ErrorKind::WriteZero,
                            "Failed to write to TLS session",
                        )));
                    }
                    written += n;
                }
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                    // Session buffer full, save remainder and drain
                    log::debug!(
                        "VISION WRITE TLS: Session buffer full, saving {} bytes to pending",
                        buf.len() - written
                    );
                    self.pending_plain_writes.extend_from_slice(&buf[written..]);
                    ready!(self.drain_all_writes(cx))?;
                    // Return Pending - next poll will drain pending_plain_writes
                    return Poll::Pending;
                }
                Err(e) => return Poll::Ready(Err(e)),
            }
        }

        // Drain all writes
        ready!(self.drain_all_writes(cx))?;

        Poll::Ready(Ok(buf.len()))
    }

    /// Read the VLESS response header from the TLS session
    ///
    /// The VLESS response consists of:
    /// - 1 byte: version (must be 0)
    /// - 1 byte: addon length
    /// - N bytes: addon data (if addon_length > 0)
    ///
    /// This is called lazily on the first read when vless_response_pending is true.
    /// Uses packetization to feed TLS records one at a time to avoid bad MAC errors.
    ///
    /// Returns any unused data that came after the response
    #[inline(always)]
    fn poll_read_vless_response(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<Vec<u8>>> {
        use futures::ready;
        use std::io::BufRead;

        let session = self
            .session
            .as_mut()
            .expect("session must exist when reading VLESS response");

        loop {
            // Always read from TCP to feed the packetizer
            let mut temp_tcp_buf = [0u8; 8192];
            let mut read_buf = ReadBuf::new(&mut temp_tcp_buf);
            match ready!(Pin::new(&mut self.tcp).poll_read(cx, &mut read_buf)) {
                Ok(()) => {}
                Err(e) => return Poll::Ready(Err(e)),
            };

            let tcp_bytes = read_buf.filled();

            if tcp_bytes.is_empty() {
                return Poll::Ready(Err(io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    "Connection closed while reading VLESS response",
                )));
            }

            log::debug!(
                "VLESS: Read {} bytes from TCP for response header",
                tcp_bytes.len()
            );

            // Feed to outer read packetizer
            self.outer_read_packetizer
                .as_mut()
                .expect("outer_read_packetizer must exist in PaddingTls mode")
                .feed(tcp_bytes);

            // Process TLS records one at a time
            loop {
                let tls_record = match self
                    .outer_read_packetizer
                    .as_mut()
                    .expect("outer_read_packetizer must exist in PaddingTls mode")
                    .next_packet()?
                {
                    Some(record) => record,
                    None => break, // Need more TCP data
                };

                log::debug!("VLESS: Processing {} byte TLS record", tls_record.len());

                // Feed single TLS record to session
                let mut cursor = std::io::Cursor::new(tls_record.as_ref());
                session.read_tls(&mut cursor)?;
                session
                    .process_new_packets()
                    .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;

                // Access the decrypted buffer and accumulate into partial_vless_response
                let mut reader = session.reader();
                let decrypted_buf = match reader.fill_buf() {
                    Ok(buf) => buf,
                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                        continue; // Process next TLS record
                    }
                    Err(e) => return Poll::Ready(Err(e)),
                };

                if decrypted_buf.is_empty() {
                    continue; // Process next TLS record
                }

                // Accumulate decrypted data into partial buffer
                self.partial_vless_response.extend_from_slice(decrypted_buf);
                let consumed_len = decrypted_buf.len();
                reader.consume(consumed_len);

                log::debug!(
                    "VLESS: Accumulated {} bytes, total buffered: {}",
                    consumed_len,
                    self.partial_vless_response.len()
                );

                // Try to parse the VLESS response from accumulated data
                if self.partial_vless_response.len() < 2 {
                    // Need at least 2 bytes for header
                    continue;
                }

                // Read the 2-byte header
                let version = self.partial_vless_response[0];
                let addon_length = self.partial_vless_response[1];

                // Validate version
                if version != 0 {
                    return Poll::Ready(Err(io::Error::new(
                        io::ErrorKind::InvalidData,
                        format!("Invalid VLESS response version: {}", version),
                    )));
                }

                // Check if we have enough data for the full response
                let total_response_len = 2 + addon_length as usize;
                if self.partial_vless_response.len() < total_response_len {
                    // Need more data - continue to next TLS record or read more TCP
                    continue;
                }

                // We have the complete VLESS response! Consume it from the buffer
                self.partial_vless_response.advance(total_response_len);

                log::debug!(
                    "VLESS: Successfully parsed {} byte response header (version={}, addon_length={}), {} bytes remaining in buffer",
                    total_response_len,
                    version,
                    addon_length,
                    self.partial_vless_response.len()
                );

                // If there's any remaining data in partial buffer, it's already TLS-decrypted
                // and needs to be unpadded (not fed to outer packetizer!)
                let remaining_data =
                    std::mem::replace(&mut self.partial_vless_response, BytesMut::new());

                if !remaining_data.is_empty() {
                    log::debug!(
                        "VLESS: {} bytes remaining after response header - feeding to unpadder",
                        remaining_data.len()
                    );
                }

                return Poll::Ready(Ok(remaining_data.to_vec()));
            }
        }
    }

    #[inline(always)]
    fn poll_read_padding_tls(
        &mut self,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
        mut previous_decrypted_data: Option<Vec<u8>>,
    ) -> Poll<io::Result<()>> {
        use futures::ready;
        use std::io::BufRead;

        let session = self
            .session
            .as_mut()
            .expect("session must exist in PaddingTls mode");

        loop {
            match previous_decrypted_data.take() {
                Some(data) => {
                    // This data is already TLS-decrypted (came from VLESS response parsing)
                    // Feed it directly to the unpadder
                    log::debug!(
                        "VISION READ: Processing {} bytes of previous decrypted data (from VLESS response)",
                        data.len()
                    );

                    let UnpadResult {
                        content: unpadded,
                        command: maybe_command,
                    } = self.read_unpadder.unpad(&data)?;

                    log::debug!("VISION READ: Unpadded to {} bytes", unpadded.len());

                    if !unpadded.is_empty() {
                        // Filter TLS patterns (INNER TLS detection)
                        Self::filter_data(&mut self.inner_read_packetizer, &mut self.filter, &unpadded);

                        // Store in pending_read and return
                        let len = buf.remaining().min(unpadded.len());
                        if len > 0 {
                            buf.put_slice(&unpadded[..len]);
                        }
                        if len < unpadded.len() {
                            self.pending_read.extend_from_slice(&unpadded[len..]);
                        }
                        return Poll::Ready(Ok(()));
                    }

                    // Check for mode switch commands
                    if let Some(cmd) = maybe_command {
                        match cmd {
                            UnpadCommand::Direct => {
                                log::debug!("VISION READ: Received DIRECT command from previous data");
                                self.read_mode = VisionMode::Direct;
                                self.peer_switched_direct = true;
                                return Poll::Ready(Ok(()));
                            }
                            UnpadCommand::End => {
                                log::debug!("VISION READ: Received END command from previous data");
                                self.read_mode = VisionMode::Tls;
                                return Poll::Ready(Ok(()));
                            }
                            UnpadCommand::Continue => {
                                // Continue processing
                            }
                        }
                    }

                    // If unpadder returned nothing, continue to read more TLS records
                }
                None => {
                    // Read from TCP to feed the packetizer
                    let mut temp_tcp_buf = [0u8; 8192];
                    let mut read_buf = ReadBuf::new(&mut temp_tcp_buf);
                    match ready!(Pin::new(&mut self.tcp).poll_read(cx, &mut read_buf)) {
                        Ok(()) => {}
                        Err(e) => return Poll::Ready(Err(e)),
                    };

                    let tcp_bytes = read_buf.filled();

                    if tcp_bytes.is_empty() {
                        // TCP connection closed
                        self.tls_state.shutdown_read();
                        return Poll::Ready(Ok(()));
                    }

                    log::debug!("VISION READ: Read {} bytes from TCP, first 20 bytes: {:?}",
                        tcp_bytes.len(),
                        &tcp_bytes[..tcp_bytes.len().min(20)]);

                    // Feed to outer TLS packetizer
                    let packetizer = self.outer_read_packetizer
                        .as_mut()
                        .expect("outer_read_packetizer must exist in PaddingTls mode");

                    log::debug!("VISION READ: Packetizer has {} pending bytes before feed",
                        packetizer.pending_bytes());

                    packetizer.feed(tcp_bytes);

                    log::debug!("VISION READ: Packetizer has {} pending bytes after feed",
                        packetizer.pending_bytes());
                }
            }

            // Process ALL TLS records first (following rustls Stream pattern)
            // Feed all available TLS records to the session BEFORE creating reader
            loop {
                let tls_record = match self
                    .outer_read_packetizer
                    .as_mut()
                    .expect("outer_read_packetizer must exist in PaddingTls mode")
                    .next_packet()?
                {
                    Some(record) => record,
                    None => {
                        log::debug!("VISION READ: No more complete TLS records in packetizer, need more TCP data");
                        break; // Need more TCP data - exit to read loop
                    }
                };

                log::debug!(
                    "VISION READ: *** PROCESSING {} byte TLS record, first 20 bytes: {:?} ***",
                    tls_record.len(),
                    &tls_record[..tls_record.len().min(20)]
                );

                // Feed single TLS record to session
                let mut cursor = std::io::Cursor::new(tls_record.as_ref());
                let tls_bytes_fed = session.read_tls(&mut cursor)?;
                log::debug!(
                    "VISION READ: Fed {} bytes to TLS session (from {} byte record)",
                    tls_bytes_fed,
                    tls_record.len()
                );

                // Try to process the TLS record
                // If we get a DecryptError, it likely means the peer has already switched to direct mode
                // and this "TLS record" is actually raw inner TLS data
                let tls_record_hex: String = tls_record[..tls_record.len().min(40)]
                    .iter()
                    .map(|b| format!("{:02x}", b))
                    .collect();

                match session.process_new_packets() {
                    Ok(io_state) => {
                        // Successfully decrypted, continue processing more records
                        log::debug!(
                            "VISION READ: TLS process_new_packets succeeded, io_state: plaintext_bytes_to_read={} tls_bytes_to_write={} peer_has_closed={}, record_hex={}",
                            io_state.plaintext_bytes_to_read(),
                            io_state.tls_bytes_to_write(),
                            io_state.peer_has_closed(),
                            tls_record_hex
                        );
                    }
                    Err(rustls::Error::DecryptError) => {
                        log::debug!(
                            "VISION READ: DecryptError on {} byte record - peer likely switched to direct mode",
                            tls_record.len()
                        );

                        // CRITICAL: Before switching modes, drain ANY plaintext that was successfully
                        // decrypted from previous TLS records! The session may have plaintext buffered
                        // from records we processed before hitting this DecryptError.
                        let mut previously_decrypted = Vec::new();
                        {
                            let mut reader = session.reader();
                            loop {
                                match reader.fill_buf() {
                                    Ok(decrypted_buf) if !decrypted_buf.is_empty() => {
                                        log::debug!(
                                            "VISION READ: Draining {} bytes of plaintext before mode switch",
                                            decrypted_buf.len()
                                        );
                                        previously_decrypted.extend_from_slice(decrypted_buf);
                                        let consumed_len = decrypted_buf.len();
                                        reader.consume(consumed_len);
                                    }
                                    _ => break,
                                }
                            }
                        }

                        // Unpad any drained plaintext
                        if !previously_decrypted.is_empty() {
                            log::debug!(
                                "VISION READ: Unpadding {} bytes of previously decrypted data before mode switch",
                                previously_decrypted.len()
                            );
                            let UnpadResult {
                                content: unpadded,
                                command: maybe_command,
                            } = self.read_unpadder.unpad(&previously_decrypted)?;

                            if !unpadded.is_empty() {
                                // Filter TLS patterns (INNER TLS detection)
                                Self::filter_data(&mut self.inner_read_packetizer, &mut self.filter, &unpadded);

                                // Store in pending_read
                                self.pending_read.extend_from_slice(&unpadded);
                                log::debug!(
                                    "VISION READ: Stored {} unpadded bytes in pending_read before mode switch",
                                    unpadded.len()
                                );
                            }

                            // Handle command from previously decrypted data
                            if let Some(cmd) = maybe_command {
                                log::debug!(
                                    "VISION READ: Found command {:?} in previously decrypted data",
                                    cmd
                                );
                            }
                        }

                        // CRITICAL: The failed TLS record is encrypted outer TLS data that we can't decrypt.
                        // It's NOT raw inner TLS data! We must DISCARD it.
                        // Only extract any remaining raw data AFTER the failed record from the packetizer.
                        let packetizer_remaining = self
                            .outer_read_packetizer
                            .take()
                            .expect("outer_read_packetizer must exist")
                            .into_remaining_data();

                        log::debug!(
                            "VISION READ: Discarding {} byte failed TLS record (encrypted garbage), extracted {} bytes of remaining data from packetizer",
                            tls_record.len(),
                            packetizer_remaining.len()
                        );

                        // Switch to direct mode
                        self.read_mode = VisionMode::Direct;
                        self.peer_switched_direct = true;

                        // Return any pending_read data (previously decrypted and unpadded)
                        if !self.pending_read.is_empty() {
                            let len = buf.remaining().min(self.pending_read.len());
                            buf.put_slice(&self.pending_read[..len]);
                            self.pending_read.advance(len);

                            // Store any remaining packetizer data for next read (should be rare)
                            if !packetizer_remaining.is_empty() {
                                self.pending_read.extend_from_slice(&packetizer_remaining);
                            }

                            log::debug!(
                                "VISION READ: Switched to direct mode, returning {} previously decrypted bytes ({} raw bytes buffered)",
                                len,
                                packetizer_remaining.len()
                            );
                        } else if !packetizer_remaining.is_empty() {
                            // No previously decrypted data, but we have raw data from packetizer
                            let len = buf.remaining().min(packetizer_remaining.len());
                            buf.put_slice(&packetizer_remaining[..len]);
                            if len < packetizer_remaining.len() {
                                self.pending_read.extend_from_slice(&packetizer_remaining[len..]);
                            }

                            log::debug!(
                                "VISION READ: Switched to direct mode, returning {} raw bytes from packetizer",
                                len
                            );
                        } else {
                            // No data to return, switch to direct mode and wait for next TCP read
                            log::debug!(
                                "VISION READ: Switched to direct mode, no data to return (will read fresh TCP data)"
                            );
                        }

                        return Poll::Ready(Ok(()));
                    }
                    Err(e) => {
                        // Other TLS errors - propagate
                        return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, e)));
                    }
                }

                // Continue to process next TLS record without reading plaintext yet
            }

            // NOW create ONE reader and drain ALL available plaintext in a loop
            // This follows rustls Stream pattern: process all TLS first, then drain ALL plaintext
            // CRITICAL: We must loop here to drain all chunks, not just one!
            let mut reader = session.reader();
            loop {
                let decrypted = match reader.fill_buf() {
                    Ok(buf) => buf,
                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                        // No more plaintext available - loop back to read more TCP
                        break;
                    }
                    Err(e) => return Poll::Ready(Err(e)),
                };

                if decrypted.is_empty() {
                    // No more plaintext - loop back to read more TCP
                    break;
                }

                let decrypted_len = decrypted.len();
                let log_len = decrypted_len.min(100);
                let hex_dump = decrypted[..log_len]
                    .iter()
                    .map(|b| format!("{:02x}", b))
                    .collect::<String>();
                log::debug!(
                    "[SHOES DECRYPTED] total={} first100hex={}",
                    decrypted_len,
                    hex_dump
                );

                // Feed newly decrypted bytes to the unpadder.
                // All data should be consumed from the rustls session, as any actual content
                // will eventually be returned by the unpadder.
                // The unpadder maintains internal state for partial parsing
                let UnpadResult {
                    content: mut unpadded,
                    command: maybe_command,
                } = self.read_unpadder.unpad(decrypted)?;
                reader.consume(decrypted_len);

                log::debug!("VISION READ: Unpadded to {} bytes", unpadded.len());

            if !unpadded.is_empty() {
                // Filter TLS patterns (INNER TLS detection)
                Self::filter_data(&mut self.inner_read_packetizer, &mut self.filter, &unpadded);
            }

            match maybe_command {
                Some(UnpadCommand::Direct) => {
                    log::debug!("VISION READ: Received DIRECT command");

                    // Extract remaining data from outer read packetizer
                    let remaining = self
                        .outer_read_packetizer
                        .take()
                        .expect("outer_read_packetizer must exist")
                        .into_remaining_data();

                    // Switch to Direct mode
                    self.read_mode = VisionMode::Direct;

                    if !remaining.is_empty() {
                        log::debug!(
                            "VISION READ: Extracted {} bytes from packetizer as raw data",
                            remaining.len()
                        );
                        unpadded.extend_from_slice(&remaining);
                    }

                    self.peer_switched_direct = true;

                    // Return unpadded data if any, or return ready with 0 bytes
                    // This ensures we don't continue processing more "TLS records" after DIRECT
                    let len = buf.remaining().min(unpadded.len());
                    if len > 0 {
                        buf.put_slice(&unpadded[..len]);
                    }
                    if len < unpadded.len() {
                        self.pending_read.extend_from_slice(&unpadded[len..]);
                    }
                    return Poll::Ready(Ok(()));
                }
                Some(UnpadCommand::End) => {
                    log::debug!("VISION READ: Received END command, switching to Tls mode");

                    // Drop reader FIRST to release the borrow on session
                    drop(reader);

                    // Extract remaining data from outer read packetizer
                    let remaining = self
                        .outer_read_packetizer
                        .take()
                        .expect("outer_read_packetizer must exist")
                        .into_remaining_data();

                    // Switch to Tls mode
                    self.read_mode = VisionMode::Tls;

                    if !remaining.is_empty() {
                        log::debug!(
                            "VISION READ: Extracted {} bytes from packetizer as raw data",
                            remaining.len()
                        );
                        unpadded.extend_from_slice(&remaining);

                        log::debug!(
                            "VISION READ: Feeding {} bytes from packetizer to TLS session",
                            remaining.len()
                        );
                        // Feed remaining data to TLS session
                        let mut cursor = std::io::Cursor::new(remaining.as_ref());
                        session.read_tls(&mut cursor)?;
                        session
                            .process_new_packets()
                            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
                    }

                    // Return unpadded data if any
                    if !unpadded.is_empty() {
                        let len = buf.remaining().min(unpadded.len());
                        buf.put_slice(&unpadded[..len]);
                        if len < unpadded.len() {
                            self.pending_read.extend_from_slice(&unpadded[len..]);
                        }
                        return Poll::Ready(Ok(()));
                    }

                    // Break out of inner loop since we switched modes
                    break;
                }
                Some(UnpadCommand::Continue) => {
                    if !unpadded.is_empty() {
                        // Log the actual unpadded content for debugging
                        let log_len = unpadded.len().min(100);
                        let hex_dump = unpadded[..log_len]
                            .iter()
                            .map(|b| format!("{:02x}", b))
                            .collect::<String>();
                        log::debug!(
                            "VISION READ: Returning {} unpadded bytes (CONTINUE), first100hex={}",
                            unpadded.len(),
                            hex_dump
                        );

                        let len = buf.remaining().min(unpadded.len());
                        buf.put_slice(&unpadded[..len]);
                        if len < unpadded.len() {
                            self.pending_read.extend_from_slice(&unpadded[len..]);
                        }
                        return Poll::Ready(Ok(()));
                    }
                }
                None => {
                    // Might not be TLS, return data
                    if !unpadded.is_empty() {
                        let len = buf.remaining().min(unpadded.len());
                        buf.put_slice(&unpadded[..len]);
                        if len < unpadded.len() {
                            self.pending_read.extend_from_slice(&unpadded[len..]);
                        }
                        return Poll::Ready(Ok(()));
                    }
                    log::debug!(
                        "VISION READ: Unpadder needs more data, will read more TLS records"
                    );
                }
            }
            } // Close inner plaintext draining loop
            // Need more TCP data, loop back to read
        }
    }

    fn poll_read_tls(
        &mut self,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        // Regular TLS mode (no padding)
        // Similar to poll_read_vless_response logic
        use futures::ready;
        use std::io::BufRead;

        let session = self
            .session
            .as_mut()
            .expect("session must exist in Tls mode");

        let mut reader = session.reader();
        match reader.fill_buf() {
            Ok(decrypted_buf) => {
                if !decrypted_buf.is_empty() {
                    let len = buf.remaining().min(decrypted_buf.len());
                    buf.put_slice(&decrypted_buf[..len]);
                    reader.consume(len);
                    return Poll::Ready(Ok(()));
                }
            }
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
            Err(e) => return Poll::Ready(Err(e)),
        };

        loop {
            let mut read_buf = ReadBuf::new(&mut self.tls_read_buffer);
            match ready!(Pin::new(&mut self.tcp).poll_read(cx, &mut read_buf)) {
                Ok(()) => {}
                Err(e) => return Poll::Ready(Err(e)),
            };
            let tcp_bytes_read = read_buf.filled().len();

            if tcp_bytes_read == 0 {
                self.tls_state.shutdown_read();
                return Poll::Ready(Ok(()));
            }

            // Feed to TLS session
            let mut cursor = std::io::Cursor::new(&self.tls_read_buffer[..tcp_bytes_read]);
            session.read_tls(&mut cursor)?;
            session
                .process_new_packets()
                .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;

            // Access the decrypted buffer
            let mut reader = session.reader();
            let decrypted_buf = match reader.fill_buf() {
                Ok(buf) => buf,
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                    drop(reader);
                    continue;
                }
                Err(e) => return Poll::Ready(Err(e)),
            };

            if decrypted_buf.is_empty() {
                continue;
            }

            // Copy to output buffer
            let len = buf.remaining().min(decrypted_buf.len());
            buf.put_slice(&decrypted_buf[..len]);
            reader.consume(len);

            return Poll::Ready(Ok(()));
        }
    }

    fn poll_write_padding_tls(
        &mut self,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        use futures::ready;

        // STEP 1: Drain all pending writes (backpressure) - CRITICAL: must drain before mode switch!
        // The session could contain the DIRECT/END packet that triggered the mode switch flag.
        // If we switch modes before draining, that packet might never get sent!
        {
            let session = self
                .session
                .as_mut()
                .expect("session must exist in PaddingTls mode");

            if !self.pending_plain_writes.is_empty()
                || session.wants_write()
                || !self.pending_tcp_writes.is_empty()
            {
                ready!(self.drain_all_writes(cx))?;
            }
        }

        // STEP 3: Check if we should switch to direct mode from previous write
        // This flag is set AFTER sending the DIRECT packet, so by now the packet is drained
        if self.pending_direct_mode_switch {
            log::debug!("VISION WRITE: Switching to direct mode (flag set by previous write)");
            self.pending_direct_mode_switch = false;
            self.switch_write_to_direct_mode()?;
            return Pin::new(&mut self.tcp).poll_write(cx, buf);
        }

        if self.pending_tls_mode_switch {
            log::debug!("VISION WRITE: Switching to Tls mode (flag set by previous write)");
            self.pending_tls_mode_switch = false;
            self.switch_write_to_tls_mode()?;
            return self.poll_write_tls(cx, buf);
        }

        // STEP 4: Feed write buffer to inner packetizer and check for ApplicationData
        let existing_inner_len = self.inner_write_packetizer.pending_bytes();
        self.inner_write_packetizer.feed(buf);

        // Process all complete TLS packets in the buffer
        let mut processed_len = 0;
        loop {
            match self.inner_write_packetizer.next_packet() {
                Ok(Some(packet)) => {
                    processed_len += packet.len();

                    // Feed to filter for TLS pattern detection
                    self.filter.filter_packet(&packet);

                    // Check if this packet is ApplicationData
                    if self.filter.is_tls() && packet.len() >= 3
                        && packet[0] == 0x17  // ApplicationData
                        && packet[1] == 0x03
                    // TLS version major
                    {
                        log::debug!(
                            "VISION WRITE: Detected ApplicationData in {} byte packet",
                            packet.len()
                        );
                        // DIRECT or END
                        let command = if self.filter.supports_xtls() {
                            0x02
                        } else {
                            0x01
                        };

                        let final_padded_packet = if self.write_first_packet {
                            self.write_first_packet = false;
                            super::vision_pad::pad_with_uuid_and_command(
                                &packet,
                                &self.user_uuid,
                                command,
                                true, // is_tls
                            )
                        } else {
                            super::vision_pad::pad_with_command(
                                &packet, command, true, // is_tls
                            )
                        };

                        self.write_to_session(&final_padded_packet)?;

                        // Set flag to switch on next write
                        if self.filter.supports_xtls() {
                            self.pending_direct_mode_switch = true;
                        } else {
                            self.pending_tls_mode_switch = true;
                        }

                        // this must be true because else we would have a successful next_packet call on previous iteration
                        assert!(processed_len > existing_inner_len);

                        // Drain and handle result
                        match self.drain_all_writes(cx) {
                            Poll::Ready(Ok(())) => {
                                // Fully drained, continue
                            }
                            Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                            Poll::Pending => {
                                // Still draining - clear packetizer since caller will re-feed remaining data
                                self.inner_write_packetizer.clear();
                            }
                        }

                        // tell caller to resend the next packets after the mode switch
                        return Poll::Ready(Ok(processed_len - existing_inner_len));
                    }

                    if !self.filter.is_filtering() && !self.filter.is_tls12_or_above() {
                        let final_padded_packet = if self.write_first_packet {
                            self.write_first_packet = false;
                            super::vision_pad::pad_with_uuid_and_command(
                                &packet,
                                &self.user_uuid,
                                0x01, // END
                                self.filter.is_tls(),
                            )
                        } else {
                            super::vision_pad::pad_with_command(
                                &packet,
                                0x01, // END
                                self.filter.is_tls(),
                            )
                        };

                        self.write_to_session(&final_padded_packet)?;

                        // Set flag to switch on next write
                        self.pending_tls_mode_switch = true;

                        // this must be true because else we would have a successful next_packet call on previous iteration
                        assert!(processed_len > existing_inner_len);

                        // Drain and handle result
                        match self.drain_all_writes(cx) {
                            Poll::Ready(Ok(())) => {
                                // Fully drained, continue
                            }
                            Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                            Poll::Pending => {
                                // Still draining - clear packetizer since caller will re-feed remaining data
                                self.inner_write_packetizer.clear();
                            }
                        }

                        // tell caller to resend the next packets after the mode switch
                        return Poll::Ready(Ok(processed_len - existing_inner_len));
                    }

                    // if we got here, it's a TLS packet that is not app data and we need to continue padding
                    // at this point is_filtering() might already be false, but we continue as we are tls 1.2 or above
                    // ref: https://github.com/XTLS/Xray-core/blob/9f5dcb15910aadc7ef450514747576827a389853/proxy/proxy.go#L371

                    let padded_packet = if self.write_first_packet {
                        self.write_first_packet = false;
                        super::vision_pad::pad_with_uuid_and_command(
                            &packet,
                            &self.user_uuid,
                            0x00, // CONTINUE
                            self.filter.is_tls(),
                        )
                    } else {
                        super::vision_pad::pad_with_command(
                            &packet,
                            0x00, // CONTINUE
                            self.filter.is_tls(),
                        )
                    };

                    self.write_to_session(&padded_packet)?;
                    match self.drain_all_writes(cx) {
                        Poll::Pending => {
                            let unprocessed_buf_len =
                                buf.len() - (processed_len - existing_inner_len);
                            // sanity check
                            assert!(
                                self.inner_write_packetizer.pending_bytes() == unprocessed_buf_len
                            );
                            // clear since the user will re-feed
                            self.inner_write_packetizer.clear();
                            return Poll::Ready(Ok(processed_len - existing_inner_len));
                        }
                        Poll::Ready(Ok(())) => {
                            // continue since it drained
                        }
                        Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                    }
                }
                Ok(None) => {
                    // need more data
                    // TODO: investigate this case. it's possible that this is not TLS data and the packetizer is waiting
                    // for more writes from the caller to complete a full TLS record, but no more writes are coming.
                    // do we need to change our packetizer approach?
                    break;
                }
                Err(e) => {
                    log::debug!("VISION WRITE: Packetization failed: {}", e);
                    // packetization failed, to match Xray-core behavior, we treat this as a single packet
                    // TODO: check Xray-core about padding lengths and see if we need to split into multiple packets

                    self.filter.decrement_filter_count();

                    // continue or end command
                    let command = if self.filter.is_filtering() {
                        0x00
                    } else {
                        0x01
                    };
                    let remaining_data =
                        std::mem::replace(&mut self.inner_write_packetizer, TlsPacketizer::new())
                            .into_remaining_data();

                    let padded_packet = if self.write_first_packet {
                        self.write_first_packet = false;
                        super::vision_pad::pad_with_uuid_and_command(
                            &remaining_data,
                            &self.user_uuid,
                            command,
                            self.filter.is_tls(),
                        )
                    } else {
                        super::vision_pad::pad_with_command(
                            &remaining_data,
                            command,
                            self.filter.is_tls(),
                        )
                    };

                    self.write_to_session(&padded_packet)?;
                    ready!(self.drain_all_writes(cx))?;

                    return Poll::Ready(Ok(buf.len()));
                }
            }
        }

        // we processed all packets if we got here
        Poll::Ready(Ok(buf.len()))
    }
}

// AsyncRead implementation
impl<IO> AsyncRead for VisionStream<IO>
where
    IO: AsyncRead + AsyncWrite + Unpin,
{
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        let this = self.get_mut();

        // If we have leftover buffered data, return it first
        if !this.pending_read.is_empty() {
            let len = buf.remaining().min(this.pending_read.len());
            buf.put_slice(&this.pending_read[..len]);
            this.pending_read.advance(len);
            return Poll::Ready(Ok(()));
        }

        match this.read_mode {
            VisionMode::PaddingTls => {
                let previous_read_data = if this.vless_response_pending {
                    match this.poll_read_vless_response(cx) {
                        Poll::Ready(Ok(remaining_data)) => {
                            this.vless_response_pending = false;
                            if remaining_data.is_empty() {
                                None
                            } else {
                                Some(remaining_data)
                            }
                        }
                        Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                        Poll::Pending => return Poll::Pending,
                    }
                } else {
                    None
                };
                this.poll_read_padding_tls(cx, buf, previous_read_data)
            }
            VisionMode::Tls => this.poll_read_tls(cx, buf),
            VisionMode::Direct => Pin::new(&mut this.tcp).poll_read(cx, buf),
        }
    }
}

// AsyncWrite implementation
impl<IO> AsyncWrite for VisionStream<IO>
where
    IO: AsyncRead + AsyncWrite + Unpin,
{
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        let this = self.get_mut();

        match this.write_mode {
            VisionMode::PaddingTls => this.poll_write_padding_tls(cx, buf),
            VisionMode::Tls => this.poll_write_tls(cx, buf),
            VisionMode::Direct => Pin::new(&mut this.tcp).poll_write(cx, buf),
        }
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        use futures::ready;

        let this = self.get_mut();

        match this.write_mode {
            VisionMode::PaddingTls | VisionMode::Tls => {
                // Drain all pending writes
                ready!(this.drain_all_writes(cx))?;
                Pin::new(&mut this.tcp).poll_flush(cx)
            }
            VisionMode::Direct => Pin::new(&mut this.tcp).poll_flush(cx),
        }
    }

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        Pin::new(&mut self.get_mut().tcp).poll_shutdown(cx)
    }
}

// Implement AsyncPing trait
impl<IO> crate::async_stream::AsyncPing for VisionStream<IO>
where
    IO: AsyncStream,
{
    fn supports_ping(&self) -> bool {
        self.tcp.supports_ping()
    }

    fn poll_write_ping(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<bool>> {
        Pin::new(&mut self.get_mut().tcp).poll_write_ping(cx)
    }
}

impl<IO> AsyncStream for VisionStream<IO> where IO: AsyncStream {}
