ruststream-sqlx 0.7.0

//! What every database gives a by-name subscription: the integer, text and byte columns every
//! sqlx driver reads and binds alike.

use sqlx::error::BoxDynError;
use sqlx::{
    Arguments, ColumnIndex, Database, Decode, Encode, Error, Row, Type, TypeInfo, ValueRef,
};

use super::row::{NamedBytes, NamedId};

/// The integer, text and byte columns a by-name subscription reads and binds: what every sqlx
/// driver decodes and encodes. Machinery; every such database implements it.
#[doc(hidden)]
pub trait RoleColumns: Database {
    /// The value of the column at `ordinal`.
    ///
    /// # Errors
    ///
    /// The driver's error for an ordinal past the row's columns.
    fn value(row: &Self::Row, ordinal: usize) -> Result<Self::ValueRef<'_>, Error>;

    /// An id, as the first of `i64`, `i32`, `i16`, `String` and `Vec<u8>` its column holds, and
    /// which of them the column holds, a bit each in that order.
    ///
    /// # Errors
    ///
    /// A column of none of those types, or the driver's decoding error.
    fn id(value: Self::ValueRef<'_>) -> Result<(NamedId, u8), BoxDynError>;

    /// Whether a column of type `ty` holds an id [`RoleColumns::id`] reads.
    fn holds_id(ty: &Self::TypeInfo) -> bool;

    /// The type an id is declared as.
    fn id_type() -> Self::TypeInfo;

    /// Bytes or text, the first of `Vec<u8>` and `String` the column holds, and which of them it
    /// holds, a bit each in that order.
    ///
    /// # Errors
    ///
    /// A column of neither, or the driver's decoding error.
    fn bytes(value: Self::ValueRef<'_>) -> Result<(NamedBytes, u8), BoxDynError>;

    /// An attempt, from the first of `i64`, `i32` and `i16` its column holds, and which of them
    /// it holds, a bit each in that order; below zero reads as zero.
    ///
    /// # Errors
    ///
    /// A column of none of those types, or the driver's decoding error.
    fn attempt(value: Self::ValueRef<'_>) -> Result<(u64, u8), BoxDynError>;

    /// Binds `id`.
    ///
    /// # Errors
    ///
    /// The driver's encoding error.
    fn bind_id(arguments: &mut Self::Arguments, id: &NamedId) -> Result<(), Error>;
}

impl<DB> RoleColumns for DB
where
    DB: Database,
    usize: ColumnIndex<DB::Row>,
    i16: for<'r> Decode<'r, DB> + for<'q> Encode<'q, DB> + Type<DB>,
    i32: for<'r> Decode<'r, DB> + for<'q> Encode<'q, DB> + Type<DB>,
    i64: for<'r> Decode<'r, DB> + for<'q> Encode<'q, DB> + Type<DB>,
    String: for<'r> Decode<'r, DB> + Type<DB>,
    Vec<u8>: for<'r> Decode<'r, DB> + Type<DB>,
    for<'q> &'q str: Encode<'q, DB> + Type<DB>,
    for<'q> &'q [u8]: Encode<'q, DB> + Type<DB>,
{
    fn value(row: &Self::Row, ordinal: usize) -> Result<Self::ValueRef<'_>, Error> {
        row.try_get_raw(ordinal)
    }

    fn id(value: Self::ValueRef<'_>) -> Result<(NamedId, u8), BoxDynError> {
        let held = {
            let ty = value.type_info();
            let held = bits([
                <i64 as Type<DB>>::compatible(&ty),
                <i32 as Type<DB>>::compatible(&ty),
                <i16 as Type<DB>>::compatible(&ty),
                <String as Type<DB>>::compatible(&ty),
                <Vec<u8> as Type<DB>>::compatible(&ty),
            ]);
            if held == 0 {
                return Err(unread(&*ty, "i64, i32, i16, String or Vec<u8>"));
            }
            held
        };
        let id = match held.trailing_zeros() {
            0 => NamedId::I64(<i64 as Decode<'_, DB>>::decode(value)?),
            1 => NamedId::I32(<i32 as Decode<'_, DB>>::decode(value)?),
            2 => NamedId::I16(<i16 as Decode<'_, DB>>::decode(value)?),
            3 => NamedId::Text(<String as Decode<'_, DB>>::decode(value)?),
            _ => NamedId::Bytes(<Vec<u8> as Decode<'_, DB>>::decode(value)?),
        };
        Ok((id, held))
    }

    fn holds_id(ty: &Self::TypeInfo) -> bool {
        <i64 as Type<DB>>::compatible(ty)
            || <i32 as Type<DB>>::compatible(ty)
            || <i16 as Type<DB>>::compatible(ty)
            || <String as Type<DB>>::compatible(ty)
            || <Vec<u8> as Type<DB>>::compatible(ty)
    }

    fn id_type() -> Self::TypeInfo {
        <i64 as Type<DB>>::type_info()
    }

    fn bytes(value: Self::ValueRef<'_>) -> Result<(NamedBytes, u8), BoxDynError> {
        let held = {
            let ty = value.type_info();
            let held = bits([
                <Vec<u8> as Type<DB>>::compatible(&ty),
                <String as Type<DB>>::compatible(&ty),
            ]);
            if held == 0 {
                return Err(unread(&*ty, "Vec<u8> or String"));
            }
            held
        };
        let bytes = if held.trailing_zeros() == 0 {
            NamedBytes::Bytes(<Vec<u8> as Decode<'_, DB>>::decode(value)?)
        } else {
            NamedBytes::Text(<String as Decode<'_, DB>>::decode(value)?)
        };
        Ok((bytes, held))
    }

    fn attempt(value: Self::ValueRef<'_>) -> Result<(u64, u8), BoxDynError> {
        let held = {
            let ty = value.type_info();
            let held = bits([
                <i64 as Type<DB>>::compatible(&ty),
                <i32 as Type<DB>>::compatible(&ty),
                <i16 as Type<DB>>::compatible(&ty),
            ]);
            if held == 0 {
                return Err(unread(&*ty, "i64, i32 or i16"));
            }
            held
        };
        let attempt = match held.trailing_zeros() {
            0 => <i64 as Decode<'_, DB>>::decode(value)?,
            1 => i64::from(<i32 as Decode<'_, DB>>::decode(value)?),
            _ => i64::from(<i16 as Decode<'_, DB>>::decode(value)?),
        };
        Ok((u64::try_from(attempt).unwrap_or(0), held))
    }

    fn bind_id(arguments: &mut Self::Arguments, id: &NamedId) -> Result<(), Error> {
        match id {
            NamedId::I16(id) => arguments.add(*id),
            NamedId::I32(id) => arguments.add(*id),
            NamedId::I64(id) => arguments.add(*id),
            NamedId::Text(id) => arguments.add(id.as_str()),
            NamedId::Bytes(id) => arguments.add(id.as_slice()),
        }
        .map_err(Error::Encode)
    }
}

/// A bit for each of `held` that is true, the first in the lowest bit.
fn bits<const TYPES: usize>(held: [bool; TYPES]) -> u8 {
    held.iter()
        .rev()
        .fold(0, |bits, &held| (bits << 1) | u8::from(held))
}

/// The error of a column whose type a by-name subscription does not read as `read_as`.
pub(super) fn unread<Info: TypeInfo>(ty: &Info, read_as: &str) -> BoxDynError {
    format!(
        "a by-name subscription reads this column as {read_as}, and it holds {}",
        ty.name()
    )
    .into()
}

impl<DB: RoleColumns> Type<DB> for NamedId {
    fn type_info() -> DB::TypeInfo {
        DB::id_type()
    }

    fn compatible(ty: &DB::TypeInfo) -> bool {
        DB::holds_id(ty)
    }
}

impl<'r, DB: RoleColumns> Decode<'r, DB> for NamedId {
    fn decode(value: DB::ValueRef<'r>) -> Result<Self, BoxDynError> {
        DB::id(value).map(|(id, _)| id)
    }
}

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

    #[test]
    fn bits_follow_the_order_of_the_types() {
        assert_eq!(bits([true, false, true]), 0b101);
        assert_eq!(bits([false, true]), 0b10);
        assert_eq!(bits([false; 5]), 0);
    }
}