//! Async TLS 1.2 + SRP transport layered on top of any `AsyncRead + AsyncWrite`.

use std::collections::VecDeque;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};

use gsocket_tls_srp::cipher::{decrypt_record, encrypt_record, RECORD_IV_LEN};
use gsocket_tls_srp::handshake::{
    ClientHandshake, ClientState, HandshakeError, ServerHandshake, ServerState,
};
use gsocket_tls_srp::messages::HANDSHAKE_HEADER_LEN;
use gsocket_tls_srp::prf::{DirectionKeys, KeyBlock};
use gsocket_tls_srp::record::{
    ContentType, HeaderError, RecordHeader, MAX_PLAINTEXT, RECORD_HEADER_LEN, TLS_VERSION_1_2,
};
use rand::RngCore;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};

const MAX_RECORD_WIRE_LEN: usize = RECORD_HEADER_LEN + RECORD_IV_LEN + MAX_PLAINTEXT + 32 + 256;

pub struct TlsStream<S> {
    inner: S,
    write_keys: DirectionKeys,
    read_keys: DirectionKeys,
    write_seq: u64,
    read_seq: u64,
    read_buf: Vec<u8>,
    plaintext: VecDeque<u8>,
    pending_write: Vec<u8>,
    pending_write_offset: usize,
    seen_eof: bool,
}

impl<S> TlsStream<S> {
    pub fn get_ref(&self) -> &S {
        &self.inner
    }
    pub fn into_inner(self) -> S {
        self.inner
    }
}

fn io_other<E: std::fmt::Display>(e: E) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidData, e.to_string())
}
fn handshake_io_err(e: HandshakeError) -> io::Error {
    io_other(e)
}
fn header_io_err(e: HeaderError) -> io::Error {
    io_other(e)
}

async fn write_plaintext_record<S: AsyncWriteExt + Unpin>(
    stream: &mut S,
    content_type: ContentType,
    body: &[u8],
) -> io::Result<()> {
    let hdr = RecordHeader {
        content_type,
        version: TLS_VERSION_1_2,
        length: u16::try_from(body.len())
            .map_err(|_| io::Error::other("record body exceeds u16"))?,
    };
    stream.write_all(&hdr.encode()).await?;
    stream.write_all(body).await?;
    Ok(())
}

async fn read_one_record<S: AsyncReadExt + Unpin>(
    stream: &mut S,
    buf: &mut Vec<u8>,
) -> io::Result<RecordHeader> {
    while buf.len() < RECORD_HEADER_LEN {
        let mut tmp = [0u8; 4096];
        let n = stream.read(&mut tmp).await?;
        if n == 0 {
            return Err(io::ErrorKind::UnexpectedEof.into());
        }
        buf.extend_from_slice(&tmp[..n]);
    }
    let hdr = RecordHeader::parse(&buf[..RECORD_HEADER_LEN]).map_err(header_io_err)?;
    let frag_end = RECORD_HEADER_LEN + hdr.length as usize;
    while buf.len() < frag_end {
        let mut tmp = [0u8; 4096];
        let n = stream.read(&mut tmp).await?;
        if n == 0 {
            return Err(io::ErrorKind::UnexpectedEof.into());
        }
        buf.extend_from_slice(&tmp[..n]);
    }
    Ok(hdr)
}

fn next_handshake_message_len(buf: &[u8]) -> Option<usize> {
    if buf.len() < HANDSHAKE_HEADER_LEN {
        return None;
    }
    let body_len = ((buf[1] as usize) << 16) | ((buf[2] as usize) << 8) | (buf[3] as usize);
    Some(HANDSHAKE_HEADER_LEN + body_len)
}

/// Drive the SRP-TLS client handshake to completion over `stream`,
/// returning an established [`TlsStream`].
///
/// `srp_identity` is the SRP username (the GSRN address) and
/// `srp_password_wrapped` is the SRP secret in the on-wire form that
/// the C reference uses (the `gsocket-proto` crate produces this via
/// [`gsocket_proto::address::SrpPassword`]).
pub async fn client_handshake<S: AsyncRead + AsyncWrite + Unpin>(
    stream: S,
    srp_identity: Vec<u8>,
    srp_password_wrapped: Vec<u8>,
) -> io::Result<TlsStream<S>> {
    client_handshake_with_prefix(stream, Vec::new(), srp_identity, srp_password_wrapped).await
}

/// Variant of [`client_handshake`] that prepends `initial_bytes` to the
/// stream's read side before any record is parsed.
///
/// Use this when you've already consumed bytes from the underlying
/// transport (e.g. while completing a GSRN handshake) and need to feed
/// them into the TLS handshake's read path.
pub async fn client_handshake_with_prefix<S: AsyncRead + AsyncWrite + Unpin>(
    mut stream: S,
    initial_bytes: Vec<u8>,
    srp_identity: Vec<u8>,
    srp_password_wrapped: Vec<u8>,
) -> io::Result<TlsStream<S>> {
    let mut hs = ClientHandshake::new(srp_identity, srp_password_wrapped);

    let ch = hs.client_hello_bytes().map_err(handshake_io_err)?;
    write_plaintext_record(&mut stream, ContentType::Handshake, &ch).await?;

    let mut rb: Vec<u8> = initial_bytes;
    let mut hs_acc: Vec<u8> = Vec::new();
    while hs.state() != ClientState::SendingClientKeyExchange {
        let hdr = read_one_record(&mut stream, &mut rb).await?;
        if hdr.content_type != ContentType::Handshake {
            return Err(io::Error::other(format!(
                "unexpected record type during handshake: {:?}",
                hdr.content_type
            )));
        }
        let frag_end = RECORD_HEADER_LEN + hdr.length as usize;
        hs_acc.extend_from_slice(&rb[RECORD_HEADER_LEN..frag_end]);
        rb.drain(..frag_end);

        while let Some(msg_len) = next_handshake_message_len(&hs_acc) {
            if hs_acc.len() < msg_len {
                break;
            }
            let msg = hs_acc[..msg_len].to_vec();
            hs.process_handshake(&msg).map_err(handshake_io_err)?;
            hs_acc.drain(..msg_len);
            if hs.state() == ClientState::SendingClientKeyExchange {
                break;
            }
        }
    }

    let cke = hs.client_key_exchange_bytes().map_err(handshake_io_err)?;
    write_plaintext_record(&mut stream, ContentType::Handshake, &cke).await?;
    write_plaintext_record(&mut stream, ContentType::ChangeCipherSpec, &[0x01]).await?;

    let kb: KeyBlock = hs.keys().ok_or_else(|| io::Error::other("no keys"))?.clone();
    let write_keys = kb.client.clone();
    let read_keys = kb.server.clone();
    let cf_plain = hs.client_finished_bytes().map_err(handshake_io_err)?;
    let mut iv = [0u8; RECORD_IV_LEN];
    rand::rngs::OsRng.fill_bytes(&mut iv);
    let cf_record = encrypt_and_frame(&write_keys, 0, ContentType::Handshake, &cf_plain, &iv);
    stream.write_all(&cf_record).await?;
    stream.flush().await?;
    let write_seq: u64 = 1;

    let hdr = read_one_record(&mut stream, &mut rb).await?;
    if hdr.content_type != ContentType::ChangeCipherSpec {
        return Err(io::Error::other("expected ChangeCipherSpec from server"));
    }
    let frag_end = RECORD_HEADER_LEN + hdr.length as usize;
    if rb[RECORD_HEADER_LEN..frag_end] != [0x01] {
        return Err(io::Error::other("malformed ChangeCipherSpec"));
    }
    rb.drain(..frag_end);
    hs.note_change_cipher_spec_received()
        .map_err(handshake_io_err)?;

    let hdr = read_one_record(&mut stream, &mut rb).await?;
    if hdr.content_type != ContentType::Handshake {
        return Err(io::Error::other("expected encrypted Finished record"));
    }
    let frag_end = RECORD_HEADER_LEN + hdr.length as usize;
    let server_finished = decrypt_record(
        &read_keys.enc_key,
        &read_keys.mac_key,
        0,
        ContentType::Handshake,
        hdr.version,
        &rb[RECORD_HEADER_LEN..frag_end],
    )
    .map_err(io_other)?;
    rb.drain(..frag_end);
    hs.process_handshake(&server_finished)
        .map_err(handshake_io_err)?;

    if hs.state() != ClientState::Established {
        return Err(io::Error::other("client handshake did not reach Established"));
    }

    Ok(TlsStream {
        inner: stream,
        write_keys,
        read_keys,
        write_seq,
        read_seq: 1,
        read_buf: rb,
        plaintext: VecDeque::new(),
        pending_write: Vec::new(),
        pending_write_offset: 0,
        seen_eof: false,
    })
}

/// Drive the SRP-TLS server handshake to completion over `stream`,
/// returning an established [`TlsStream`].
///
/// `srp_password_wrapped` is the SRP verifier secret in the on-wire form
/// used by the C reference; see [`gsocket_proto::address::SrpPassword`].
pub async fn server_handshake<S: AsyncRead + AsyncWrite + Unpin>(
    stream: S,
    srp_password_wrapped: Vec<u8>,
) -> io::Result<TlsStream<S>> {
    server_handshake_with_prefix(stream, Vec::new(), srp_password_wrapped).await
}

/// Variant of [`server_handshake`] that prepends `initial_bytes` to the
/// stream's read side before any record is parsed. Useful when GSRN
/// handshake bytes were consumed eagerly and must be folded back into
/// the TLS read stream.
pub async fn server_handshake_with_prefix<S: AsyncRead + AsyncWrite + Unpin>(
    mut stream: S,
    initial_bytes: Vec<u8>,
    srp_password_wrapped: Vec<u8>,
) -> io::Result<TlsStream<S>> {
    let mut hs = ServerHandshake::from_password(&srp_password_wrapped);

    let mut rb: Vec<u8> = initial_bytes;
    let mut hs_acc: Vec<u8> = Vec::new();

    while hs.state() == ServerState::AwaitingClientHello {
        let hdr = read_one_record(&mut stream, &mut rb).await?;
        if hdr.content_type != ContentType::Handshake {
            return Err(io::Error::other(format!(
                "expected Handshake record, got {:?}",
                hdr.content_type
            )));
        }
        let frag_end = RECORD_HEADER_LEN + hdr.length as usize;
        hs_acc.extend_from_slice(&rb[RECORD_HEADER_LEN..frag_end]);
        rb.drain(..frag_end);

        if let Some(msg_len) = next_handshake_message_len(&hs_acc) {
            if hs_acc.len() >= msg_len {
                hs.process_client_hello(&hs_acc[..msg_len])
                    .map_err(handshake_io_err)?;
                hs_acc.drain(..msg_len);
            }
        }
    }

    let flight = hs.flight1_bytes().map_err(handshake_io_err)?;
    write_plaintext_record(&mut stream, ContentType::Handshake, &flight).await?;
    stream.flush().await?;

    while hs.state() == ServerState::AwaitingClientKeyExchange {
        let hdr = read_one_record(&mut stream, &mut rb).await?;
        if hdr.content_type != ContentType::Handshake {
            return Err(io::Error::other(format!(
                "expected Handshake (ClientKeyExchange), got {:?} (len={}, first bytes: {})",
                hdr.content_type,
                hdr.length,
                rb.iter().take(32).map(|b| format!("{:02x}", b)).collect::<String>()
            )));
        }
        let frag_end = RECORD_HEADER_LEN + hdr.length as usize;
        hs_acc.extend_from_slice(&rb[RECORD_HEADER_LEN..frag_end]);
        rb.drain(..frag_end);

        if let Some(msg_len) = next_handshake_message_len(&hs_acc) {
            if hs_acc.len() >= msg_len {
                hs.process_client_key_exchange(&hs_acc[..msg_len])
                    .map_err(handshake_io_err)?;
                hs_acc.drain(..msg_len);
            }
        }
    }

    let hdr = read_one_record(&mut stream, &mut rb).await?;
    if hdr.content_type != ContentType::ChangeCipherSpec {
        return Err(io::Error::other("expected ChangeCipherSpec from client"));
    }
    let frag_end = RECORD_HEADER_LEN + hdr.length as usize;
    rb.drain(..frag_end);
    hs.note_change_cipher_spec_received()
        .map_err(handshake_io_err)?;

    let kb = hs.keys().ok_or_else(|| io::Error::other("no keys"))?.clone();
    let write_keys = kb.server.clone();
    let read_keys = kb.client.clone();

    let hdr = read_one_record(&mut stream, &mut rb).await?;
    if hdr.content_type != ContentType::Handshake {
        return Err(io::Error::other("expected encrypted client Finished"));
    }
    let frag_end = RECORD_HEADER_LEN + hdr.length as usize;
    let cf_plain = decrypt_record(
        &read_keys.enc_key,
        &read_keys.mac_key,
        0,
        ContentType::Handshake,
        hdr.version,
        &rb[RECORD_HEADER_LEN..frag_end],
    )
    .map_err(io_other)?;
    rb.drain(..frag_end);
    hs.process_client_finished(&cf_plain)
        .map_err(handshake_io_err)?;

    write_plaintext_record(&mut stream, ContentType::ChangeCipherSpec, &[0x01]).await?;

    let sf_plain = hs.server_finished_bytes().map_err(handshake_io_err)?;
    let mut iv = [0u8; RECORD_IV_LEN];
    rand::rngs::OsRng.fill_bytes(&mut iv);
    let sf_record = encrypt_and_frame(&write_keys, 0, ContentType::Handshake, &sf_plain, &iv);
    stream.write_all(&sf_record).await?;
    stream.flush().await?;

    if hs.state() != ServerState::Established {
        return Err(io::Error::other("server handshake did not reach Established"));
    }

    Ok(TlsStream {
        inner: stream,
        write_keys,
        read_keys,
        write_seq: 1,
        read_seq: 1,
        read_buf: rb,
        plaintext: VecDeque::new(),
        pending_write: Vec::new(),
        pending_write_offset: 0,
        seen_eof: false,
    })
}

fn encrypt_and_frame(
    keys: &DirectionKeys,
    seq: u64,
    content_type: ContentType,
    plaintext: &[u8],
    iv: &[u8; RECORD_IV_LEN],
) -> Vec<u8> {
    let fragment = encrypt_record(
        &keys.enc_key,
        &keys.mac_key,
        seq,
        content_type,
        TLS_VERSION_1_2,
        plaintext,
        iv,
    );
    let hdr = RecordHeader {
        content_type,
        version: TLS_VERSION_1_2,
        length: u16::try_from(fragment.len()).expect("fragment fits in u16"),
    };
    let mut out = Vec::with_capacity(RECORD_HEADER_LEN + fragment.len());
    out.extend_from_slice(&hdr.encode());
    out.extend_from_slice(&fragment);
    out
}

impl<S: AsyncRead + Unpin> AsyncRead for TlsStream<S> {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        let this = self.get_mut();
        loop {
            if !this.plaintext.is_empty() {
                let want = buf.remaining();
                if want == 0 {
                    return Poll::Ready(Ok(()));
                }
                let take = want.min(this.plaintext.len());
                let (front, back) = this.plaintext.as_slices();
                let from_front = take.min(front.len());
                buf.put_slice(&front[..from_front]);
                let still = take - from_front;
                if still > 0 {
                    buf.put_slice(&back[..still]);
                }
                this.plaintext.drain(..take);
                return Poll::Ready(Ok(()));
            }
            if try_decrypt_one_record(this)? > 0 {
                continue;
            }
            if this.seen_eof {
                return Poll::Ready(Ok(()));
            }
            let mut tmp = [0u8; 8192];
            let mut rb = ReadBuf::new(&mut tmp);
            match Pin::new(&mut this.inner).poll_read(cx, &mut rb) {
                Poll::Ready(Ok(())) => {
                    if rb.filled().is_empty() {
                        this.seen_eof = true;
                        if this.read_buf.is_empty() {
                            return Poll::Ready(Ok(()));
                        }
                        return Poll::Ready(Err(io::ErrorKind::UnexpectedEof.into()));
                    }
                    this.read_buf.extend_from_slice(rb.filled());
                }
                Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                Poll::Pending => return Poll::Pending,
            }
        }
    }
}

fn try_decrypt_one_record<S>(state: &mut TlsStream<S>) -> io::Result<usize> {
    if state.read_buf.len() < RECORD_HEADER_LEN {
        return Ok(0);
    }
    let hdr = RecordHeader::parse(&state.read_buf[..RECORD_HEADER_LEN]).map_err(header_io_err)?;
    let frag_end = RECORD_HEADER_LEN + hdr.length as usize;
    if frag_end > MAX_RECORD_WIRE_LEN {
        return Err(io::Error::other("oversized record"));
    }
    if state.read_buf.len() < frag_end {
        return Ok(0);
    }
    match hdr.content_type {
        ContentType::ApplicationData => {
            let pt = decrypt_record(
                &state.read_keys.enc_key,
                &state.read_keys.mac_key,
                state.read_seq,
                ContentType::ApplicationData,
                hdr.version,
                &state.read_buf[RECORD_HEADER_LEN..frag_end],
            )
            .map_err(io_other)?;
            state.read_seq = state
                .read_seq
                .checked_add(1)
                .ok_or_else(|| io::Error::other("read sequence overflow"))?;
            state.plaintext.extend(pt);
            state.read_buf.drain(..frag_end);
            Ok(frag_end)
        }
        ContentType::Alert => {
            state.seen_eof = true;
            state.read_buf.drain(..frag_end);
            Ok(frag_end)
        }
        other => Err(io::Error::other(format!(
            "unexpected post-handshake record type: {other:?}"
        ))),
    }
}

impl<S: AsyncWrite + Unpin> AsyncWrite for TlsStream<S> {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        let this = self.get_mut();
        while this.pending_write_offset < this.pending_write.len() {
            let remaining = &this.pending_write[this.pending_write_offset..];
            match Pin::new(&mut this.inner).poll_write(cx, remaining) {
                Poll::Ready(Ok(0)) => return Poll::Ready(Err(io::ErrorKind::WriteZero.into())),
                Poll::Ready(Ok(n)) => this.pending_write_offset += n,
                Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                Poll::Pending => return Poll::Pending,
            }
        }
        this.pending_write.clear();
        this.pending_write_offset = 0;
        if buf.is_empty() {
            return Poll::Ready(Ok(0));
        }
        let chunk_len = buf.len().min(MAX_PLAINTEXT);
        let mut iv = [0u8; RECORD_IV_LEN];
        rand::rngs::OsRng.fill_bytes(&mut iv);
        let record = encrypt_and_frame(
            &this.write_keys,
            this.write_seq,
            ContentType::ApplicationData,
            &buf[..chunk_len],
            &iv,
        );
        this.write_seq = this
            .write_seq
            .checked_add(1)
            .ok_or_else(|| io::Error::other("write sequence overflow"))?;
        this.pending_write = record;
        while this.pending_write_offset < this.pending_write.len() {
            let remaining = &this.pending_write[this.pending_write_offset..];
            match Pin::new(&mut this.inner).poll_write(cx, remaining) {
                Poll::Ready(Ok(0)) => break,
                Poll::Ready(Ok(n)) => this.pending_write_offset += n,
                Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                Poll::Pending => break,
            }
        }
        Poll::Ready(Ok(chunk_len))
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        let this = self.get_mut();
        while this.pending_write_offset < this.pending_write.len() {
            let remaining = &this.pending_write[this.pending_write_offset..];
            match Pin::new(&mut this.inner).poll_write(cx, remaining) {
                Poll::Ready(Ok(0)) => return Poll::Ready(Err(io::ErrorKind::WriteZero.into())),
                Poll::Ready(Ok(n)) => this.pending_write_offset += n,
                Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                Poll::Pending => return Poll::Pending,
            }
        }
        this.pending_write.clear();
        this.pending_write_offset = 0;
        Pin::new(&mut this.inner).poll_flush(cx)
    }

    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        let this = self.get_mut();
        while this.pending_write_offset < this.pending_write.len() {
            let remaining = &this.pending_write[this.pending_write_offset..];
            match Pin::new(&mut this.inner).poll_write(cx, remaining) {
                Poll::Ready(Ok(0)) => break,
                Poll::Ready(Ok(n)) => this.pending_write_offset += n,
                Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
                Poll::Pending => return Poll::Pending,
            }
        }
        Pin::new(&mut this.inner).poll_shutdown(cx)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::net::{TcpListener, TcpStream};

    #[tokio::test]
    async fn end_to_end_tls_over_tcp_echo() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let password = b"Blah.deadbeefdeadbeefdeadbeefdeadb.blubb-SRPSEC".to_vec();

        let server_password = password.clone();
        let server = tokio::spawn(async move {
            let (sock, _) = listener.accept().await.unwrap();
            let mut s = server_handshake(sock, server_password).await.unwrap();
            for _ in 0..3 {
                let mut buf = vec![0u8; 5];
                s.read_exact(&mut buf).await.unwrap();
                s.write_all(&buf).await.unwrap();
                s.flush().await.unwrap();
            }
        });

        let sock = TcpStream::connect(addr).await.unwrap();
        let mut c = client_handshake(sock, b"user".to_vec(), password).await.unwrap();
        for msg in [b"hello", b"world", b"again"] {
            c.write_all(msg).await.unwrap();
            c.flush().await.unwrap();
            let mut buf = vec![0u8; 5];
            c.read_exact(&mut buf).await.unwrap();
            assert_eq!(&buf, msg);
        }

        server.await.unwrap();
    }

    #[tokio::test]
    async fn end_to_end_large_payload_spans_multiple_records() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let password = b"Blah.aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.blubb-SRPSEC".to_vec();

        let payload: Vec<u8> = (0..(MAX_PLAINTEXT * 3 + 7))
            .map(|i| (i & 0xff) as u8)
            .collect();
        let payload_for_server = payload.clone();

        let server_password = password.clone();
        let server = tokio::spawn(async move {
            let (sock, _) = listener.accept().await.unwrap();
            let mut s = server_handshake(sock, server_password).await.unwrap();
            let mut buf = vec![0u8; payload_for_server.len()];
            s.read_exact(&mut buf).await.unwrap();
            assert_eq!(buf, payload_for_server);
            s.write_all(&buf).await.unwrap();
            s.flush().await.unwrap();
        });

        let sock = TcpStream::connect(addr).await.unwrap();
        let mut c = client_handshake(sock, b"user".to_vec(), password).await.unwrap();
        c.write_all(&payload).await.unwrap();
        c.flush().await.unwrap();
        let mut buf = vec![0u8; payload.len()];
        c.read_exact(&mut buf).await.unwrap();
        assert_eq!(buf, payload);

        server.await.unwrap();
    }

    #[tokio::test]
    async fn wrong_password_fails_handshake() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let server_password = b"Blah.aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.blubb-SRPSEC".to_vec();
        let client_password = b"Blah.bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb.blubb-SRPSEC".to_vec();

        let server = tokio::spawn(async move {
            let (sock, _) = listener.accept().await.unwrap();
            server_handshake(sock, server_password).await
        });

        let sock = TcpStream::connect(addr).await.unwrap();
        let res = client_handshake(sock, b"user".to_vec(), client_password).await;
        assert!(res.is_err());
        let res = server.await.unwrap();
        assert!(res.is_err());
    }
}
