//! Reusable TCP client for connecting to a fafb-q server.
//!
//! `FafbClient` splits the connection into read and write halves. Writes go
//! through an mpsc channel consumed by a dedicated writer task, and a
//! background keepalive task sends `Ping` when idle. It supports both plain
//! TCP and TLS and is the foundation for CLI commands and a future Rust SDK.

use std::io;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;

use futures::SinkExt;
use futures::stream::{SplitSink, SplitStream};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio::net::TcpStream;
use tokio::sync::{Notify, mpsc};
use tokio_util::codec::Framed;

use bytes::Bytes;

use crate::codec::MessageCodec;
use crate::codec::frame::{
    CAP_HEADERS, CAP_MESSAGE_TTL, Capabilities, FetchedMessage, Frame, SERVER_CAPABILITIES,
};
use crate::error::Error;
use crate::types::{ApiKey, Headers, MessageId, NamespaceId, TopicName, TopicOffset};

/// Default max frame size used by the client (1 MB).
const DEFAULT_MAX_FRAME_SIZE: u32 = 1_048_576;

// ---------------------------------------------------------------------------
// ClientStream — enum over plain TCP and TLS
// ---------------------------------------------------------------------------

/// Unified stream type that supports both plain TCP and TLS connections.
pub enum ClientStream {
    Plain(TcpStream),
    Tls(Box<tokio_rustls::client::TlsStream<TcpStream>>),
}

impl AsyncRead for ClientStream {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        match self.get_mut() {
            ClientStream::Plain(s) => Pin::new(s).poll_read(cx, buf),
            ClientStream::Tls(s) => Pin::new(s).poll_read(cx, buf),
        }
    }
}

impl AsyncWrite for ClientStream {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        match self.get_mut() {
            ClientStream::Plain(s) => Pin::new(s).poll_write(cx, buf),
            ClientStream::Tls(s) => Pin::new(s).poll_write(cx, buf),
        }
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match self.get_mut() {
            ClientStream::Plain(s) => Pin::new(s).poll_flush(cx),
            ClientStream::Tls(s) => Pin::new(s).poll_flush(cx),
        }
    }

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match self.get_mut() {
            ClientStream::Plain(s) => Pin::new(s).poll_shutdown(cx),
            ClientStream::Tls(s) => Pin::new(s).poll_shutdown(cx),
        }
    }
}

// ---------------------------------------------------------------------------
// TLS configuration for the client
// ---------------------------------------------------------------------------

/// TLS settings for client connections.
#[derive(Debug, Clone, Default)]
pub struct ClientTlsConfig {
    /// Enable TLS.
    pub enabled: bool,
    /// CA certificate file for server verification. If `None`, uses
    /// the system default root certificates (via `rustls` webpki roots).
    pub ca_cert_file: Option<String>,
    /// Client certificate file for mTLS.
    pub client_cert_file: Option<String>,
    /// Client key file for mTLS.
    pub client_key_file: Option<String>,
    /// Server name for SNI (defaults to hostname from server address).
    pub server_name: Option<String>,
}

// ---------------------------------------------------------------------------
// ClientConfig
// ---------------------------------------------------------------------------

/// Configuration for establishing a client connection.
#[derive(Debug, Clone)]
pub struct ClientConfig {
    /// Server address in `host:port` form.
    pub server: String,
    /// Namespace to authenticate against.
    pub namespace: NamespaceId,
    /// API key for authentication.
    pub api_key: ApiKey,
    /// Connection timeout.
    pub timeout: Duration,
    /// Maximum frame size.
    pub max_frame_size: u32,
    /// TLS configuration. When `tls.enabled` is true, the connection is
    /// wrapped in TLS before the protocol handshake.
    pub tls: ClientTlsConfig,
    /// When true, print protocol frames to stderr for debugging.
    pub verbose: bool,
    /// Keepalive ping interval. The client sends Ping frames at this interval
    /// when idle. Set to Duration::ZERO to disable keepalive.
    /// Default: 10 seconds.
    pub ping_interval: Duration,
}

impl ClientConfig {
    pub fn new(
        server: impl Into<String>,
        namespace: impl Into<String>,
        api_key: impl Into<String>,
    ) -> Self {
        Self {
            server: server.into(),
            namespace: NamespaceId(namespace.into()),
            api_key: ApiKey(api_key.into()),
            timeout: Duration::from_secs(30),
            max_frame_size: DEFAULT_MAX_FRAME_SIZE,
            tls: ClientTlsConfig::default(),
            verbose: false,
            ping_interval: Duration::from_secs(10),
        }
    }
}

// ---------------------------------------------------------------------------
// FafbClient
// ---------------------------------------------------------------------------

/// A client connection to a fafb-q server (plain TCP or TLS).
///
/// After `connect()`, the client has authenticated and negotiated capabilities.
/// Writes go through an mpsc channel consumed by a background writer task, and
/// a keepalive task sends `Ping` when idle.
pub struct FafbClient {
    /// `None` after `close()` has been called.
    frame_tx: Option<mpsc::Sender<Frame>>,
    reader: SplitStream<Framed<ClientStream, MessageCodec>>,
    capabilities: Capabilities,
    namespace: NamespaceId,
    verbose: bool,
    activity: Arc<Notify>,
    writer_handle: tokio::task::JoinHandle<()>,
    keepalive_handle: Option<tokio::task::JoinHandle<()>>,
}

impl FafbClient {
    /// Connect to a fafb-q server, authenticate, and negotiate capabilities.
    ///
    /// When `config.tls.enabled` is true, the TCP connection is wrapped in
    /// TLS before the protocol handshake. The client advertises all known
    /// capabilities; the server's ConnectAck returns the intersection.
    pub async fn connect(config: &ClientConfig) -> Result<Self, Error> {
        let tcp_stream = tokio::time::timeout(config.timeout, TcpStream::connect(&config.server))
            .await
            .map_err(|_| Error::Config(format!("connection to {} timed out", config.server)))?
            .map_err(|e| Error::Config(format!("failed to connect to {}: {e}", config.server)))?;

        tcp_stream.set_nodelay(true).ok();

        let stream = if config.tls.enabled {
            let tls_stream = wrap_tls(tcp_stream, config).await?;
            ClientStream::Tls(Box::new(tls_stream))
        } else {
            ClientStream::Plain(tcp_stream)
        };

        let codec = MessageCodec::new(config.max_frame_size);
        let mut framed = Framed::new(stream, codec);
        let verbose = config.verbose;

        // Send Connect frame with all capabilities the client supports.
        let client_caps = SERVER_CAPABILITIES;
        let connect_frame = Frame::Connect {
            version: 1,
            namespace: config.namespace.clone(),
            api_key: config.api_key.clone(),
            capabilities: client_caps,
        };

        if verbose {
            log_frame_send(&connect_frame);
        }

        framed
            .send(connect_frame)
            .await
            .map_err(|e| Error::Config(format!("failed to send Connect frame: {e}")))?;

        // Receive ConnectAck.
        use futures::StreamExt;
        let ack = tokio::time::timeout(config.timeout, framed.next())
            .await
            .map_err(|_| Error::Config("timed out waiting for ConnectAck".into()))?
            .ok_or_else(|| Error::Config("connection closed before ConnectAck".into()))?
            .map_err(|e| Error::Config(format!("failed to read ConnectAck: {e}")))?;

        if verbose {
            log_frame_recv(&ack);
        }

        match ack {
            Frame::ConnectAck {
                success,
                message,
                server_capabilities,
            } => {
                if !success {
                    return Err(Error::Config(format!("authentication failed: {message}")));
                }

                let negotiated = client_caps & server_capabilities;

                // Enable codec features based on negotiated capabilities.
                let codec = framed.codec_mut();
                if negotiated & CAP_HEADERS != 0 {
                    codec.set_headers_enabled(true);
                }
                if negotiated & CAP_MESSAGE_TTL != 0 {
                    codec.set_ttl_enabled(true);
                }

                // Split the stream into read/write halves.
                use futures::StreamExt;
                let (writer, reader) = framed.split();
                let (frame_tx, frame_rx) = mpsc::channel(64);
                let writer_handle = tokio::spawn(writer_loop(writer, frame_rx));
                let activity = Arc::new(Notify::new());
                let keepalive_handle = if config.ping_interval > Duration::ZERO {
                    Some(tokio::spawn(keepalive_loop(
                        frame_tx.clone(),
                        Arc::clone(&activity),
                        config.ping_interval,
                    )))
                } else {
                    None
                };

                Ok(Self {
                    frame_tx: Some(frame_tx),
                    reader,
                    capabilities: negotiated,
                    namespace: config.namespace.clone(),
                    verbose,
                    activity,
                    writer_handle,
                    keepalive_handle,
                })
            }
            Frame::Error { code, message, .. } => Err(Error::Config(format!(
                "server returned error on connect: [{code:?}] {message}"
            ))),
            other => Err(Error::Config(format!(
                "unexpected frame during connect: {other:?}"
            ))),
        }
    }

    /// Returns the negotiated capabilities bitmask.
    pub fn capabilities(&self) -> Capabilities {
        self.capabilities
    }

    /// Returns the namespace this client is authenticated to.
    pub fn namespace(&self) -> &NamespaceId {
        &self.namespace
    }

    // -----------------------------------------------------------------------
    // Low-level frame send/receive
    // -----------------------------------------------------------------------

    /// Send a single frame via the mpsc writer channel.
    pub async fn send_frame(&mut self, frame: Frame) -> Result<(), Error> {
        if self.verbose {
            log_frame_send(&frame);
        }
        self.activity.notify_one();
        let tx = self
            .frame_tx
            .as_ref()
            .ok_or_else(|| Error::Config("connection closed".into()))?;
        tx.send(frame)
            .await
            .map_err(|e| Error::Config(format!("failed to send frame: {e}")))
    }

    /// Receive a single frame. Returns an error if the connection is closed.
    ///
    /// `Frame::Ping` frames sent by the server are handled transparently: a
    /// `Frame::Pong` reply is sent automatically and the loop continues until a
    /// non-Ping frame arrives.  Callers never observe `Frame::Ping`.
    pub async fn recv_frame(&mut self) -> Result<Frame, Error> {
        use futures::StreamExt;
        loop {
            let frame = self
                .reader
                .next()
                .await
                .ok_or_else(|| Error::Config("connection closed".into()))?
                .map_err(|e| Error::Config(format!("failed to receive frame: {e}")))?;

            if self.verbose {
                log_frame_recv(&frame);
            }

            if matches!(frame, Frame::Ping) {
                // Auto-reply Pong via the writer channel and continue.
                if let Some(tx) = &self.frame_tx {
                    let _ = tx.send(Frame::Pong).await;
                }
                continue;
            }

            if matches!(frame, Frame::Pong) {
                // Server replied to one of our keepalive Pings — discard silently.
                continue;
            }

            return Ok(frame);
        }
    }

    // -----------------------------------------------------------------------
    // High-level operations
    // -----------------------------------------------------------------------

    /// Publish a message and wait for the `PublishAck`.
    ///
    /// Returns the server-assigned `MessageId`. When `wait_ack` is false the
    /// frame is sent but we do not wait for a response (fire-and-forget).
    pub async fn publish(
        &mut self,
        topic: TopicName,
        payload: Bytes,
        headers: Headers,
        ttl_ms: u32,
    ) -> Result<PublishResult, Error> {
        self.send_frame(Frame::Publish {
            topic: topic.clone(),
            headers,
            ttl_ms,
            payload,
        })
        .await?;

        let resp = self.recv_frame().await?;

        match resp {
            Frame::PublishAck { message_id } => Ok(PublishResult { message_id }),
            Frame::Error { code, message, .. } => Err(Error::Config(format!(
                "publish error [{code:?}]: {message}"
            ))),
            other => Err(Error::Config(format!(
                "unexpected frame after Publish: {other:?}"
            ))),
        }
    }

    /// Send a Publish frame and drain the server's ack without blocking on it.
    ///
    /// The server always sends a `PublishAck` — this method sends the frame,
    /// then spawns a brief background drain so the ack doesn't pollute the
    /// stream for subsequent operations. If you need maximum throughput for
    /// bulk no-ack publishes, use `send_frame()` in a loop without waiting
    /// for acks.
    pub async fn publish_no_ack(
        &mut self,
        topic: TopicName,
        payload: Bytes,
        headers: Headers,
        ttl_ms: u32,
    ) -> Result<(), Error> {
        self.send_frame(Frame::Publish {
            topic,
            headers,
            ttl_ms,
            payload,
        })
        .await?;

        // Drain the PublishAck so subsequent operations see a clean stream.
        // Use a short timeout — the ack should arrive quickly.
        let drain =
            tokio::time::timeout(std::time::Duration::from_secs(5), self.recv_frame()).await;

        // We don't care about the result — just ensure the frame is consumed.
        // If it times out or errors, the stream is still usable for the next op.
        let _ = drain;

        Ok(())
    }

    /// Fetch messages starting at `offset`.
    pub async fn fetch(
        &mut self,
        topic: TopicName,
        offset: TopicOffset,
        max_count: u32,
    ) -> Result<FetchResult, Error> {
        self.send_frame(Frame::Fetch {
            topic,
            offset,
            max_count,
        })
        .await?;

        self.recv_fetch_response().await
    }

    /// Fetch the last `count` messages from a topic (tail).
    ///
    /// Resolves client-side: first discovers `head_offset` via a lightweight
    /// fetch, then fetches `head - count` to `head`. Returns fewer than
    /// `count` if the topic has fewer messages.
    pub async fn fetch_tail(&mut self, topic: TopicName, count: u32) -> Result<FetchResult, Error> {
        // Discover head_offset with a minimal fetch at offset 0, max_count 1.
        let probe = self.fetch(topic.clone(), TopicOffset::ZERO, 1).await?;
        let head = probe.head_offset.0;
        if head == 0 {
            // Empty topic — return the probe result directly (empty messages).
            return Ok(probe);
        }
        let start = head.saturating_sub(count as u64);
        self.fetch(topic, TopicOffset(start), count).await
    }

    /// Fetch messages starting at a timestamp.
    pub async fn fetch_from(
        &mut self,
        topic: TopicName,
        timestamp_millis: u64,
        max_count: u32,
    ) -> Result<FetchResult, Error> {
        self.send_frame(Frame::FetchFrom {
            topic,
            timestamp_millis,
            max_count,
        })
        .await?;

        self.recv_fetch_response().await
    }

    /// Receive and decode a `FetchResponse` (shared by `fetch` and `fetch_from`).
    async fn recv_fetch_response(&mut self) -> Result<FetchResult, Error> {
        // The server may send an OffsetNotAvailable warning before the response.
        let mut warning: Option<String> = None;
        loop {
            let frame = self.recv_frame().await?;

            match frame {
                Frame::FetchResponse {
                    topic,
                    next_offset,
                    head_offset,
                    messages,
                } => {
                    return Ok(FetchResult {
                        topic,
                        next_offset,
                        head_offset,
                        messages,
                        warning,
                    });
                }
                Frame::Error { code, message } => {
                    if !code.is_fatal() {
                        // Non-fatal errors (e.g. OffsetNotAvailable) are warnings.
                        warning = Some(message);
                        continue;
                    }
                    return Err(Error::Config(format!("fetch error [{code:?}]: {message}")));
                }
                other => {
                    return Err(Error::Config(format!(
                        "unexpected frame during fetch: {other:?}"
                    )));
                }
            }
        }
    }

    /// Gracefully close the connection.
    ///
    /// Stops the keepalive task, closes the writer channel (letting the writer
    /// loop flush queued frames), then waits up to 5s for the writer to finish.
    /// If the writer doesn't finish in time, Drop aborts it.
    pub async fn close(mut self) -> Result<(), Error> {
        // Stop keepalive first so it doesn't queue more Pings.
        if let Some(h) = self.keepalive_handle.take() {
            h.abort();
        }
        // Drop the sender — writer_loop will drain remaining frames then exit.
        self.frame_tx.take();
        // Wait for the writer to flush and close the TCP stream.
        let _ = tokio::time::timeout(Duration::from_secs(5), &mut self.writer_handle).await;
        Ok(())
        // Drop impl aborts writer_handle if it's still running after timeout.
    }
}

impl Drop for FafbClient {
    fn drop(&mut self) {
        // Abort background tasks to prevent orphaned tokio tasks when the
        // client is dropped without calling close().
        if let Some(h) = self.keepalive_handle.take() {
            h.abort();
        }
        self.writer_handle.abort();
    }
}

// ---------------------------------------------------------------------------
// Background tasks
// ---------------------------------------------------------------------------

/// Drains the mpsc channel and writes frames to the sink. Exits when the
/// channel is closed (all senders dropped) or when a write error occurs.
async fn writer_loop(
    mut writer: SplitSink<Framed<ClientStream, MessageCodec>, Frame>,
    mut rx: mpsc::Receiver<Frame>,
) {
    while let Some(frame) = rx.recv().await {
        if writer.send(frame).await.is_err() {
            break;
        }
    }
    let _ = writer.close().await;
}

/// Sends `Ping` when idle for `interval`. Each activity notification resets
/// the idle timer. Exits when the writer channel is closed.
async fn keepalive_loop(tx: mpsc::Sender<Frame>, activity: Arc<Notify>, interval: Duration) {
    loop {
        tokio::select! {
            () = tokio::time::sleep(interval) => {
                if tx.send(Frame::Ping).await.is_err() {
                    break;
                }
            }
            () = activity.notified() => {
                continue;
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Result types for high-level operations
// ---------------------------------------------------------------------------

/// Result of a successful publish operation.
#[derive(Debug)]
pub struct PublishResult {
    pub message_id: MessageId,
}

/// Result of a successful fetch operation.
#[derive(Debug)]
pub struct FetchResult {
    pub topic: TopicName,
    pub next_offset: TopicOffset,
    pub head_offset: TopicOffset,
    pub messages: Vec<FetchedMessage>,
    /// Non-fatal warning (e.g. "offset not available, clamped to oldest").
    pub warning: Option<String>,
}

// ---------------------------------------------------------------------------
// TLS helper
// ---------------------------------------------------------------------------

/// Wrap a TCP stream in TLS using the client config.
async fn wrap_tls(
    tcp: TcpStream,
    config: &ClientConfig,
) -> Result<tokio_rustls::client::TlsStream<TcpStream>, Error> {
    use rustls::pki_types::ServerName;

    let tls_config = build_client_tls_config(&config.tls)?;
    let connector = tokio_rustls::TlsConnector::from(Arc::new(tls_config));

    // Derive server name for SNI from config or from the server address.
    let sni = config.tls.server_name.clone().unwrap_or_else(|| {
        config
            .server
            .split(':')
            .next()
            .unwrap_or("localhost")
            .to_owned()
    });

    let server_name = ServerName::try_from(sni.clone())
        .map_err(|_| Error::Config(format!("invalid TLS server name: {sni}")))?;

    tokio::time::timeout(config.timeout, connector.connect(server_name, tcp))
        .await
        .map_err(|_| Error::Config("TLS handshake timed out".into()))?
        .map_err(|e| Error::Config(format!("TLS handshake failed: {e}")))
}

/// Build a rustls `ClientConfig` from the client TLS settings.
fn build_client_tls_config(tls: &ClientTlsConfig) -> Result<rustls::ClientConfig, Error> {
    let provider = Arc::new(rustls::crypto::ring::default_provider());

    // Root certificates: custom CA or system defaults.
    let root_store = if let Some(ref ca_path) = tls.ca_cert_file {
        let pem_data = std::fs::read(ca_path)
            .map_err(|e| Error::Config(format!("failed to read CA cert {ca_path}: {e}")))?;
        let mut store = rustls::RootCertStore::empty();
        let certs = rustls_pemfile::certs(&mut &pem_data[..])
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| Error::Config(format!("failed to parse CA cert: {e}")))?;
        for cert in certs {
            store
                .add(cert)
                .map_err(|e| Error::Config(format!("failed to add CA cert: {e}")))?;
        }
        store
    } else {
        // No custom CA — use an empty store with dangerous verifier disabled.
        // In production, you'd want webpki-roots or native-certs here.
        // For now, custom CA is required when TLS is enabled.
        return Err(Error::Config(
            "TLS enabled but no CA certificate specified (--tls-ca)".into(),
        ));
    };

    // Client certificate for mTLS (optional).
    if let (Some(cert_path), Some(key_path)) = (&tls.client_cert_file, &tls.client_key_file) {
        let cert_pem = std::fs::read(cert_path)
            .map_err(|e| Error::Config(format!("failed to read client cert {cert_path}: {e}")))?;
        let key_pem = std::fs::read(key_path)
            .map_err(|e| Error::Config(format!("failed to read client key {key_path}: {e}")))?;

        let certs: Vec<_> = rustls_pemfile::certs(&mut &cert_pem[..])
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| Error::Config(format!("failed to parse client cert: {e}")))?;
        let key = rustls_pemfile::private_key(&mut &key_pem[..])
            .map_err(|e| Error::Config(format!("failed to parse client key: {e}")))?
            .ok_or_else(|| Error::Config("no private key found in client key file".into()))?;

        rustls::ClientConfig::builder_with_provider(provider)
            .with_protocol_versions(&[&rustls::version::TLS13, &rustls::version::TLS12])
            .map_err(|e| Error::Config(format!("TLS version config error: {e}")))?
            .with_root_certificates(root_store)
            .with_client_auth_cert(certs, key)
            .map_err(|e| Error::Config(format!("client cert config error: {e}")))
    } else {
        rustls::ClientConfig::builder_with_provider(provider)
            .with_protocol_versions(&[&rustls::version::TLS13, &rustls::version::TLS12])
            .map_err(|e| Error::Config(format!("TLS version config error: {e}")))?
            .with_root_certificates(root_store)
            .with_no_client_auth()
            .pipe(Ok)
    }
}

/// Helper for `Ok(value)` in method chains.
trait PipeOk: Sized {
    fn pipe(self, f: fn(Self) -> Result<Self, Error>) -> Result<Self, Error> {
        f(self)
    }
}

impl PipeOk for rustls::ClientConfig {}

// ---------------------------------------------------------------------------
// Verbose frame logging helpers
// ---------------------------------------------------------------------------

/// Log a sent frame to stderr for `--verbose` debugging.
fn log_frame_send(frame: &Frame) {
    eprintln!(">>> {}", format_frame_summary(frame));
}

/// Log a received frame to stderr for `--verbose` debugging.
fn log_frame_recv(frame: &Frame) {
    eprintln!("<<< {}", format_frame_summary(frame));
}

/// Format a frame as a concise summary for verbose output.
///
/// Uses `Debug` with truncation for most frames. Special-cases `Connect`
/// to redact the API key. Automatically covers new frame variants without
/// maintenance.
fn format_frame_summary(frame: &Frame) -> String {
    // Redact API key from Connect frames.
    if let Frame::Connect {
        version,
        namespace,
        capabilities,
        ..
    } = frame
    {
        return format!(
            "Connect {{ version: {version}, namespace: {:?}, api_key: \"****\", capabilities: 0x{capabilities:04X} }}",
            namespace.0
        );
    }

    let debug = format!("{frame:?}");

    // Truncate long output (large payloads, many messages in FetchResponse).
    const MAX_LEN: usize = 256;
    if debug.len() > MAX_LEN {
        format!("{}...", &debug[..MAX_LEN])
    } else {
        debug
    }
}

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

    #[test]
    fn client_config_new() {
        let cfg = ClientConfig::new("localhost:9876", "myns", "sk_test");
        assert_eq!(cfg.server, "localhost:9876");
        assert_eq!(cfg.namespace, NamespaceId("myns".into()));
        assert_eq!(cfg.api_key, ApiKey("sk_test".into()));
        assert_eq!(cfg.timeout, Duration::from_secs(30));
        assert_eq!(cfg.max_frame_size, DEFAULT_MAX_FRAME_SIZE);
        assert!(!cfg.tls.enabled);
    }

    #[test]
    fn client_stream_enum_is_unpin() {
        // Verify ClientStream is Unpin (required for Framed).
        fn assert_unpin<T: Unpin>() {}
        assert_unpin::<ClientStream>();
    }

    #[test]
    fn sni_derived_from_server_address() {
        let config = ClientConfig {
            server: "myhost.example.com:9876".into(),
            namespace: NamespaceId("ns".into()),
            api_key: ApiKey("key".into()),
            timeout: Duration::from_secs(5),
            max_frame_size: DEFAULT_MAX_FRAME_SIZE,
            tls: ClientTlsConfig {
                enabled: true,
                server_name: None,
                ..Default::default()
            },
            verbose: false,
            ping_interval: Duration::from_secs(10),
        };
        // The SNI should be derived as "myhost.example.com"
        let sni = config.tls.server_name.clone().unwrap_or_else(|| {
            config
                .server
                .split(':')
                .next()
                .unwrap_or("localhost")
                .to_owned()
        });
        assert_eq!(sni, "myhost.example.com");
    }

    #[test]
    fn sni_explicit_override() {
        let config = ClientConfig {
            server: "myhost:9876".into(),
            namespace: NamespaceId("ns".into()),
            api_key: ApiKey("key".into()),
            timeout: Duration::from_secs(5),
            max_frame_size: DEFAULT_MAX_FRAME_SIZE,
            tls: ClientTlsConfig {
                enabled: true,
                server_name: Some("custom.sni.name".into()),
                ..Default::default()
            },
            verbose: false,
            ping_interval: Duration::from_secs(10),
        };
        let sni = config.tls.server_name.clone().unwrap_or_else(|| {
            config
                .server
                .split(':')
                .next()
                .unwrap_or("localhost")
                .to_owned()
        });
        assert_eq!(sni, "custom.sni.name");
    }

    // Verbose frame logging tests (Phase 23.6)

    #[test]
    fn format_frame_summary_connect_redacts_api_key() {
        let frame = Frame::Connect {
            version: 1,
            namespace: NamespaceId("test_ns".into()),
            api_key: ApiKey("super_secret_key".into()),
            capabilities: 0x7F,
        };
        let summary = format_frame_summary(&frame);
        assert!(summary.contains("Connect"));
        assert!(summary.contains("test_ns"));
        assert!(summary.contains("0x007F"));
        assert!(summary.contains("****"));
        // Must NOT leak the API key.
        assert!(!summary.contains("super_secret_key"));
    }

    #[test]
    fn format_frame_summary_uses_debug_for_other_frames() {
        let frame = Frame::Publish {
            topic: TopicName("orders".into()),
            headers: vec![],
            ttl_ms: 5000,
            payload: Bytes::from("hello world"),
        };
        let summary = format_frame_summary(&frame);
        assert!(summary.contains("Publish"));
        assert!(summary.contains("orders"));
        assert!(summary.contains("hello world"));
    }

    #[test]
    fn format_frame_summary_truncates_long_output() {
        let payload = "x".repeat(500);
        let frame = Frame::Publish {
            topic: TopicName("big".into()),
            headers: vec![],
            ttl_ms: 0,
            payload: Bytes::from(payload),
        };
        let summary = format_frame_summary(&frame);
        assert!(summary.ends_with("..."));
        assert!(summary.len() <= 260); // 256 + "..."
    }

    #[test]
    fn format_frame_summary_error() {
        let frame = Frame::Error {
            code: crate::codec::frame::ErrorCode::InternalError,
            message: "something broke".into(),
        };
        let summary = format_frame_summary(&frame);
        assert!(summary.contains("Error"));
        assert!(summary.contains("something broke"));
    }
}
