rs-bag 0.1.0

use cdr::CdrBe;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use turso::Row;

use super::Message;
use crate::Error;
use crate::error::Result;

#[derive(Debug)]
pub struct TypedMessage<T: serde::Serialize + for<'de> serde::Deserialize<'de>> {
    pub id: i64,
    pub topic_id: i64,
    pub timestamp: i64,
    pub data: T,
}

impl<T> Message for TypedMessage<T> where T: Serialize + for<'de> Deserialize<'de> {}

impl<T: for<'de> Deserialize<'de>> Serialize for TypedMessage<T>
where
    T: Serialize + for<'a> Deserialize<'a>,
{
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        use serde::ser::SerializeStruct;

        // Todo: Replace by correct endianess
        let data_bytes: Vec<u8> = cdr::serialize::<_, _, CdrBe>(&self.data, cdr::size::Infinite)
            .map_err(serde::ser::Error::custom)?;

        let mut s = serializer.serialize_struct("Message", 4)?;
        s.serialize_field("id", &self.id)?;
        s.serialize_field("topic_id", &self.topic_id)?;
        s.serialize_field("timestamp", &self.timestamp)?;
        s.serialize_field("data", &data_bytes)?;
        s.end()
    }
}

impl<'de, T> Deserialize<'de> for TypedMessage<T>
where
    T: Serialize + for<'a> Deserialize<'a>,
{
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        #[derive(Deserialize)]
        #[serde(field_identifier, rename_all = "snake_case")]
        enum Field {
            Id,
            TopicId,
            Timestamp,
            Data,
        }

        struct Visitor<T>(std::marker::PhantomData<T>);

        impl<'de, T> serde::de::Visitor<'de> for Visitor<T>
        where
            T: Serialize + for<'a> Deserialize<'a>,
        {
            type Value = TypedMessage<T>;

            fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
                f.write_str("struct Message")
            }

            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
            where
                A: serde::de::MapAccess<'de>,
            {
                let mut id = None;
                let mut topic_id = None;
                let mut timestamp = None;
                let mut data: Option<Vec<u8>> = None;

                while let Some(key) = map.next_key()? {
                    match key {
                        Field::Id => id = Some(map.next_value()?),
                        Field::TopicId => topic_id = Some(map.next_value()?),
                        Field::Timestamp => timestamp = Some(map.next_value()?),
                        Field::Data => data = Some(map.next_value()?),
                    }
                }

                let id = id.ok_or_else(|| serde::de::Error::missing_field("id"))?;
                let topic_id =
                    topic_id.ok_or_else(|| serde::de::Error::missing_field("topic_id"))?;
                let timestamp =
                    timestamp.ok_or_else(|| serde::de::Error::missing_field("timestamp"))?;
                let data_bytes = data.ok_or_else(|| serde::de::Error::missing_field("data"))?;

                let data =
                    cdr::deserialize::<T>(&data_bytes[..]).map_err(serde::de::Error::custom)?;

                Ok(TypedMessage {
                    id,
                    topic_id,
                    timestamp,
                    data,
                })
            }
        }

        const FIELDS: &[&str] = &["id", "topic_id", "timestamp", "data"];
        deserializer.deserialize_struct("Message", FIELDS, Visitor(std::marker::PhantomData))
    }
}

impl<T: Serialize + for<'de> Deserialize<'de>> TypedMessage<T> {
    pub fn from_row_cdr(row: Row) -> Result<Self> {
        let id = row
            .get_value(0)?
            .as_integer()
            .copied()
            .ok_or(Error::row_conversion("id", "an integer"))?;

        let timestamp = row
            .get_value(1)?
            .as_integer()
            .copied()
            .ok_or(Error::row_conversion("timestamp", "an integer"))?;

        let topic_id = row
            .get_value(2)?
            .as_integer()
            .copied()
            .ok_or(Error::row_conversion("topic_id", "an integer"))?;

        let data_bytes = super::raw::take_blob(row.get_value(3)?, "data")?;

        let data = cdr::deserialize::<T>(&data_bytes)?;

        Ok(Self {
            id,
            topic_id,
            timestamp,
            data,
        })
    }
}