qusql-mysql 0.1.0

//! Handle a single connection to Mariadb/Mysql
use std::{
    borrow::Cow,
    fmt::Display,
    marker::PhantomData,
    mem::ManuallyDrop,
    net::{IpAddr, Ipv4Addr, SocketAddr, ToSocketAddrs},
    ops::Range,
};

use bytes::{Buf, BufMut, BytesMut};
use thiserror::Error;
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    net::{TcpSocket, UnixStream},
};

use crate::{
    args::Args,
    auth::{AuthPlugin, compute_auth},
    bind::{Bind, BindError},
    constants::{client, com},
    decode::Column,
    lru::{Entry, LRUCache},
    package_parser::{DecodeError, DecodeResult, PackageParser},
    row::{FromRow, Row},
};

/// Error handling connection
#[derive(Error, Debug)]
#[non_exhaustive]
pub enum ConnectionErrorContent {
    /// An error returned by Mariadb/Mysql
    #[error("mysql error {code}: {message}")]
    Mysql {
        /// 2-byte mysql error code
        code: u16,
        /// 5-byte mysql status,
        status: [u8; 5],
        /// Error message
        message: String,
    },
    /// Network error from tokio
    #[error(transparent)]
    Io(#[from] tokio::io::Error),
    /// Error decoding row
    #[error("error reading {0}: {1}")]
    Decode(&'static str, DecodeError),
    /// Error binding parameter
    #[error("error binding paramater {0}: {1}")]
    Bind(u16, BindError),
    /// Mariadb/Mysql did not speak the protocol correctly
    #[error("protocol error {0}")]
    ProtocolError(String),
    /// You executed a mysql statement that does not return any columns
    #[error("fetch return no columns")]
    ExpectedRows,
    /// You executed a mysql statement that does return columns, so you need to read the rowss
    #[error("rows return for execute")]
    UnexpectedRows,
    #[cfg(feature = "cancel_testing")]
    /// For testing cancel safety
    #[doc(hidden)]
    #[error("await threshold reached")]
    TestCancelled,
    /// There where more instances of _LIST_ in the query, than there where list arguments
    #[error("await threshold reached")]
    TooFewListArguments,
    /// There where less instances of _LIST_ in the query, than there where list arguments
    #[error("await threshold reached")]
    TooManyListArguments,
    /// The url supplied is invalid
    #[error("Invalid url")]
    InvalidUrl,
}

/// Error handling connection
///
/// This types is a Box around ErrorContent, to make sure
/// that the error type is as small as possible
pub struct ConnectionError(Box<ConnectionErrorContent>);

const _: () = {
    assert!(size_of::<ConnectionError>() == size_of::<usize>());
};

impl ConnectionError {
    /// Return the content of the error
    pub fn content(&self) -> &ConnectionErrorContent {
        &self.0
    }
}

impl std::ops::Deref for ConnectionError {
    type Target = ConnectionErrorContent;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<E: Into<ConnectionErrorContent>> From<E> for ConnectionError {
    fn from(value: E) -> Self {
        ConnectionError(Box::new(value.into()))
    }
}

impl std::fmt::Debug for ConnectionError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Debug::fmt(&self.0, f)
    }
}

impl std::fmt::Display for ConnectionError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Display::fmt(&self.0, f)
    }
}

impl std::error::Error for ConnectionError {}

/// Result return by the connection
pub type ConnectionResult<T> = std::result::Result<T, ConnectionError>;
/// Convert [crate::package_parser::DecodeError] into [ConnectionErrorContent::Decode] with an attached location
pub trait WithLoc<T> {
    /// Convert [crate::package_parser::DecodeError] into [ConnectionErrorContent::Decode] with an attached location
    fn loc(self, loc: &'static str) -> ConnectionResult<T>;
}

impl<T> WithLoc<T> for DecodeResult<T> {
    fn loc(self, loc: &'static str) -> ConnectionResult<T> {
        self.map_err(|v| ConnectionErrorContent::Decode(loc, v).into())
    }
}

/// Trait used to check expected values
trait Except {
    /// The values to expect
    type Value;

    /// Check that I have the expected value
    fn ev(self, loc: &'static str, expected: Self::Value) -> ConnectionResult<()>;
}

impl<T: Eq + Display> Except for DecodeResult<T> {
    type Value = T;

    fn ev(self, loc: &'static str, expected: T) -> ConnectionResult<()> {
        let v = self.loc(loc)?;
        if v != expected {
            Err(ConnectionErrorContent::ProtocolError(format!(
                "Expected {expected} for {loc} got {v}"
            ))
            .into())
        } else {
            Ok(())
        }
    }
}

/// Map a row into some concrete type
pub trait RowMap<'a> {
    /// The error type returned by map
    type E: From<ConnectionError>;

    /// The value type returned by map
    type T;

    /// Map the row into a concrete type
    fn map(row: Row<'a>) -> Result<Self::T, Self::E>;
}

/// Wrapper to Use a [FromRow] as a [RowMap]
struct FromRowMapper<T>(PhantomData<T>);

impl<'a, T: FromRow<'a>> RowMap<'a> for FromRowMapper<T> {
    type E = ConnectionError;

    type T = T;

    fn map(row: Row<'a>) -> Result<Self::T, Self::E> {
        T::from_row(&row).loc("row")
    }
}

/// Read half of a split [tokio::net::tcp::TcpStream] or [tokio::net::unix::UnixStream]
enum OwnedReadHalf {
    /// Read half of a split [tokio::net::tcp::TcpStream]
    Tcp(tokio::net::tcp::OwnedReadHalf),
    /// Read half of a split [tokio::net::unix::UnixStream]
    Unix(tokio::net::unix::OwnedReadHalf),
}

/// Write half of a split [tokio::net::tcp::TcpStream] or [tokio::net::unix::UnixStream]
enum OwnedWriteHalf {
    /// Write half of a split [tokio::net::tcp::TcpStream]
    Tcp(tokio::net::tcp::OwnedWriteHalf),
    /// Write half of a split [tokio::net::unix::UnixStream]
    Unix(tokio::net::unix::OwnedWriteHalf),
}

/// Reader used to read packages from Mariadb/Mysql
struct Reader {
    /// Buffer used to contain one or more packages
    buff: BytesMut,
    /// Socket to read from
    read: OwnedReadHalf,
    /// Number of bytes to skip on next read
    skip_on_read: usize,
    /// Keep old packages, do not advance the buffer
    buffer_packages: bool,
}

impl Reader {
    /// Construct a new reader instance
    fn new(read: OwnedReadHalf) -> Self {
        Self {
            read,
            buff: BytesMut::with_capacity(1234),
            skip_on_read: 0,
            buffer_packages: false,
        }
    }

    /// Read next package, into buff and return the byte ranges in the buffer
    ///
    /// The returned future is cancel-safe
    async fn read_raw(&mut self) -> ConnectionResult<Range<usize>> {
        if !self.buffer_packages {
            self.buff.advance(self.skip_on_read);
            self.skip_on_read = 0;
        }

        while self.buff.remaining() < 4 + self.skip_on_read {
            match &mut self.read {
                OwnedReadHalf::Tcp(r) => r.read_buf(&mut self.buff).await?,
                OwnedReadHalf::Unix(r) => r.read_buf(&mut self.buff).await?,
            };
        }
        let y: u32 = u32::from_le_bytes(
            self.buff[self.skip_on_read..self.skip_on_read + 4]
                .try_into()
                .unwrap(),
        );
        let len: usize = (y & 0xFFFFFF).try_into().unwrap();
        let _s = (y >> 24) as u8;
        if len == 0xFFFFFF {
            return Err(ConnectionErrorContent::ProtocolError(
                "Extended packages not supported".to_string(),
            )
            .into());
        }
        while self.buff.remaining() < self.skip_on_read + 4 + len {
            match &mut self.read {
                OwnedReadHalf::Tcp(r) => r.read_buf(&mut self.buff).await?,
                OwnedReadHalf::Unix(r) => r.read_buf(&mut self.buff).await?,
            };
        }
        let r = self.skip_on_read + 4..self.skip_on_read + 4 + len;
        self.skip_on_read += 4 + len;
        Ok(r)
    }

    /// Read next package, into buff and return its data
    ///
    /// The returned future is cancel-safe
    #[inline]
    async fn read(&mut self) -> ConnectionResult<&[u8]> {
        let r = self.read_raw().await?;
        Ok(self.bytes(r))
    }

    /// Return the bytes for a given range.
    ///
    /// The range should have been returned by a previous read_raw
    ///
    /// [Self::buffer_packages] must have been set to true for any
    /// reads in between the read call that return r and this call
    /// to bytes for the returned bytes to make sense
    #[inline]
    fn bytes(&self, r: Range<usize>) -> &[u8] {
        &self.buff[r]
    }
}

/// Writer used to send packages to Mariadb/Mysql
struct Writer {
    /// Writer to write package to
    write: OwnedWriteHalf,
    /// Buffer containing package
    buff: BytesMut,
    /// Sequence number of package
    seq: u8,
}

impl Writer {
    /// Construct a new [Writer] instance
    fn new(write: OwnedWriteHalf) -> Self {
        Writer {
            write,
            buff: BytesMut::with_capacity(1234),
            seq: 1,
        }
    }

    /// Compose a new package
    fn compose(&mut self) -> Composer<'_> {
        self.buff.clear();
        self.buff.put_u32(0);
        Composer { writer: self }
    }

    /// Send the last composed package
    async fn send(&mut self) -> ConnectionResult<()> {
        match &mut self.write {
            OwnedWriteHalf::Tcp(r) => Ok(r.write_all_buf(&mut self.buff).await?),
            OwnedWriteHalf::Unix(r) => Ok(r.write_all_buf(&mut self.buff).await?),
        }
    }
}

/// Struct used to compose a singe network package in a [Writer]
struct Composer<'a> {
    /// Writer to compose package into
    writer: &'a mut Writer,
}

impl<'a> Composer<'a> {
    /// Write a u32 to the package
    fn put_u32(&mut self, v: u32) {
        self.writer.buff.put_u32_le(v)
    }

    /// Write a u16 to the package
    fn put_u16(&mut self, v: u16) {
        self.writer.buff.put_u16_le(v)
    }

    /// Write a u8 to the package
    fn put_u8(&mut self, v: u8) {
        self.writer.buff.put_u8(v)
    }

    /// Write a null terminated string to the package
    fn put_str_null(&mut self, s: &str) {
        self.writer.buff.put(s.as_bytes());
        self.writer.buff.put_u8(0);
    }

    /// Write write some bytes to the package
    fn put_bytes(&mut self, s: &[u8]) {
        self.writer.buff.put(s);
    }

    /// Finalize the package header
    fn finalize(self) {
        let len = self.writer.buff.len();
        let mut x = &mut self.writer.buff[..4];
        x.put_u32_le((len - 4) as u32 | ((self.writer.seq as u32) << 24));
        self.writer.seq = self.writer.seq.wrapping_add(1);
    }
}

/// Options used to establish connection to Mariadb/Mysql
pub struct ConnectionOptions<'a> {
    /// The TCP adders to connect to
    address: SocketAddr,
    /// The user to connect as
    user: Cow<'a, str>,
    /// The password for the user
    password: Cow<'a, str>,
    /// The database to connect to
    database: Option<Cow<'a, str>>,
    /// The maximum number of prepared statements in the cache
    statement_case_size: usize,
    /// Unix socket
    unix_socket: Option<Cow<'a, std::path::Path>>,
}

impl<'a> ConnectionOptions<'a> {
    /// Construct new default connection options
    pub fn new() -> ConnectionOptions<'a> {
        Default::default()
    }

    /// Construct an owned copy of the options
    pub fn into_owned(self) -> ConnectionOptions<'static> {
        ConnectionOptions {
            address: self.address,
            user: self.user.into_owned().into(),
            password: self.password.into_owned().into(),
            database: self.database.map(|v| v.into_owned().into()),
            statement_case_size: self.statement_case_size,
            unix_socket: self.unix_socket.map(|v| v.into_owned().into()),
        }
    }

    /// Construct a connection options from a url
    /// ```
    /// # use qusql_mysql::ConnectionOptions;
    /// # fn main() {
    /// let o = ConnectionOptions::from_url(
    ///     "mysql://user:password@localhost/database?socket=/var/run/mysql.sock"
    /// ).unwrap();
    /// # }
    /// ```
    pub fn from_url(url: &'a str) -> Result<Self, ConnectionError> {
        let Some(v) = url.strip_prefix("mysql://") else {
            return Err(ConnectionErrorContent::InvalidUrl.into());
        };
        let (authority, path) = v
            .split_once('/')
            .map(|(a, b)| (a, Some(b)))
            .unwrap_or((v, None));
        let (user_info, address) = authority
            .split_once('@')
            .map(|(a, b)| (Some(a), b))
            .unwrap_or((None, authority));
        let (user, password) = user_info
            .map(|v| {
                v.split_once(':')
                    .map(|(a, b)| (Some(a), Some(b)))
                    .unwrap_or((Some(v), None))
            })
            .unwrap_or_default();
        let (host, port) = address
            .rsplit_once(':')
            .map(|(a, b)| (a, Some(b)))
            .unwrap_or((address, None));
        let port: u16 = match port {
            Some(v) => v.parse().map_err(|_| ConnectionErrorContent::InvalidUrl)?,
            None => 3306,
        };
        let (db, unix_socket) = path
            .map(|v| {
                v.split_once("?socket=")
                    .map(|(a, b)| (Some(a), Some(std::path::Path::new(b).into())))
                    .unwrap_or((Some(v), None))
            })
            .unwrap_or_default();

        let mut addrs = (host, port).to_socket_addrs()?;
        let Some(address) = addrs.next() else {
            return Err(ConnectionErrorContent::InvalidUrl.into());
        };

        Ok(ConnectionOptions {
            address,
            user: user.unwrap_or("root").into(),
            password: password.unwrap_or("password").into(),
            database: db.map(|v| v.into()),
            unix_socket,
            ..Default::default()
        })
    }

    /// Set the user
    pub fn user(self, user: impl Into<Cow<'a, str>>) -> Self {
        Self {
            user: user.into(),
            ..self
        }
    }

    /// Set the password
    pub fn password(self, password: impl Into<Cow<'a, str>>) -> Self {
        Self {
            password: password.into(),
            ..self
        }
    }

    /// Set the database to connect to
    pub fn database(self, database: impl Into<Cow<'a, str>>) -> Self {
        Self {
            database: Some(database.into()),
            ..self
        }
    }

    /// The address to connect to
    pub fn address(self, address: impl std::net::ToSocketAddrs) -> Result<Self, std::io::Error> {
        match address.to_socket_addrs()?.next() {
            Some(v) => Ok(Self { address: v, ..self }),
            None => Err(std::io::Error::new(
                std::io::ErrorKind::NotFound,
                "No host resolved",
            )),
        }
    }

    /// Set the unix socket to connect to
    pub fn unix_socket(self, path: impl Into<Cow<'a, std::path::Path>>) -> Self {
        Self {
            unix_socket: Some(path.into()),
            ..self
        }
    }

    /// The maximum number of prepared statements in the cache
    pub fn statment_case_size(self, size: usize) -> Self {
        Self {
            statement_case_size: size,
            ..self
        }
    }
}

impl<'a> Default for ConnectionOptions<'a> {
    fn default() -> Self {
        Self {
            address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 3306),
            user: Cow::Borrowed("root"),
            password: Cow::Borrowed("password"),
            database: None,
            statement_case_size: 1024,
            unix_socket: None,
        }
    }
}

/// Definition of a column or parameter, see <https://mariadb.com/docs/server/reference/clientserver-protocol/4-server-response-packets/result-set-packets#column-definition-packet>
#[derive(Debug)]
#[non_exhaustive]
pub struct ColumnDefinition<'a> {
    /// Schema
    pub schema: &'a str,
    /// Alias for the table
    pub table_alias: &'a str,
    /// Table
    pub table: &'a str,
    /// Alias for column
    pub column_alias: &'a str,
    /// Column
    pub column: &'a str,
    /// Extended metadata, currently always null
    pub extended_matadata: Option<&'a str>,
    /// Number of the character set used
    pub character_set_number: u16,
    /// The maximal column size
    pub max_column_size: u32,
    /// The type of the field, see <https://mariadb.com/docs/server/reference/clientserver-protocol/4-server-response-packets/result-set-packets#field-types>
    pub field_types: u8,
    /// Flags set on the field, see <https://mariadb.com/docs/server/reference/clientserver-protocol/4-server-response-packets/result-set-packets#field-details-flag>
    pub field_detail_flags: u16,
    /// decimals used
    pub decimals: u8,
}

impl<'a> ColumnDefinition<'a> {
    /// Parse a ColumnDefinition package
    fn new(data: &'a [u8]) -> DecodeResult<Self> {
        let mut p = PackageParser::new(data);
        p.skip_lenenc_str()?; // Skip def
        let schema = p.get_lenenc_str()?;
        let table_alias = p.get_lenenc_str()?;
        let table = p.get_lenenc_str()?;
        let column_alias = p.get_lenenc_str()?;
        let column = p.get_lenenc_str()?;
        let extended_matadata = None;
        p.get_lenenc()?;
        let character_set_number = p.get_u16()?;
        let max_column_size = p.get_u32()?;
        let field_types = p.get_u8()?;
        let field_detail_flags = p.get_u16()?;
        let decimals = p.get_u8()?;
        Ok(ColumnDefinition {
            schema,
            table_alias,
            table,
            column_alias,
            column,
            extended_matadata,
            character_set_number,
            max_column_size,
            field_types,
            field_detail_flags,
            decimals,
        })
    }
}

/// Columns or parameters of a prepared query
pub struct ColumnsInformation<'a> {
    /// Bytes of statement information
    data: &'a [u8],
    /// Offset in bytes of the column information packages
    ranges: &'a [Range<usize>],
}

impl<'a> ColumnsInformation<'a> {
    /// Return the idx eth element
    pub fn get(&self, idx: usize) -> Option<DecodeResult<ColumnDefinition<'a>>> {
        self.ranges
            .get(idx)
            .map(|v| ColumnDefinition::new(&self.data[v.clone()]))
    }
}

impl<'a> Iterator for ColumnsInformation<'a> {
    type Item = DecodeResult<ColumnDefinition<'a>>;

    fn next(&mut self) -> Option<Self::Item> {
        match self.ranges.split_off_first() {
            Some(v) => Some(ColumnDefinition::new(&self.data[v.clone()])),
            None => None,
        }
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        (self.ranges.len(), Some(self.ranges.len()))
    }

    fn nth(&mut self, n: usize) -> Option<Self::Item> {
        self.get(n)
    }
}

impl<'a> ExactSizeIterator for ColumnsInformation<'a> {
    fn len(&self) -> usize {
        self.ranges.len()
    }
}

impl StatementInformation {
    /// Information about the columns returned by the statement
    pub fn columns(&self) -> ColumnsInformation<'_> {
        ColumnsInformation {
            data: &self.info,
            ranges: &self.ranges[self.num_params as usize..],
        }
    }

    /// Information about the parameters assigned to the statement
    pub fn parameters(&self) -> ColumnsInformation<'_> {
        ColumnsInformation {
            data: &self.info,
            ranges: &self.ranges[..self.num_params as usize],
        }
    }
}

/// Contains information about a prepared statement
pub struct StatementInformation {
    /// The number of parameters the statement takes
    num_params: u16,
    /// Bytes of statement information
    info: Vec<u8>,
    /// Offset of params and columns
    ranges: Vec<Range<usize>>,
}

/// A prepared statement
struct Statement {
    /// The id of the statement
    stmt_id: u32,
    /// The number of parameters the statement takes
    num_params: u16,
    /// Information about the statement
    information: Option<StatementInformation>,
}

/// Iterate over rows in a query result
pub struct QueryIterator<'a> {
    /// A reference to the connection
    connection: &'a mut RawConnection,
}

impl<'a> QueryIterator<'a> {
    /// Fetch the next row from the result set
    ///
    /// The returned future is cancel-safe.
    pub async fn next(&mut self) -> ConnectionResult<Option<Row<'_>>> {
        match self.connection.state {
            ConnectionState::Clean => return Ok(None),
            ConnectionState::QueryReadRows => (),
            _ => panic!("Logic error"),
        }
        // safety-cancel: The cleanup on the connection will skip the remaining rows
        self.connection.test_cancel()?;
        let start_instant = self.connection.stats.get_instant();
        let package = self.connection.reader.read().await?;
        self.connection.stats.add_fetch(start_instant);
        let mut pp = PackageParser::new(package);
        match pp.get_u8().loc("Row first byte")? {
            0x00 => Ok(Some(Row::new(&self.connection.columns, package))),
            0xFE => {
                //EOD
                self.connection.state = ConnectionState::Clean;
                Ok(None)
            }
            0xFF => {
                handle_mysql_error(&mut pp)?;
                unreachable!()
            }
            v => Err(ConnectionErrorContent::ProtocolError(format!(
                "Unexpected response type {v} to row package"
            ))
            .into()),
        }
    }
}

/// Iterate over mapped rows in a query result
pub struct MapQueryIterator<'a, M>
where
    for<'b> M: RowMap<'b>,
{
    /// A reference to the connection
    connection: &'a mut RawConnection,
    /// We need a phantom data fro M
    _phantom: PhantomData<M>,
}

impl<'a, M> MapQueryIterator<'a, M>
where
    for<'b> M: RowMap<'b>,
{
    /// Fetch the next row from the result mapped using M
    ///
    /// The returned future is cancel-safe.
    pub async fn next<'b>(
        &'b mut self,
    ) -> Result<Option<<M as RowMap<'b>>::T>, <M as RowMap<'b>>::E> {
        match self.connection.state {
            ConnectionState::Clean => return Ok(None),
            ConnectionState::QueryReadRows => (),
            _ => panic!("Logic error"),
        }
        // safety-cancel: The cleanup on the connection will skip the remaining rows
        self.connection.test_cancel()?;
        let start_instant = self.connection.stats.get_instant();
        let package = self.connection.reader.read().await?;
        self.connection.stats.add_fetch(start_instant);
        let mut pp = PackageParser::new(package);
        match pp.get_u8().loc("Row first byte")? {
            0x00 => Ok(Some(M::map(Row::new(&self.connection.columns, package))?)),
            0xFE => {
                //EOD
                self.connection.state = ConnectionState::Clean;
                Ok(None)
            }
            0xFF => {
                handle_mysql_error(&mut pp)?;
                unreachable!()
            }
            v => Err(
                ConnectionError::from(ConnectionErrorContent::ProtocolError(format!(
                    "Unexpected response type {v} to row package"
                )))
                .into(),
            ),
        }
    }
}

/// The result of a execute
pub struct ExecuteResult {
    /// The number of rows affected by the query
    affected_rows: u64,
    /// The id of the last row inserted
    last_insert_id: u64,
}

impl ExecuteResult {
    /// The number of rows affected by the query
    pub fn affected_rows(&self) -> u64 {
        self.affected_rows
    }

    /// The id of the last row inserted
    pub fn last_insert_id(&self) -> u64 {
        self.last_insert_id
    }
}

/// The result of a query
enum QueryResult {
    /// The query returned columns
    WithColumns,
    /// The query returns no columns
    ExecuteResult(ExecuteResult),
}

/// Internal state of the connection. used for cleanup.
#[derive(Clone, Copy, Debug)]
enum ConnectionState {
    /// The connection is in a clean state
    Clean,
    /// We are sending the prepare statement package
    PrepareStatementSend,
    /// We are reading the prepare statement response
    PrepareStatementReadHead,
    /// We are reading the prepare statement parameters
    PrepareStatementReadParams {
        /// Number of params left to read
        params: u16,
        /// Number of columns left to read
        columns: u16,
        /// The id of the prepared statement
        stmt_id: u32,
    },
    /// We are sending a close prepared statement package
    ClosePreparedStatement,
    /// We are sending a query
    QuerySend,
    /// We are reading a query response
    QueryReadHead,
    /// We are reading query columns
    QueryReadColumns(u64),
    /// We are reading query rows
    QueryReadRows,
    /// We are sending an unprepared statement
    UnpreparedSend,
    /// We are receiving the response of an unprepared statement
    UnpreparedRecv,
    /// We are sending an ping
    PingSend,
    /// We are receiving the response to a ping
    PingRecv,
    /// The connection is in a broken state and cannot be recovered
    Broken,
}

/// Interface implemented by [Stats] and [NoStats]
trait IStats {
    /// Instant in time
    type Instant: Sized;
    /// Get the current instant in time
    fn get_instant(&self) -> Self::Instant;
    /// Register a prepare
    fn add_prepare(&mut self, start_instant: Self::Instant);
    /// Register an execute
    fn add_execute(&mut self, start_instant: Self::Instant);
    /// Register som fetching
    fn add_fetch(&mut self, start_instant: Self::Instant);
}

/// Statistics about a connection
#[allow(unused)]
#[derive(Default, Debug)]
pub struct Stats {
    /// Number of statements prepared
    pub prepare_counts: usize,
    /// Time used preparing statements
    pub prepare_time: std::time::Duration,
    /// Number of statements executed
    pub execute_counts: usize,
    /// Time used in executing statements
    pub execute_time: std::time::Duration,
    /// Time used fetching statement results
    pub fetch_time: std::time::Duration,
}

impl IStats for Stats {
    type Instant = std::time::Instant;

    fn get_instant(&self) -> Self::Instant {
        std::time::Instant::now()
    }

    fn add_prepare(&mut self, start_instant: Self::Instant) {
        self.prepare_counts += 1;
        self.prepare_time += start_instant.elapsed();
    }

    fn add_execute(&mut self, start_instant: Self::Instant) {
        self.execute_counts += 1;
        self.execute_time += start_instant.elapsed()
    }

    fn add_fetch(&mut self, start_instant: Self::Instant) {
        self.fetch_time += start_instant.elapsed();
    }
}
/// Statistics about a connection
#[allow(unused)]
#[derive(Default)]
struct NoStats;

impl IStats for NoStats {
    type Instant = NoStats;

    fn get_instant(&self) -> Self::Instant {
        NoStats
    }

    fn add_prepare(&mut self, _: Self::Instant) {}

    fn add_execute(&mut self, _: Self::Instant) {}

    fn add_fetch(&mut self, _: Self::Instant) {}
}

/// A raw connection to Mariadb/Mysql. This is split from [Connection] to that connection
/// can handle the caching of prepared statement
struct RawConnection {
    /// The reader to read packages from
    reader: Reader,
    /// The writer to write packages to
    writer: Writer,
    /// The current state of the connection
    state: ConnectionState,
    /// Buffer to store column information
    columns: Vec<Column>,
    /// Ranges used by fetch all
    ranges: Vec<Range<usize>>,
    #[cfg(feature = "cancel_testing")]
    /// Return TestCancelled after this many sends
    cancel_count: Option<usize>,
    #[cfg(feature = "stats")]
    /// Connection statistics
    stats: Stats,
    #[cfg(not(feature = "stats"))]
    /// Connection statistics
    stats: NoStats,
    #[cfg(feature = "list_hack")]
    /// List lengths allocation
    list_lengths: Vec<usize>,
}

/// Parse a column definition package
fn parse_column_definition(p: &mut PackageParser) -> ConnectionResult<Column> {
    p.skip_lenenc_str().loc("catalog")?;
    p.skip_lenenc_str().loc("schema")?;
    p.skip_lenenc_str().loc("table")?;
    p.skip_lenenc_str().loc("org_table")?;
    p.skip_lenenc_str().loc("name")?;
    p.skip_lenenc_str().loc("org_name")?;
    p.get_lenenc().loc("length of fixed length fields")?;
    let character_set = p.get_u16().loc("character_set")?;
    p.get_u32().loc("column_length")?;
    let r#type = p.get_u8().loc("type")?;
    let flags = p.get_u16().loc("flags")?;
    p.get_u8().loc("decimals")?;
    p.get_u16().loc("res")?;
    Ok(Column {
        r#type,
        flags,
        character_set,
    })
}

/// Parse an error package
fn handle_mysql_error(pp: &mut PackageParser) -> ConnectionResult<std::convert::Infallible> {
    // We assume the 255 has been read
    let code = pp.get_u16().loc("code")?;
    pp.get_u8().ev("sharp", b'#')?;
    let a = pp.get_u8().loc("status0")?;
    let b = pp.get_u8().loc("status1")?;
    let c = pp.get_u8().loc("status2")?;
    let d = pp.get_u8().loc("status3")?;
    let e = pp.get_u8().loc("status4")?;
    let msg = pp.get_eof_str().loc("message")?;
    Err(ConnectionErrorContent::Mysql {
        code,
        status: [a, b, c, d, e],
        message: msg.to_string(),
    }
    .into())
}

/// Compute string to begin transaction
fn begin_transaction_query(depth: usize) -> Cow<'static, str> {
    match depth {
        0 => "BEGIN".into(),
        1 => "SAVEPOINT _sqly_savepoint_1".into(),
        2 => "SAVEPOINT _sqly_savepoint_2".into(),
        3 => "SAVEPOINT _sqly_savepoint_3".into(),
        v => format!("SAVEPOINT _sqly_savepoint_{}", v).into(),
    }
}

/// Compute string to commit transaction
fn commit_transaction_query(depth: usize) -> Cow<'static, str> {
    match depth {
        0 => "COMMIT".into(),
        1 => "RELEASE SAVEPOINT _sqly_savepoint_1".into(),
        2 => "RELEASE SAVEPOINT _sqly_savepoint_2".into(),
        3 => "RELEASE SAVEPOINT _sqly_savepoint_3".into(),
        v => format!("RELEASE SAVEPOINT _sqly_savepoint_{}", v).into(),
    }
}

/// Compute string to rollback transaction
fn rollback_transaction_query(depth: usize) -> Cow<'static, str> {
    match depth {
        0 => "ROLLBACK".into(),
        1 => "ROLLBACK TO SAVEPOINT _sqly_savepoint_1".into(),
        2 => "ROLLBACK TO SAVEPOINT _sqly_savepoint_2".into(),
        3 => "ROLLBACK TO SAVEPOINT _sqly_savepoint_3".into(),
        v => format!("RELEASE TO SAVEPOINT _sqly_savepoint_{}", v).into(),
    }
}

impl RawConnection {
    /// Connect to Mariadb/Mysql
    ///
    /// The returned future is cancel safe
    async fn connect(options: &ConnectionOptions<'_>) -> ConnectionResult<Self> {
        // safety-cancel: It is safe to drop this future since it does not mute shared state

        // Connect to socket
        let (read, write) = if let Some(path) = &options.unix_socket {
            let socket = UnixStream::connect(path).await?;
            let (read, write) = socket.into_split();
            (OwnedReadHalf::Unix(read), OwnedWriteHalf::Unix(write))
        } else {
            let stream = if options.address.is_ipv4() {
                let socket = TcpSocket::new_v4()?;
                socket.connect(options.address).await?
            } else {
                let socket = TcpSocket::new_v6()?;
                socket.connect(options.address).await?
            };
            let (read, write) = stream.into_split();
            (OwnedReadHalf::Tcp(read), OwnedWriteHalf::Tcp(write))
        };

        let mut reader = Reader::new(read);
        let mut writer = Writer::new(write);

        // Read and parse handshake package
        let package = reader.read().await?;
        let mut p = PackageParser::new(package);
        p.get_u8().ev("protocol version", 10)?;
        p.skip_null_str().loc("status")?;
        let _wthread_id = p.get_u32().loc("thread_id")?;
        let nonce1 = p.get_bytes(8).loc("nonce1")?;
        p.get_u8().ev("nonce1_end", 0)?;
        let capability_flags_1 = p.get_u16().loc("capability_flags_1")?;
        let _character_set = p.get_u8().loc("character_set")?;
        p.get_u16().loc("status_flags")?;
        let capability_flags_2 = p.get_u16().loc("capability_flags_2")?;
        let auth_plugin_data_len = p.get_u8().loc("auth_plugin_data_len")?;
        let _capability_flags = capability_flags_1 as u32 | (capability_flags_2 as u32) << 16;
        p.get_bytes(10).loc("reserved")?;
        let nonce2 = p
            .get_bytes(auth_plugin_data_len as usize - 9)
            .loc("nonce2")?;
        p.get_u8().ev("nonce2_end", 0)?;

        let auth_plugin = p.get_null_str().loc("auth_plugin")?;
        let auth_method = match auth_plugin {
            "mysql_native_password" => AuthPlugin::NativePassword,
            #[cfg(feature = "sha2_auth")]
            "caching_sha2_password" => AuthPlugin::CachingSha2Password,
            v => {
                return Err(ConnectionErrorContent::ProtocolError(format!(
                    "Unhandled auth plugin {v}"
                ))
                .into());
            }
        };

        // Compose and send handshake response
        let mut p = writer.compose();
        let mut opts = client::LONG_PASSWORD
            | client::LONG_FLAG
            | client::LOCAL_FILES
            | client::PROTOCOL_41
            | client::DEPRECATE_EOF
            | client::TRANSACTIONS
            | client::SECURE_CONNECTION
            | client::MULTI_STATEMENTS
            | client::MULTI_RESULTS
            | client::PS_MULTI_RESULTS
            | client::PLUGIN_AUTH;
        if options.database.is_some() {
            opts |= client::CONNECT_WITH_DB
        }
        p.put_u32(opts);
        p.put_u32(0x1000000); // Max package size
        p.put_u16(45); //utf8mb4_general_ci
        for _ in 0..22 {
            p.put_u8(0);
        }
        p.put_str_null(&options.user);

        let mut nonce = Vec::with_capacity(nonce1.len() + nonce2.len());
        nonce.extend_from_slice(nonce1);
        nonce.extend_from_slice(nonce2);

        let auth = compute_auth(&options.password, &nonce, auth_method);
        let auth = auth.as_slice();
        p.put_u8(auth.len() as u8);
        for v in auth {
            p.put_u8(*v);
        }
        if let Some(database) = &options.database {
            p.put_str_null(database);
        }
        // mysql_native_password
        p.put_str_null(auth_plugin);
        p.finalize();

        writer.send().await?;

        loop {
            let p = reader.read().await?;
            let mut pp = PackageParser::new(p);
            match pp.get_u8().loc("response type")? {
                0xFF => {
                    handle_mysql_error(&mut pp)?;
                }
                0x00 => {
                    let _rows = pp.get_lenenc().loc("rows")?;
                    let _last_inserted_id = pp.get_lenenc().loc("last_inserted_id")?;
                    break;
                }
                0xFE => {
                    return Err(ConnectionErrorContent::ProtocolError(
                        "Unexpected auth switch".into(),
                    )
                    .into());
                }
                #[cfg(feature = "sha2_auth")]
                0x01 if matches!(auth_method, AuthPlugin::CachingSha2Password) => {
                    match pp.get_u8().loc("auth_status")? {
                        // AUTH_OK
                        0x03 => break,
                        // AUTH_CONTINUE
                        0x04 => {
                            // https://mariadb.com/kb/en/caching_sha2_password-authentication-plugin/
                            writer.seq = 3;
                            let mut p = writer.compose();
                            p.put_u8(0x02);
                            p.finalize();
                            writer.send().await?;

                            let p = reader.read().await?;
                            let mut pp = PackageParser::new(p);
                            pp.get_u8().ev("first", 1)?;
                            let pem = pp.get_eof_str().loc("pem")?;

                            let pwd = crate::auth::encrypt_rsa(pem, &options.password, &nonce)?;

                            writer.seq = 5;
                            let mut p = writer.compose();
                            p.put_bytes(&pwd);
                            p.finalize();
                            writer.send().await?;
                        }
                        v => {
                            return Err(ConnectionErrorContent::ProtocolError(format!(
                                "Unexpected auth status {v} to handshake response"
                            ))
                            .into());
                        }
                    }
                }
                v => {
                    return Err(ConnectionErrorContent::ProtocolError(format!(
                        "Unexpected response type {v} to handshake response"
                    ))
                    .into());
                }
            }
        }
        writer.seq = 0;
        Ok(RawConnection {
            reader,
            writer,
            state: ConnectionState::Clean,
            columns: Vec::new(),
            ranges: Vec::new(),
            #[cfg(feature = "cancel_testing")]
            cancel_count: None,
            stats: Default::default(),
            #[cfg(feature = "list_hack")]
            list_lengths: Vec::new(),
        })
    }

    /// Can be called self.cancel_count times
    /// before it returns Err(ConnectionError::TestCancelled)
    ///
    /// This is used to to test that that we can properly recover
    /// from dropped futures
    #[inline]
    fn test_cancel(&mut self) -> ConnectionResult<()> {
        #[cfg(feature = "cancel_testing")]
        if let Some(v) = &mut self.cancel_count {
            if *v == 0 {
                return Err(ConnectionErrorContent::TestCancelled.into());
            }
            *v -= 1;
        }
        Ok(())
    }

    /// Cleanup The connection if it is dirty
    async fn cleanup(&mut self) -> ConnectionResult<()> {
        loop {
            match self.state {
                ConnectionState::Clean => break,
                ConnectionState::PrepareStatementSend => {
                    self.test_cancel()?;
                    self.writer.send().await?;
                    self.state = ConnectionState::PrepareStatementReadHead;
                    continue;
                }
                ConnectionState::PrepareStatementReadHead => {
                    self.test_cancel()?;
                    let package = self.reader.read().await?;
                    let mut p = PackageParser::new(package);
                    match p.get_u8().loc("response type")? {
                        0 => {
                            let stmt_id = p.get_u32().loc("stmt_id")?;
                            let columns = p.get_u16().loc("num_columns")?;
                            let params = p.get_u16().loc("num_params")?;
                            self.state = ConnectionState::PrepareStatementReadParams {
                                params,
                                columns,
                                stmt_id,
                            };
                            continue;
                        }
                        255 => {
                            self.state = ConnectionState::Clean;
                        }
                        v => {
                            self.state = ConnectionState::Broken;
                            return Err(ConnectionErrorContent::ProtocolError(format!(
                                "Unexpected response type {v} to prepare statement"
                            ))
                            .into());
                        }
                    }
                }
                ConnectionState::PrepareStatementReadParams {
                    params: 0,
                    columns: 0,
                    stmt_id,
                } => {
                    self.writer.seq = 0;
                    let mut p = self.writer.compose();
                    p.put_u8(com::STMT_CLOSE);
                    p.put_u32(stmt_id);
                    p.finalize();
                    self.state = ConnectionState::ClosePreparedStatement;
                }
                ConnectionState::PrepareStatementReadParams {
                    params: 0,
                    columns,
                    stmt_id,
                } => {
                    self.test_cancel()?;
                    self.reader.read().await?;
                    self.state = ConnectionState::PrepareStatementReadParams {
                        params: 0,
                        columns: columns - 1,
                        stmt_id,
                    };
                }
                ConnectionState::PrepareStatementReadParams {
                    params,
                    columns,
                    stmt_id,
                } => {
                    self.test_cancel()?;
                    self.reader.read().await?;
                    self.state = ConnectionState::PrepareStatementReadParams {
                        params: params - 1,
                        columns,
                        stmt_id,
                    };
                }
                ConnectionState::ClosePreparedStatement => {
                    self.test_cancel()?;
                    self.writer.send().await?;
                    self.state = ConnectionState::Clean;
                }
                ConnectionState::QuerySend => {
                    self.test_cancel()?;
                    self.writer.send().await?;
                    self.state = ConnectionState::QueryReadHead;
                }
                ConnectionState::QueryReadHead => {
                    self.test_cancel()?;
                    let package = self.reader.read().await?;
                    {
                        let mut pp = PackageParser::new(package);
                        match pp.get_u8().loc("first_byte")? {
                            255 | 0 => {
                                self.state = ConnectionState::Clean;
                                continue;
                            }
                            _ => (),
                        }
                    }
                    let column_count = PackageParser::new(package)
                        .get_lenenc()
                        .loc("column_count")?;
                    self.state = ConnectionState::QueryReadColumns(column_count)
                }
                ConnectionState::QueryReadColumns(0) => {
                    self.state = ConnectionState::QueryReadRows;
                }
                ConnectionState::QueryReadColumns(cnt) => {
                    self.test_cancel()?;
                    self.reader.read().await?;
                    self.state = ConnectionState::QueryReadColumns(cnt - 1);
                }
                ConnectionState::QueryReadRows => {
                    self.test_cancel()?;
                    let package = self.reader.read().await?;
                    let mut pp = PackageParser::new(package);
                    match pp.get_u8().loc("Row first byte")? {
                        0x00 => (),
                        0xFE => {
                            //EOD
                            self.state = ConnectionState::Clean;
                        }
                        0xFF => {
                            self.state = ConnectionState::Broken;
                            handle_mysql_error(&mut pp)?;
                            unreachable!()
                        }
                        v => {
                            self.state = ConnectionState::Broken;
                            return Err(ConnectionErrorContent::ProtocolError(format!(
                                "Unexpected response type {v} to row package"
                            ))
                            .into());
                        }
                    }
                }
                ConnectionState::UnpreparedSend => {
                    self.test_cancel()?;
                    self.writer.send().await?;
                    self.state = ConnectionState::QueryReadHead;
                }
                ConnectionState::UnpreparedRecv => {
                    self.test_cancel()?;
                    let package = self.reader.read().await?;
                    let mut pp = PackageParser::new(package);
                    match pp.get_u8().loc("first_byte")? {
                        255 => {
                            self.state = ConnectionState::Broken;
                            handle_mysql_error(&mut pp)?;
                            unreachable!()
                        }
                        0 => {
                            self.state = ConnectionState::Clean;
                            return Ok(());
                        }
                        v => {
                            self.state = ConnectionState::Broken;
                            return Err(ConnectionErrorContent::ProtocolError(format!(
                                "Unexpected response type {v} to row package"
                            ))
                            .into());
                        }
                    }
                }
                ConnectionState::PingSend => {
                    self.test_cancel()?;
                    self.writer.send().await?;
                    self.state = ConnectionState::PingRecv;
                }
                ConnectionState::PingRecv => {
                    self.test_cancel()?;
                    let package = self.reader.read().await?;
                    let mut pp = PackageParser::new(package);
                    match pp.get_u8().loc("first_byte")? {
                        255 => {
                            self.state = ConnectionState::Broken;
                            handle_mysql_error(&mut pp)?;
                            unreachable!()
                        }
                        0 => {
                            self.state = ConnectionState::Clean;
                            return Ok(());
                        }
                        v => {
                            self.state = ConnectionState::Broken;
                            return Err(ConnectionErrorContent::ProtocolError(format!(
                                "Unexpected response type {v} to ping"
                            ))
                            .into());
                        }
                    }
                }
                ConnectionState::Broken => {
                    return Err(ConnectionErrorContent::ProtocolError(
                        "Previous protocol error reported".to_string(),
                    )
                    .into());
                }
            }
        }
        Ok(())
    }

    /// Prepare a statement.
    ///
    /// The cleanup functionally ensures that the future is drop drop safe
    async fn prepare_query(&mut self, stmt: &str, with_info: bool) -> ConnectionResult<Statement> {
        assert!(matches!(self.state, ConnectionState::Clean));
        self.writer.seq = 0;
        let mut p = self.writer.compose();
        p.put_u8(com::STMT_PREPARE);
        p.put_bytes(stmt.as_bytes());
        p.finalize();

        let start_instant = self.stats.get_instant();
        self.state = ConnectionState::PrepareStatementSend;
        // safety-cancel: We set the state so that cleanup will get us in a good state
        self.test_cancel()?;
        self.writer.send().await?;

        self.state = ConnectionState::PrepareStatementReadHead;
        // safety-cancel: We set the state so that cleanup will get us in a good state
        self.test_cancel()?;
        let package = self.reader.read().await?;

        let mut p = PackageParser::new(package);
        match p.get_u8().loc("response type")? {
            0 => {
                let stmt_id = p.get_u32().loc("stmt_id")?;
                let num_columns = p.get_u16().loc("num_columns")?;
                let num_params = p.get_u16().loc("num_params")?;
                // We skip the rest of the package here

                let mut info_bytes: Vec<_> = Vec::new();
                let mut info_ranges = Vec::new();

                // Read param definitions
                for p in 0..num_params {
                    self.state = ConnectionState::PrepareStatementReadParams {
                        params: num_params - p,
                        columns: num_columns,
                        stmt_id,
                    };
                    // safety-cancel: We set the state so that cleanup will get us in a good state
                    self.test_cancel()?;
                    let pkg = self.reader.read().await?;
                    if with_info {
                        let start = info_bytes.len();
                        info_bytes.extend(pkg);
                        info_ranges.push(start..info_bytes.len())
                    }
                }

                // Skip column definitions
                for c in 0..num_columns {
                    self.state = ConnectionState::PrepareStatementReadParams {
                        params: 0,
                        columns: num_columns - c,
                        stmt_id,
                    };
                    // safety-cancel: We set the state so that cleanup will get us in a good state
                    self.test_cancel()?;
                    let pkg = self.reader.read().await?;
                    if with_info {
                        let start = info_bytes.len();
                        info_bytes.extend(pkg);
                        info_ranges.push(start..info_bytes.len())
                    }
                }

                let information = if with_info {
                    Some(StatementInformation {
                        num_params,
                        info: info_bytes,
                        ranges: info_ranges,
                    })
                } else {
                    None
                };

                self.state = ConnectionState::Clean;
                self.stats.add_prepare(start_instant);
                Ok(Statement {
                    stmt_id,
                    num_params,
                    information,
                })
            }
            255 => {
                handle_mysql_error(&mut p)?;
                unreachable!()
            }
            v => {
                self.state = ConnectionState::Broken;
                Err(ConnectionErrorContent::ProtocolError(format!(
                    "Unexpected response type {v} to prepare statement"
                ))
                .into())
            }
        }
    }

    /// Begin execution of a prepared statement
    fn query<'a>(&'a mut self, statement: &'a Statement) -> Query<'a> {
        assert!(matches!(self.state, ConnectionState::Clean));

        self.writer.seq = 0;
        let mut p = self.writer.compose();
        p.put_u8(com::STMT_EXECUTE);
        p.put_u32(statement.stmt_id);
        p.put_u8(0); // flags
        p.put_u32(1); // iteration_count

        let null_offset = p.writer.buff.len();
        let mut type_offset = null_offset;
        if statement.num_params != 0 {
            let null_bytes = statement.num_params.div_ceil(8);
            // Dummy null
            for _ in 0..null_bytes {
                p.put_u8(0);
            }
            p.put_u8(1); // Send types

            type_offset = p.writer.buff.len();
            // Dummy types and flags
            for _ in 0..statement.num_params {
                p.put_u16(0);
            }
        }

        Query {
            connection: self,
            statement,
            cur_param: 0,
            null_offset,
            type_offset,
        }
    }

    /// Execute a prepared statement
    async fn query_send(&mut self) -> ConnectionResult<QueryResult> {
        let p = Composer {
            writer: &mut self.writer,
        };
        p.finalize();

        let start_instant = self.stats.get_instant();
        self.state = ConnectionState::QuerySend;
        // safety-cancel: We have set the state so that clean can finish up
        self.test_cancel()?;
        self.writer.send().await?;

        self.state = ConnectionState::QueryReadHead;
        // safety-cancel: We have set the state so that clean can finish up
        self.test_cancel()?;
        let package = self.reader.read().await?;
        {
            let mut pp = PackageParser::new(package);
            match pp.get_u8().loc("first_byte")? {
                255 => {
                    handle_mysql_error(&mut pp)?;
                }
                0 => {
                    self.stats.add_execute(start_instant);
                    self.state = ConnectionState::Clean;
                    let affected_rows = pp.get_lenenc().loc("affected_rows")?;
                    let last_insert_id = pp.get_lenenc().loc("last_insert_id")?;
                    return Ok(QueryResult::ExecuteResult(ExecuteResult {
                        affected_rows,
                        last_insert_id,
                    }));
                }
                _ => (),
            }
        }

        let column_count = PackageParser::new(package)
            .get_lenenc()
            .loc("column_count")?;

        self.columns.clear();

        // Skip column definitions
        for c in 0..column_count {
            self.state = ConnectionState::QueryReadColumns(column_count - c);
            // safety-cancel: We have set the state so that clean can finish up
            self.test_cancel()?;
            let package = self.reader.read().await?;
            let mut p = PackageParser::new(package);
            self.columns.push(parse_column_definition(&mut p)?);
        }
        self.stats.add_execute(start_instant);

        self.state = ConnectionState::QueryReadRows;
        Ok(QueryResult::WithColumns)
    }

    /// Execute an unprepared statement
    ///
    /// This is not implemented as a async method since we need to guarantee
    /// that the package has been composed correctly before the first await point
    fn execute_unprepared(
        &mut self,
        escaped_sql: Cow<'_, str>,
    ) -> impl Future<Output = ConnectionResult<ExecuteResult>> + Send {
        assert!(matches!(self.state, ConnectionState::Clean));
        self.writer.seq = 0;
        let mut p = self.writer.compose();
        p.put_u8(com::QUERY);
        p.put_bytes(escaped_sql.as_bytes());
        p.finalize();

        self.state = ConnectionState::UnpreparedSend;

        async move {
            let start_time = self.stats.get_instant();
            self.test_cancel()?;
            self.writer.send().await?;

            self.state = ConnectionState::UnpreparedRecv;
            self.test_cancel()?;
            let package = self.reader.read().await?;
            {
                let mut pp = PackageParser::new(package);
                match pp.get_u8().loc("first_byte")? {
                    255 => {
                        handle_mysql_error(&mut pp)?;
                        unreachable!()
                    }
                    0 => {
                        self.stats.add_execute(start_time);
                        self.state = ConnectionState::Clean;
                        let affected_rows = pp.get_lenenc().loc("affected_rows")?;
                        let last_insert_id = pp.get_lenenc().loc("last_insert_id")?;
                        Ok(ExecuteResult {
                            affected_rows,
                            last_insert_id,
                        })
                    }
                    v => {
                        self.state = ConnectionState::Broken;
                        Err(ConnectionErrorContent::ProtocolError(format!(
                            "Unexpected response type {v} to row package"
                        ))
                        .into())
                    }
                }
            }
        }
    }

    /// Close a prepared statement
    fn close_prepared_statement(
        &mut self,
        id: u32,
    ) -> impl Future<Output = ConnectionResult<()>> + Send {
        assert!(matches!(self.state, ConnectionState::Clean));
        self.writer.seq = 0;
        let mut p = self.writer.compose();
        p.put_u8(com::STMT_CLOSE);
        p.put_u32(id);
        p.finalize();
        self.state = ConnectionState::ClosePreparedStatement;
        async move {
            let start_time = self.stats.get_instant();
            self.test_cancel()?;
            self.writer.send().await?;
            self.stats.add_prepare(start_time);
            self.state = ConnectionState::Clean;
            Ok(())
        }
    }

    /// Ping connection
    fn ping(&mut self) -> impl Future<Output = ConnectionResult<()>> + Send {
        assert!(matches!(self.state, ConnectionState::Clean));
        self.writer.seq = 0;
        let mut p = self.writer.compose();
        p.put_u8(com::PING);
        p.finalize();
        self.state = ConnectionState::PingSend;

        async move {
            self.test_cancel()?;
            self.writer.send().await?;

            self.state = ConnectionState::PingRecv;
            self.test_cancel()?;
            let package = self.reader.read().await?;
            {
                let mut pp = PackageParser::new(package);
                match pp.get_u8().loc("first_byte")? {
                    255 => {
                        handle_mysql_error(&mut pp)?;
                        unreachable!()
                    }
                    0 => {
                        self.state = ConnectionState::Clean;
                        Ok(())
                    }
                    v => {
                        self.state = ConnectionState::Broken;
                        Err(ConnectionErrorContent::ProtocolError(format!(
                            "Unexpected response type {v} to ping"
                        ))
                        .into())
                    }
                }
            }
        }
    }
}

/// A connection to Mariadb/Mysql
pub struct Connection {
    /// Hash map of prepared statements.
    ///
    /// Note currently we do not clean up any prepared statements. In the future this will be turned into a LRU
    prepared_statements: LRUCache<Statement>,

    /// Single prepared statement not cached
    prepared_statement: Option<Statement>,
    /// Underlying raw connection
    raw: RawConnection,
    /// The current transaction depth
    transaction_depth: usize,
    /// The number of transactions to drop after cleanup
    cleanup_rollbacks: usize,
}

/// A query to Mariadb/Mysql
pub struct Query<'a> {
    /// The connection to make the query on
    connection: &'a mut RawConnection,
    /// The statement to execute
    statement: &'a Statement,
    /// The next param to bind
    cur_param: u16,
    /// Offset in the writer where the null array should be written
    null_offset: usize,
    /// Offset  in the writer where the types/unsigned should be written
    type_offset: usize,
}

impl<'a> Query<'a> {
    /// Get information about the prepared statement, if the [QueryOptions::information] was enabled
    #[inline]
    pub fn information(&self) -> Option<&StatementInformation> {
        self.statement.information.as_ref()
    }

    /// Bind the next argument to the query
    #[inline]
    pub fn bind<T: Bind + ?Sized>(mut self, v: &T) -> ConnectionResult<Self> {
        if self.cur_param == self.statement.num_params {
            return Err(ConnectionErrorContent::Bind(
                self.cur_param,
                BindError::TooManyArgumentsBound,
            )
            .into());
        }
        let mut w = crate::bind::Writer::new(&mut self.connection.writer.buff);
        if !v
            .bind(&mut w)
            .map_err(|e| ConnectionErrorContent::Bind(self.cur_param, e))?
        {
            let w = self.cur_param / 8;
            let b = self.cur_param % 8;
            self.connection.writer.buff[self.null_offset + w as usize] |= 1 << b;
        }

        self.connection.writer.buff[self.type_offset + (self.cur_param * 2) as usize] = T::TYPE;
        if T::UNSIGNED {
            self.connection.writer.buff[self.type_offset + (self.cur_param * 2) as usize + 1] = 128;
        }
        self.cur_param += 1;
        Ok(self)
    }

    /// Execute the query and return zero or one mapped rows
    ///
    /// All arguments must have been bound
    ///
    /// If the query returns more than one row an error is returned
    pub async fn fetch_optional_map<M: RowMap<'a>>(self) -> Result<Option<M::T>, M::E> {
        if self.cur_param != self.statement.num_params {
            return Err(ConnectionError::from(ConnectionErrorContent::Bind(
                self.cur_param,
                BindError::TooFewArgumentsBound,
            ))
            .into());
        }
        match self.connection.query_send().await? {
            QueryResult::WithColumns => (),
            QueryResult::ExecuteResult(_) => {
                return Err(ConnectionError::from(ConnectionErrorContent::ExpectedRows).into());
            }
        }

        let start_instant = self.connection.stats.get_instant();
        // safety-cancel: The cleanup on the connection will skip the remaining rows
        self.connection.test_cancel()?;
        let p1 = self.connection.reader.read_raw().await?;
        {
            let mut pp = PackageParser::new(self.connection.reader.bytes(p1.clone()));
            match pp.get_u8().loc("Row first byte")? {
                0x00 => (),
                0xFE => {
                    //EOD
                    self.connection.state = ConnectionState::Clean;
                    return Ok(None);
                }
                0xFF => {
                    handle_mysql_error(&mut pp)?;
                    unreachable!()
                }
                v => {
                    return Err(ConnectionError::from(ConnectionErrorContent::ProtocolError(
                        format!("Unexpected response type {v} to row package"),
                    ))
                    .into());
                }
            }
        }

        // We need to keep two packages in memory, cleanup will unset this bool
        self.connection.reader.buffer_packages = true;

        // safety-cancel: The cleanup on the connection will skip the remaining rows
        self.connection.test_cancel()?;
        let p2 = self.connection.reader.read_raw().await?;
        {
            let mut pp = PackageParser::new(self.connection.reader.bytes(p2));
            match pp.get_u8().loc("Row first byte")? {
                0x00 => {
                    return Err(
                        ConnectionError::from(ConnectionErrorContent::UnexpectedRows).into(),
                    );
                }
                0xFE => {
                    self.connection.state = ConnectionState::Clean;
                }
                0xFF => {
                    handle_mysql_error(&mut pp)?;
                    unreachable!()
                }
                v => {
                    return Err(ConnectionError::from(ConnectionErrorContent::ProtocolError(
                        format!("Unexpected response type {v} to row package"),
                    ))
                    .into());
                }
            }
        }

        self.connection.stats.add_fetch(start_instant);
        let row = Row::new(&self.connection.columns, self.connection.reader.bytes(p1));
        Ok(Some(M::map(row)?))
    }

    /// Execute the query and return zero or one rows
    ///
    /// All arguments must have been bound
    ///
    /// If the query returns more than one row an error is returned
    pub fn fetch_optional<T: FromRow<'a>>(
        self,
    ) -> impl Future<Output = ConnectionResult<Option<T>>> + Send {
        self.fetch_optional_map::<FromRowMapper<T>>()
    }

    /// Execute the query and return one row
    ///
    /// All arguments must have been bound
    ///
    /// If the query does not return exactly one row an error is returned
    #[inline]
    pub async fn fetch_one<T: FromRow<'a>>(self) -> ConnectionResult<T> {
        match self.fetch_optional().await? {
            Some(v) => Ok(v),
            None => Err(ConnectionErrorContent::ExpectedRows.into()),
        }
    }

    /// Execute the query and return all mapped rows in a vector
    ///
    /// All arguments must have been bound
    pub async fn fetch_all_map<M: RowMap<'a>>(self) -> Result<Vec<M::T>, M::E> {
        if self.cur_param != self.statement.num_params {
            return Err(ConnectionError::from(ConnectionErrorContent::Bind(
                self.cur_param,
                BindError::TooFewArgumentsBound,
            ))
            .into());
        }
        let start_instant = self.connection.stats.get_instant();
        match self.connection.query_send().await? {
            QueryResult::WithColumns => (),
            QueryResult::ExecuteResult(_) => {
                return Err(ConnectionError::from(ConnectionErrorContent::ExpectedRows).into());
            }
        };

        self.connection.ranges.clear();
        loop {
            // safety-cancel: The cleanup on the connection will skip the remaining rows
            self.connection.test_cancel()?;
            let p = self.connection.reader.read_raw().await?;
            {
                let mut pp = PackageParser::new(self.connection.reader.bytes(p.clone()));
                match pp.get_u8().loc("Row first byte")? {
                    0x00 => self.connection.ranges.push(p),
                    0xFE => {
                        //EOD
                        self.connection.state = ConnectionState::Clean;
                        break;
                    }
                    0xFF => {
                        handle_mysql_error(&mut pp)?;
                        unreachable!()
                    }
                    v => {
                        return Err(ConnectionError::from(ConnectionErrorContent::ProtocolError(
                            format!("Unexpected response type {v} to row package"),
                        ))
                        .into());
                    }
                }
            }

            // We need to keep two packages in memory, cleanup will unset this bool
            self.connection.reader.buffer_packages = true;
        }

        self.connection.stats.add_fetch(start_instant);
        let mut ans = Vec::with_capacity(self.connection.ranges.len());
        for p in &self.connection.ranges {
            let row = Row::new(
                &self.connection.columns,
                self.connection.reader.bytes(p.clone()),
            );
            ans.push(M::map(row)?);
        }
        Ok(ans)
    }

    /// Execute the query and return all rows in a vector
    ///
    /// All arguments must have been bound
    pub fn fetch_all<T: FromRow<'a>>(
        self,
    ) -> impl Future<Output = ConnectionResult<Vec<T>>> + Send {
        self.fetch_all_map::<FromRowMapper<T>>()
    }

    /// Execute the query and return an iterator that can return the results
    pub async fn fetch(self) -> ConnectionResult<QueryIterator<'a>> {
        if self.cur_param != self.statement.num_params {
            return Err(ConnectionErrorContent::Bind(
                self.cur_param,
                BindError::TooFewArgumentsBound,
            )
            .into());
        }
        match self.connection.query_send().await? {
            QueryResult::ExecuteResult(_) => Err(ConnectionErrorContent::ExpectedRows.into()),
            QueryResult::WithColumns => Ok(QueryIterator {
                connection: self.connection,
            }),
        }
    }

    /// Execute the query and return an iterator that can return the mapped results
    pub async fn fetch_map<M>(self) -> ConnectionResult<MapQueryIterator<'a, M>>
    where
        for<'b> M: RowMap<'b>,
    {
        if self.cur_param != self.statement.num_params {
            return Err(ConnectionErrorContent::Bind(
                self.cur_param,
                BindError::TooFewArgumentsBound,
            )
            .into());
        }
        match self.connection.query_send().await? {
            QueryResult::ExecuteResult(_) => Err(ConnectionErrorContent::ExpectedRows.into()),
            QueryResult::WithColumns => Ok(MapQueryIterator {
                connection: self.connection,
                _phantom: Default::default(),
            }),
        }
    }

    /// Execute query that does not return any columns
    pub async fn execute(self) -> ConnectionResult<ExecuteResult> {
        if self.cur_param != self.statement.num_params {
            return Err(ConnectionErrorContent::Bind(
                self.cur_param,
                BindError::TooFewArgumentsBound,
            )
            .into());
        }
        match self.connection.query_send().await? {
            QueryResult::WithColumns => Err(ConnectionErrorContent::UnexpectedRows.into()),
            QueryResult::ExecuteResult(v) => Ok(v),
        }
    }
}

/// Represents an ongoing transaction in the connection
///
/// Note: Since rust does not support async drops. Dropping
/// a transaction object will not roll back the transaction
/// immediately. This will instead be deferred to next time
/// the connection is used.
pub struct Transaction<'a> {
    /// The underlying connection we have started a transaction on
    connection: &'a mut Connection,
}

impl<'a> Transaction<'a> {
    /// Commit this traction to the database
    ///
    /// If the returned future is dropped. The transaction will
    /// be rolled back or committed the next time the underlying
    /// connection is used
    pub async fn commit(self) -> ConnectionResult<()> {
        self.connection.cleanup().await?;
        let mut this = ManuallyDrop::new(self);
        this.connection.commit_impl().await?;
        Ok(())
    }

    /// Commit this traction to the database
    ///
    /// If the returned future is dropped. The transaction will
    /// be rolled back connection is used
    pub async fn rollback(self) -> ConnectionResult<()> {
        self.connection.cleanup().await?;
        let mut this = ManuallyDrop::new(self);
        this.connection.rollback_impl().await?;
        Ok(())
    }
}

impl<'a> Executor for Transaction<'a> {
    #[inline]
    fn query_raw(
        &mut self,
        stmt: Cow<'static, str>,
        options: QueryOptions,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> + Send {
        self.connection.query_inner(stmt, options)
    }

    #[inline]
    fn begin(&mut self) -> impl Future<Output = ConnectionResult<Transaction<'_>>> + Send {
        self.connection.begin_impl()
    }

    #[inline]
    fn query_with_args_raw(
        &mut self,
        stmt: Cow<'static, str>,
        options: QueryOptions,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> {
        self.connection.query_with_args_raw(stmt, options, args)
    }

    #[inline]
    fn execute_unprepared(
        &mut self,
        stmt: &str,
    ) -> impl Future<Output = ConnectionResult<ExecuteResult>> + Send {
        self.connection.execute_unprepared(stmt)
    }

    #[inline]
    fn ping(&mut self) -> impl Future<Output = ConnectionResult<()>> + Send {
        self.connection.ping()
    }
}

impl<'a> Drop for Transaction<'a> {
    fn drop(&mut self) {
        // Register in connection to drop transaction in cleanup
        self.connection.cleanup_rollbacks += 1;
    }
}

/// Trait implemented by [Connection] and [Transaction] that facilitates executing queries or creating new transactions
pub trait Executor: Sized + Send {
    /// Execute a query on the connection
    ///
    /// If the returned feature is dropped, or if the returned [Query] or [QueryIterator] is dropped,
    /// the connection will be left in a unclean state. Where the query can be in a half finished stare.
    ///
    /// If this is the case [Connection::is_clean] will return false. A call to [Connection::cleanup] will finish up the
    /// query as quickly as possibly.
    ///
    /// If query is called while [Connection::is_clean] is false, query will call [Connection::cleanup] before executing the next
    /// query
    fn query_raw(
        &mut self,
        stmt: Cow<'static, str>,
        options: QueryOptions,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> + Send;

    /// Execute a query on the connection with the given arguments
    ///
    /// This is a shortcut for
    /// ```ignore
    /// args.bind_args(e.query_raw(stmt).await?)?;
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn query_with_args_raw(
        &mut self,
        stmt: Cow<'static, str>,
        options: QueryOptions,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> + Send;

    /// Execute a statement directly using the mysql text protocol
    ///
    /// Note currently we cannot recover the session if the result returns rows
    fn execute_unprepared(
        &mut self,
        stmt: &str,
    ) -> impl Future<Output = ConnectionResult<ExecuteResult>> + Send;

    /// Begin a new transaction (or Save point)
    ///
    /// If the returned future is dropped either now transaction will have been created, or it will be
    /// dropped again [Connection::cleanup]
    fn begin(&mut self) -> impl Future<Output = ConnectionResult<Transaction<'_>>> + Send;

    /// Perform ping
    fn ping(&mut self) -> impl Future<Output = ConnectionResult<()>> + Send;
}

/// Add helper methods to Executor to facilitate common operations
pub trait ExecutorExt {
    /// Execute a query on the connection
    ///
    /// If the returned feature is dropped, or if the returned [Query] or [QueryIterator] is dropped,
    /// the connection will be left in a unclean state. Where the query can be in a half finished stare.
    ///
    /// If this is the case [Connection::is_clean] will return false. A call to [Connection::cleanup] will finish up the
    /// query as quickly as possibly.
    ///
    /// If query is called while [Connection::is_clean] is false, query will call [Connection::cleanup] before executing the next
    /// query
    fn query(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> + Send;

    /// Execute a query on the connection
    ///
    /// If the returned feature is dropped, or if the returned [Query] or [QueryIterator] is dropped,
    /// the connection will be left in a unclean state. Where the query can be in a half finished stare.
    ///
    /// If this is the case [Connection::is_clean] will return false. A call to [Connection::cleanup] will finish up the
    /// query as quickly as possibly.
    ///
    /// If query is called while [Connection::is_clean] is false, query will call [Connection::cleanup] before executing the next
    /// query
    fn query_with_options(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
        options: QueryOptions,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> + Send;

    /// Execute a query on the connection with arguments
    ///
    /// If the `list_hack` feature is enabled any instance of _LIST_ in the query
    /// is replaced by an appropriate number of comma separated question marks
    /// corresponding to the supplied [crate::bind::list] argument.
    ///
    /// If the returned feature is dropped, or if the returned [Query] or [QueryIterator] is dropped,
    /// the connection will be left in a unclean state. Where the query can be in a half finished stare.
    ///
    /// If this is the case [Connection::is_clean] will return false. A call to [Connection::cleanup] will finish up the
    /// query as quickly as possibly.
    ///
    /// If query is called while [Connection::is_clean] is false, query will call [Connection::cleanup] before executing the next
    /// query
    fn query_with_args(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> + Send;

    /// Execute a query with the given arguments and return all rows as a vector
    ///
    /// This is a shortcut for
    /// ```ignore
    /// query_with_args(stmt, args).await?.fetch_all().await?
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn fetch_all<'a, T: FromRow<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<Vec<T>>> + Send;

    /// Execute a query with the given arguments and return all rows mapped to a vector vector
    ///
    /// This is a shortcut for
    /// ```ignore
    /// query_with_args(stmt, args).await?.fetch_all_map(map).await?
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn fetch_all_map<'a, M: RowMap<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = Result<Vec<M::T>, M::E>> + Send;

    /// Execute a query with the given arguments and return one row
    ///
    /// This is a shortcut for
    /// ```ignore
    /// query_with_args(stmt, args).await?.fetch_one().await?
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn fetch_one<'a, T: FromRow<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<T>> + Send;

    /// Execute a query with the given arguments and return one mapped row
    ///
    /// This is a shortcut for
    /// ```ignore
    /// query_with_args(stmt, args).await?.fetch_one_map(map).await?
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn fetch_one_map<'a, M: RowMap<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = Result<M::T, M::E>> + Send;

    /// Execute a query with the given arguments are return an optional row
    ///
    /// This is a shortcut for
    /// ```ignore
    /// query_with_args(stmt, args).await?.fetch_optional().await?
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn fetch_optional<'a, T: FromRow<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<Option<T>>> + Send;

    /// Execute a query with the given arguments are return an optional mapped row
    ///
    /// This is a shortcut for
    /// ```ignore
    /// query_with_args(stmt, args).await?.fetch_optional_map(map).await?
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn fetch_optional_map<'a, M: RowMap<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = Result<Option<M::T>, M::E>> + Send;

    /// Executing a query with the given arg
    ///
    /// This is a shortcut for
    /// ```ignore
    /// query_with_args(stmt, args).await?.execute().await?
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn execute(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<ExecuteResult>> + Send;

    /// Execute a query with the given arguments and stream the results back
    ///
    /// This is a shortcut for
    /// ```ignore
    /// query_with_args(stmt, args).await?.fetch().await?
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn fetch(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<QueryIterator<'_>>> + Send;

    /// Execute a query with the given arguments and stream the mapped results back
    ///
    /// This is a shortcut for
    /// ```ignore
    /// query_with_args(stmt, args).await?.fetch_map(map).await?
    /// ```
    ///
    /// See [Executor::query_raw] for cancel/drop safety
    fn fetch_map<'a, M>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<MapQueryIterator<'a, M>>> + Send
    where
        for<'b> M: RowMap<'b>;
}

/// Implement [ExecutorExt::fetch_all] without stmt as a generic
async fn fetch_all_impl<'a, E: Executor + Sized + Send, T: FromRow<'a>>(
    e: &'a mut E,
    stmt: Cow<'static, str>,
    args: impl Args + Send,
) -> ConnectionResult<Vec<T>> {
    let q = e
        .query_with_args_raw(stmt, QueryOptions::new(), args)
        .await?;
    q.fetch_all().await
}

/// Implement [ExecutorExt::fetch_all_map] without stmt as a generic
async fn fetch_all_map_impl<'a, E: Executor + Sized + Send, M: RowMap<'a>>(
    e: &'a mut E,
    stmt: Cow<'static, str>,
    args: impl Args + Send,
) -> Result<Vec<M::T>, M::E> {
    let q = e
        .query_with_args_raw(stmt, QueryOptions::new(), args)
        .await?;
    q.fetch_all_map::<M>().await
}

/// Implement [ExecutorExt::fetch_one] without stmt as a generic
async fn fetch_one_impl<'a, E: Executor + Sized + Send, T: FromRow<'a>>(
    e: &'a mut E,
    stmt: Cow<'static, str>,
    args: impl Args + Send,
) -> ConnectionResult<T> {
    let q = e
        .query_with_args_raw(stmt, QueryOptions::new(), args)
        .await?;
    match q.fetch_optional().await? {
        Some(v) => Ok(v),
        None => Err(ConnectionErrorContent::ExpectedRows.into()),
    }
}

/// Implement [ExecutorExt::fetch_one_map] without stmt as a generic
async fn fetch_one_map_impl<'a, E: Executor + Sized + Send, M: RowMap<'a>>(
    e: &'a mut E,
    stmt: Cow<'static, str>,
    args: impl Args + Send,
) -> Result<M::T, M::E> {
    let q = e
        .query_with_args_raw(stmt, QueryOptions::new(), args)
        .await
        .map_err(M::E::from)?;
    match q.fetch_optional_map::<M>().await? {
        Some(v) => Ok(v),
        None => Err(ConnectionError::from(ConnectionErrorContent::ExpectedRows).into()),
    }
}

/// Implement [ExecutorExt::fetch_optional] without stmt as a generic
async fn fetch_optional_impl<'a, E: Executor + Sized + Send, T: FromRow<'a>>(
    e: &'a mut E,
    stmt: Cow<'static, str>,
    args: impl Args + Send,
) -> ConnectionResult<Option<T>> {
    let q = e
        .query_with_args_raw(stmt, QueryOptions::new(), args)
        .await?;
    q.fetch_optional().await
}

/// Implement [ExecutorExt::fetch_optional_map] without stmt as a generic
async fn fetch_optional_map_impl<'a, E: Executor + Sized + Send, M: RowMap<'a>>(
    e: &'a mut E,
    stmt: Cow<'static, str>,
    args: impl Args + Send,
) -> Result<Option<M::T>, M::E> {
    let q = e
        .query_with_args_raw(stmt, QueryOptions::new(), args)
        .await?;
    q.fetch_optional_map::<M>().await
}

/// Implement [ExecutorExt::execute] without stmt as a generic
async fn execute_impl<E: Executor + Sized + Send>(
    e: &mut E,
    stmt: Cow<'static, str>,
    args: impl Args + Send,
) -> ConnectionResult<ExecuteResult> {
    let q = e
        .query_with_args_raw(stmt, QueryOptions::new(), args)
        .await?;
    q.execute().await
}

/// Implement [ExecutorExt::fetch] without stmt as a generic
async fn fetch_impl<'a, E: Executor + Sized + Send>(
    e: &'a mut E,
    stmt: Cow<'static, str>,
    args: impl Args + Send,
) -> ConnectionResult<QueryIterator<'a>> {
    let q = e
        .query_with_args_raw(stmt, QueryOptions::new(), args)
        .await?;
    q.fetch().await
}

/// Implement [ExecutorExt::fetch_map] without stmt as a generic
async fn fetch_map_impl<'a, E: Executor + Sized + Send, M>(
    e: &'a mut E,
    stmt: Cow<'static, str>,
    args: impl Args + Send,
) -> ConnectionResult<MapQueryIterator<'a, M>>
where
    for<'b> M: RowMap<'b>,
{
    let q = e
        .query_with_args_raw(stmt, QueryOptions::new(), args)
        .await?;
    q.fetch_map::<M>().await
}

impl<E: Executor + Sized + Send> ExecutorExt for E {
    #[inline]
    fn query(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> + Send {
        self.query_raw(stmt.into(), QueryOptions::new())
    }

    #[inline]
    fn query_with_options(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
        options: QueryOptions,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> + Send {
        self.query_raw(stmt.into(), options)
    }

    #[inline]
    fn query_with_args(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> {
        self.query_with_args_raw(stmt.into(), QueryOptions::new(), args)
    }

    #[inline]
    fn fetch_all<'a, T: FromRow<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<Vec<T>>> + Send {
        fetch_all_impl(self, stmt.into(), args)
    }

    #[inline]
    fn fetch_all_map<'a, M: RowMap<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = Result<Vec<M::T>, M::E>> + Send {
        fetch_all_map_impl::<E, M>(self, stmt.into(), args)
    }

    #[inline]
    fn fetch_one<'a, T: FromRow<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<T>> + Send {
        fetch_one_impl(self, stmt.into(), args)
    }

    #[inline]
    fn fetch_one_map<'a, M: RowMap<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = Result<M::T, M::E>> + Send {
        fetch_one_map_impl::<E, M>(self, stmt.into(), args)
    }

    #[inline]
    fn fetch_optional<'a, T: FromRow<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<Option<T>>> + Send {
        fetch_optional_impl(self, stmt.into(), args)
    }

    #[inline]
    fn fetch_optional_map<'a, M: RowMap<'a>>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = Result<Option<M::T>, M::E>> + Send {
        fetch_optional_map_impl::<E, M>(self, stmt.into(), args)
    }

    #[inline]
    fn execute(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<ExecuteResult>> + Send {
        execute_impl(self, stmt.into(), args)
    }

    #[inline]
    fn fetch(
        &mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<QueryIterator<'_>>> + Send {
        fetch_impl(self, stmt.into(), args)
    }

    #[inline]
    fn fetch_map<'a, M>(
        &'a mut self,
        stmt: impl Into<Cow<'static, str>>,
        args: impl Args + Send,
    ) -> impl Future<Output = ConnectionResult<MapQueryIterator<'a, M>>> + Send
    where
        for<'b> M: RowMap<'b>,
    {
        fetch_map_impl::<E, M>(self, stmt.into(), args)
    }
}

/// Options for preparing queries
pub struct QueryOptions {
    /// Should we cache the prepared statement
    cache: bool,
    /// Should be store information about columns and parameters
    information: bool,
}

impl QueryOptions {
    /// Construct default options with caching enabled and column information disabled
    pub fn new() -> Self {
        Default::default()
    }

    /// Add the prepared statement to the prepared statement cache
    pub fn cache(self, enable: bool) -> Self {
        QueryOptions {
            cache: enable,
            ..self
        }
    }

    /// Store type information about columns and parameters
    pub fn information(self, enable: bool) -> Self {
        QueryOptions {
            information: enable,
            ..self
        }
    }
}

impl Default for QueryOptions {
    fn default() -> Self {
        Self {
            cache: true,
            information: false,
        }
    }
}

impl Connection {
    /// Connect to Mariadb/Mysql
    pub async fn connect(options: &ConnectionOptions<'_>) -> ConnectionResult<Self> {
        let raw = RawConnection::connect(options).await?;
        Ok(Connection {
            raw,
            prepared_statements: LRUCache::new(options.statement_case_size),
            transaction_depth: 0,
            cleanup_rollbacks: 0,
            prepared_statement: None,
        })
    }

    /// Return false if there are partially execute queries in the connection
    pub fn is_clean(&self) -> bool {
        matches!(self.raw.state, ConnectionState::Clean)
    }

    /// Finish up any partially execute queries as quickly as possible
    pub async fn cleanup(&mut self) -> ConnectionResult<()> {
        self.raw.cleanup().await?;

        assert!(self.cleanup_rollbacks <= self.transaction_depth);
        if self.cleanup_rollbacks != 0 {
            let statement =
                rollback_transaction_query(self.transaction_depth - self.cleanup_rollbacks);
            // Once raw.query has been called we will have roled back to this transaction level once raw.cleanup succeed
            self.transaction_depth -= self.cleanup_rollbacks;
            self.cleanup_rollbacks = 0;
            self.raw.execute_unprepared(statement).await?;
        }

        if let Some(v) = self.prepared_statement.take() {
            self.raw.close_prepared_statement(v.stmt_id).await?
        }
        Ok(())
    }

    /// Execute query. This inner method exists because [Self::query] is template on the stmt type
    /// but we would like only one instantiation.
    async fn query_inner(
        &mut self,
        stmt: Cow<'static, str>,
        options: QueryOptions,
    ) -> ConnectionResult<Query<'_>> {
        self.cleanup().await?;
        if !options.cache {
            let r = self.raw.prepare_query(&stmt, options.information).await?;
            self.prepared_statement = Some(r);
            Ok(self.raw.query(self.prepared_statement.as_ref().unwrap()))
        } else {
            let statement = match self.prepared_statements.entry(stmt) {
                Entry::Occupied(mut e) => {
                    if e.get().information.is_none() && options.information {
                        // Replace old cached statement without info with a new one with information
                        let r = self.raw.prepare_query(e.key(), options.information).await?;
                        let old = e.insert(r);
                        self.raw.close_prepared_statement(old.stmt_id).await?
                    }
                    e.bump();
                    e.into_mut()
                }
                Entry::Vacant(e) => {
                    let r = self.raw.prepare_query(e.key(), options.information).await?;
                    let (r, old) = e.insert(r);
                    if let Some((_, old)) = old {
                        self.raw.close_prepared_statement(old.stmt_id).await?
                    }
                    r
                }
            };
            Ok(self.raw.query(statement))
        }
    }

    /// Begin a new transaction or save-point
    async fn begin_impl(&mut self) -> ConnectionResult<Transaction<'_>> {
        self.cleanup().await?;

        assert_eq!(self.cleanup_rollbacks, 0); // cleanup_rollback will be 0 after cleanup

        // Once we call query the state will be such that once raw.cleanup has been called
        // there will be one more transaction level
        let q = begin_transaction_query(self.transaction_depth);
        self.transaction_depth += 1;
        self.cleanup_rollbacks = 1;
        self.raw.execute_unprepared(q).await?;

        // The execute has now succeeded so there is no need to role back the transaction
        assert_eq!(self.cleanup_rollbacks, 1);
        self.cleanup_rollbacks = 0;
        Ok(Transaction { connection: self })
    }

    /// Ping the server to ensure the connection is still good
    async fn ping_impl(&mut self) -> ConnectionResult<()> {
        self.cleanup().await?;
        self.raw.ping().await?;
        Ok(())
    }

    /// Rollback the top most transaction or save point
    fn rollback_impl(&mut self) -> impl Future<Output = ConnectionResult<ExecuteResult>> + Send {
        assert!(matches!(self.raw.state, ConnectionState::Clean));
        assert_eq!(self.cleanup_rollbacks, 0);
        assert_ne!(self.transaction_depth, 0);
        self.transaction_depth -= 1;

        // Once we call query the state will be such that once raw.cleanup has been called
        // there will be one less transaction
        self.raw
            .execute_unprepared(rollback_transaction_query(self.transaction_depth))
    }

    /// Commit the top most transaction or save point
    fn commit_impl(&mut self) -> impl Future<Output = ConnectionResult<ExecuteResult>> + Send {
        assert!(matches!(self.raw.state, ConnectionState::Clean));
        assert_eq!(self.cleanup_rollbacks, 0);
        assert_ne!(self.transaction_depth, 0);

        self.transaction_depth -= 1;

        // Once we call query the state will be such that once raw.cleanup has been called
        // there will be one less transaction
        self.raw
            .execute_unprepared(commit_transaction_query(self.transaction_depth))
    }

    #[cfg(feature = "cancel_testing")]
    #[doc(hidden)]
    /// Set the cancel counts for testing
    pub fn set_cancel_count(&mut self, cnt: Option<usize>) {
        self.raw.cancel_count = cnt;
    }

    #[cfg(feature = "stats")]
    /// Return statistics for the connection
    ///
    /// Only available if the "stats" feature is enabled
    pub fn stats(&self) -> &Stats {
        &self.raw.stats
    }

    /// Clear the connection stats
    ///
    /// Only available if the "stats" feature is enabled
    #[cfg(feature = "stats")]
    pub fn clear_stats(&mut self) {
        self.raw.stats = Default::default()
    }
}

#[cfg(feature = "list_hack")]
/// Convert _LIST_ instances in a query to a sequence of ?,?,..,? or NULL depending on
/// the counts in lengths
fn convert_list_query(
    stmt: Cow<'static, str>,
    lengths: &[usize],
) -> ConnectionResult<Cow<'static, str>> {
    if let Some((head, tail)) = stmt.split_once("_LIST_") {
        let mut stmt = String::with_capacity(stmt.len() + 2 * lengths.iter().sum::<usize>());
        stmt.push_str(head);
        let mut len_it = lengths.iter();
        for part in tail.split("_LIST_") {
            let Some(len) = len_it.next() else {
                return Err(ConnectionErrorContent::TooFewListArguments.into());
            };
            if *len == 0 {
                stmt.push_str("NULL");
            } else {
                for i in 0..*len {
                    if i == 0 {
                        stmt.push('?');
                    } else {
                        stmt.push_str(", ?");
                    }
                }
            }
            stmt.push_str(part);
        }
        if len_it.next().is_some() {
            return Err(ConnectionErrorContent::TooManyListArguments.into());
        }
        Ok(stmt.into())
    } else {
        if !lengths.is_empty() {
            return Err(ConnectionErrorContent::TooManyListArguments.into());
        }
        Ok(stmt)
    }
}

impl Executor for Connection {
    #[inline]
    fn query_raw(
        &mut self,
        stmt: Cow<'static, str>,
        options: QueryOptions,
    ) -> impl Future<Output = ConnectionResult<Query<'_>>> + Send {
        self.query_inner(stmt, options)
    }

    #[inline]
    fn begin(&mut self) -> impl Future<Output = ConnectionResult<Transaction<'_>>> + Send {
        self.begin_impl()
    }

    #[inline]
    async fn query_with_args_raw(
        &mut self,
        stmt: Cow<'static, str>,
        options: QueryOptions,
        args: impl Args + Send,
    ) -> ConnectionResult<Query<'_>> {
        #[cfg(feature = "list_hack")]
        let stmt = {
            self.raw.list_lengths.clear();
            args.list_lengths(&mut self.raw.list_lengths);
            convert_list_query(stmt, &self.raw.list_lengths)?
        };
        self.cleanup().await?;
        let query = self.query_inner(stmt, options).await?;
        args.bind_args(query)
    }

    async fn execute_unprepared(&mut self, stmt: &str) -> ConnectionResult<ExecuteResult> {
        self.cleanup().await?;
        self.raw.execute_unprepared(stmt.into()).await
    }

    #[inline]
    fn ping(&mut self) -> impl Future<Output = ConnectionResult<()>> + Send {
        self.ping_impl()
    }
}