scylla-proxy 0.0.8

use std::collections::HashMap;

use bytes::{Buf, BufMut, Bytes, BytesMut};
use scylla_cql::frame::flag;
use scylla_cql::frame::frame_errors::FrameHeaderParseError;
use scylla_cql::frame::protocol_features::ProtocolFeatures;
pub use scylla_cql::frame::request::RequestOpcode;
use scylla_cql::frame::request::{RequestDeserializationError, RequestV2};
pub use scylla_cql::frame::response::ResponseOpcode;
use scylla_cql::frame::response::error::DbError;
use scylla_cql::frame::types;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

use tracing::warn;

use crate::errors::ReadFrameError;
use crate::proxy::CompressionReader;

const HEADER_SIZE: usize = 9;

// Parts of the frame header which are not determined by the request/response type.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub struct FrameParams {
    pub version: u8,
    pub flags: u8,
    pub stream: i16,
}

impl FrameParams {
    pub const fn for_request(&self) -> FrameParams {
        Self {
            version: self.version & 0x7F,
            ..*self
        }
    }
    pub const fn for_response(&self) -> FrameParams {
        Self {
            version: 0x80 | (self.version & 0x7F),
            ..*self
        }
    }

    /// Tells whether the frame carried a compressed body on the wire.
    ///
    /// Note that frames handed out by the proxy always expose a *decompressed*
    /// body, so this flag (together with [`RequestFrame::wire_body_len`]) is the
    /// only way to tell that compression was in play.
    pub const fn is_compressed(&self) -> bool {
        self.flags & flag::COMPRESSION != 0
    }
}

#[derive(Copy, Clone, Debug)]
pub(crate) enum FrameType {
    Request,
    Response,
}

#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub(crate) enum FrameOpcode {
    Request(RequestOpcode),
    Response(ResponseOpcode),
}

#[derive(Clone, Debug, Eq)]
pub struct RequestFrame {
    pub params: FrameParams,
    pub opcode: RequestOpcode,
    /// The frame body, decompressed if the frame arrived compressed.
    pub body: Bytes,
    /// Number of body bytes as they appeared on the wire, i.e. *before* decompression.
    /// For frames that were not compressed this is equal to `body.len()`.
    ///
    /// This is what makes it possible to observe how effective compression is;
    /// [`Self::body`] is always decompressed, so its length says nothing about
    /// how much data actually travelled over the network.
    pub wire_body_len: usize,
}

/// Excludes [`RequestFrame::wire_body_len`], which is an observation about how the frame
/// was transferred rather than a part of its logical content.
impl PartialEq for RequestFrame {
    fn eq(&self, other: &Self) -> bool {
        self.params == other.params && self.opcode == other.opcode && self.body == other.body
    }
}

impl RequestFrame {
    /// Creates a frame whose body did not undergo compression, so that its
    /// [`Self::wire_body_len`] is simply the body length.
    pub fn new(params: FrameParams, opcode: RequestOpcode, body: Bytes) -> Self {
        Self {
            params,
            opcode,
            wire_body_len: body.len(),
            body,
        }
    }

    pub(crate) async fn write(
        &self,
        writer: &mut (impl AsyncWrite + Unpin),
        compression: &CompressionReader,
    ) -> Result<(), tokio::io::Error> {
        write_frame(
            self.params,
            FrameOpcode::Request(self.opcode),
            &self.body,
            writer,
            compression,
        )
        .await
    }

    pub fn deserialize(
        &self,
        features: &ProtocolFeatures,
    ) -> Result<RequestV2<'_>, RequestDeserializationError> {
        RequestV2::deserialize(&mut &self.body[..], self.opcode, features)
    }
}
#[derive(Clone, Debug, Eq)]
pub struct ResponseFrame {
    pub params: FrameParams,
    pub opcode: ResponseOpcode,
    /// The frame body, decompressed if the frame arrived compressed.
    pub body: Bytes,
    /// Number of body bytes as they appeared on the wire, i.e. *before* decompression.
    /// For frames that were not compressed this is equal to `body.len()`.
    ///
    /// See [`RequestFrame::wire_body_len`] for the rationale.
    pub wire_body_len: usize,
}

/// Excludes [`ResponseFrame::wire_body_len`], which is an observation about how the frame
/// was transferred rather than a part of its logical content.
impl PartialEq for ResponseFrame {
    fn eq(&self, other: &Self) -> bool {
        self.params == other.params && self.opcode == other.opcode && self.body == other.body
    }
}

impl ResponseFrame {
    /// Creates a frame whose body did not undergo compression, so that its
    /// [`Self::wire_body_len`] is simply the body length.
    pub fn new(params: FrameParams, opcode: ResponseOpcode, body: Bytes) -> Self {
        Self {
            params,
            opcode,
            wire_body_len: body.len(),
            body,
        }
    }

    /// Creates a response frame that signifies the given DbError type.
    /// Useful for testing server-side error handling in drivers.
    pub fn forged_error(
        request_params: FrameParams,
        error: DbError,
        msg: Option<&str>,
    ) -> Result<Self, std::num::TryFromIntError> {
        let msg = msg.unwrap_or("Proxy-triggered error.");
        let len_bytes = (msg.len() as u16).to_be_bytes(); // string len is a short in CQL protocol
        let code_bytes = error.code(&ProtocolFeatures::default()).to_be_bytes(); // TODO: configurable features
        let body_len = msg.len() + code_bytes.len() + len_bytes.len();
        let mut buf = BytesMut::with_capacity(body_len);

        buf.extend_from_slice(&code_bytes);
        buf.extend_from_slice(&len_bytes);
        buf.extend_from_slice(msg.as_bytes());

        serialize_error_specific_fields(&mut buf, error)?;

        Ok(ResponseFrame::new(
            request_params.for_response(),
            ResponseOpcode::Error,
            buf.freeze(),
        ))
    }

    /// Creates a Supported response frame with given supported options.
    pub fn forged_supported(
        request_params: FrameParams,
        options: &HashMap<String, Vec<String>>,
    ) -> Result<Self, std::num::TryFromIntError> {
        let mut buf = BytesMut::new();
        types::write_string_multimap(options, &mut buf)?;

        Ok(ResponseFrame::new(
            request_params.for_response(),
            ResponseOpcode::Supported,
            buf.freeze(),
        ))
    }

    pub fn forged_ready(request_params: FrameParams) -> Self {
        ResponseFrame::new(
            request_params.for_response(),
            ResponseOpcode::Ready,
            Bytes::new(),
        )
    }

    pub(crate) async fn write(
        &self,
        writer: &mut (impl AsyncWrite + Unpin),
        compression: &CompressionReader,
    ) -> Result<(), tokio::io::Error> {
        write_frame(
            self.params,
            FrameOpcode::Response(self.opcode),
            &self.body,
            writer,
            compression,
        )
        .await
    }
}

fn serialize_error_specific_fields(
    buf: &mut BytesMut,
    error: DbError,
) -> Result<(), std::num::TryFromIntError> {
    match error {
        DbError::Unavailable {
            consistency,
            required,
            alive,
        } => {
            types::write_consistency(consistency, buf);
            types::write_int(required, buf);
            types::write_int(alive, buf);
        }
        DbError::WriteTimeout {
            consistency,
            received,
            required,
            write_type,
        } => {
            types::write_consistency(consistency, buf);
            types::write_int(received, buf);
            types::write_int(required, buf);
            types::write_string(write_type.as_str(), buf)?;
        }
        DbError::ReadTimeout {
            consistency,
            received,
            required,
            data_present,
        } => {
            types::write_consistency(consistency, buf);
            types::write_int(received, buf);
            types::write_int(required, buf);
            buf.put_u8(u8::from(data_present));
        }
        DbError::ReadFailure {
            consistency,
            received,
            required,
            numfailures,
            data_present,
        } => {
            types::write_consistency(consistency, buf);
            types::write_int(received, buf);
            types::write_int(required, buf);
            types::write_int(numfailures, buf);
            buf.put_u8(u8::from(data_present));
        }
        DbError::WriteFailure {
            consistency,
            received,
            required,
            numfailures,
            write_type,
        } => {
            types::write_consistency(consistency, buf);
            types::write_int(received, buf);
            types::write_int(required, buf);
            types::write_int(numfailures, buf);
            types::write_string(write_type.as_str(), buf)?;
        }
        DbError::FunctionFailure {
            keyspace,
            function,
            arg_types,
        } => {
            types::write_string(keyspace.as_str(), buf)?;
            types::write_string(function.as_str(), buf)?;
            types::write_string_list(&arg_types, buf)?;
        }
        DbError::AlreadyExists { keyspace, table } => {
            types::write_string(keyspace.as_str(), buf)?;
            types::write_string(table.as_str(), buf)?;
        }
        DbError::Unprepared { statement_id } => {
            types::write_short_bytes(statement_id.as_ref(), buf)?;
        }
        _ => (),
    }
    Ok(())
}

pub(crate) async fn write_frame(
    params: FrameParams,
    opcode: FrameOpcode,
    body: &[u8],
    writer: &mut (impl AsyncWrite + Unpin),
    compression: &CompressionReader,
) -> Result<(), tokio::io::Error> {
    let compressed_body = compression
        .maybe_compress_body(params.flags, body)
        .map_err(tokio::io::Error::other)?;

    let body = compressed_body.as_deref().unwrap_or(body);

    let mut header = [0; HEADER_SIZE];

    header[0] = params.version;
    header[1] = params.flags;
    header[2..=3].copy_from_slice(&params.stream.to_be_bytes());
    header[4] = match opcode {
        FrameOpcode::Request(op) => op as u8,
        FrameOpcode::Response(op) => op as u8,
    };
    header[5..9].copy_from_slice(&(body.len() as u32).to_be_bytes());

    writer.write_all(&header).await?;
    writer.write_all(body).await?;
    writer.flush().await?;
    Ok(())
}

/// Reads a single frame off the wire.
///
/// Returns the body already decompressed, accompanied by the number of body bytes
/// that were actually read from the socket (i.e. the compressed length, if the frame
/// was compressed).
pub(crate) async fn read_frame(
    reader: &mut (impl AsyncRead + Unpin),
    frame_type: FrameType,
    compression: &CompressionReader,
) -> Result<(FrameParams, FrameOpcode, Bytes, usize), ReadFrameError> {
    let mut raw_header = [0u8; HEADER_SIZE];
    reader
        .read_exact(&mut raw_header[..])
        .await
        .map_err(FrameHeaderParseError::HeaderIoError)?;

    let mut buf = &raw_header[..];

    let version = buf.get_u8();
    {
        let (err, valid_direction, direction_str) = match frame_type {
            FrameType::Request => (FrameHeaderParseError::FrameFromServer, 0x00, "request"),
            FrameType::Response => (FrameHeaderParseError::FrameFromClient, 0x80, "response"),
        };
        if version & 0x80 != valid_direction {
            return Err(err.into());
        }
        let protocol_version = version & 0x7F;
        if protocol_version != 0x04 {
            warn!(
                "Received {} with protocol version {}.",
                direction_str, protocol_version
            );
        }
    }

    let flags = buf.get_u8();
    let stream = buf.get_i16();

    let frame_params = FrameParams {
        version,
        flags,
        stream,
    };

    let opcode = match frame_type {
        FrameType::Request => FrameOpcode::Request(
            RequestOpcode::try_from(buf.get_u8())
                .map_err(|_| FrameHeaderParseError::FrameFromServer)?,
        ),
        FrameType::Response => FrameOpcode::Response(
            ResponseOpcode::try_from(buf.get_u8())
                .map_err(|_| FrameHeaderParseError::FrameFromClient)?,
        ),
    };

    let length = buf.get_u32() as usize;

    let mut body = Vec::with_capacity(length).limit(length);

    while body.has_remaining_mut() {
        let n = reader
            .read_buf(&mut body)
            .await
            .map_err(|err| FrameHeaderParseError::BodyChunkIoError(body.remaining_mut(), err))?;
        if n == 0 {
            // EOF, too early
            return Err(
                FrameHeaderParseError::ConnectionClosed(body.remaining_mut(), length).into(),
            );
        }
    }

    let body = compression.maybe_decompress_body(flags, body.into_inner().into())?;

    // `length` is the number of body bytes read off the socket, before any decompression.
    Ok((frame_params, opcode, body, length))
}

pub(crate) async fn read_request_frame(
    reader: &mut (impl AsyncRead + Unpin),
    compression: &CompressionReader,
) -> Result<RequestFrame, ReadFrameError> {
    read_frame(reader, FrameType::Request, compression)
        .await
        .map(|(params, opcode, body, wire_body_len)| RequestFrame {
            params,
            opcode: match opcode {
                FrameOpcode::Request(op) => op,
                FrameOpcode::Response(_) => unreachable!(),
            },
            body,
            wire_body_len,
        })
}

pub(crate) async fn read_response_frame(
    reader: &mut (impl AsyncRead + Unpin),
    compression: &CompressionReader,
) -> Result<ResponseFrame, ReadFrameError> {
    read_frame(reader, FrameType::Response, compression)
        .await
        .map(|(params, opcode, body, wire_body_len)| ResponseFrame {
            params,
            opcode: match opcode {
                FrameOpcode::Request(_) => unreachable!(),
                FrameOpcode::Response(op) => op,
            },
            body,
            wire_body_len,
        })
}