mocklogger/

logger1.rs

// Author: D.S. Ljungmark <spider@skuggor.se>, Modio AB
// SPDX-License-Identifier: AGPL-3.0-or-later
use crate::logger::{keys, LogErr};
use crate::timefail;
use crate::types::Metadata;
use crate::types::Unit;
use crate::types::ValueMap;
use crate::values::{
    check_data_row, valid_metric, value_to_string, zbus_row_to_json_string, TimestampRow,
};
use fsipc::logger1::SensorMode;
use modio_logger_db::{Datastore, Metric};
use std::collections::HashMap;
use tracing::{debug, error, info, instrument, warn};
use zbus::zvariant::Value as zValue;
use zbus::{interface, object_server::SignalEmitter};
mod builder;
pub use builder::Builder;

pub struct Logger1 {
    ds: Datastore,
    timefail: timefail::Timefail,
}

impl Logger1 {
    pub async fn new(timefail: timefail::Timefail, ds: Datastore) -> Result<Self, LogErr> {
        if timefail.is_timefail() {
            info!("We are currently in TIMEFAIL mode.");
            let count = ds.transaction_fail_pending().await?;
            if count > 0 {
                info!("Failed {count} pending change requests due to TIMEFAIL");
            }
        }
        Ok(Self { ds, timefail })
    }
    #[must_use]
    pub const fn builder() -> Builder {
        Builder::new()
    }

    #[cfg(test)]
    pub(crate) async fn persist_data(&self) {
        self.ds
            .persist_data()
            .await
            .expect("Failed to persist data");
    }
}

// This is a simple wrapper around the modio_logger_db `DataType`, but since that version does not
// know about "dbus" or how to implement zbus::variant::Type,  we wrap it in a struct here, in a
// pattern called "Newtype" in rust. This lets us add the missing functionality, but not
// re-implement it.
#[derive(Debug, Copy, Clone, serde::Serialize, serde::Deserialize, zbus::zvariant::Type)]
#[zvariant(signature = "s")] // Here we say that it should be a "string" on d-bus and not a struct
                             // or enum.
struct DataTypeWrp(modio_logger_db::DataType);

#[derive(Debug, serde::Serialize, serde::Deserialize, zbus::zvariant::Type)]
struct RowItem {
    name: String,
    datatype: DataTypeWrp,
}

#[must_use]
fn maybe_metric(key: impl Into<String>, val: &zValue<'_>, time: f64) -> Option<Metric> {
    value_to_string(val).map(|value| Metric {
        name: key.into(),
        value,
        time,
    })
}

fn metadata_change_metric() -> Metric {
    let key = "modio.logger.metadata.change";
    assert!(
        keys::valid_key(key).is_ok(),
        "Hardcoded key name should be valid"
    );
    let time = modio_logger_db::fxtime_ms();
    Metric {
        name: key.into(),
        value: "1".into(),
        time,
    }
}

impl Logger1 {
    // When metadata change is succesful, store a key & timestamp and fire off a signal, this
    // allows us to check the "last changed" timestamp in the submitter and decide if we need to
    // read all metadata and see what changes or not.
    async fn metadata_last_changed(
        &self,
        ctxt: &SignalEmitter<'_>,
        key: &str,
    ) -> Result<(), LogErr> {
        let timefail = self.timefail.is_timefail();
        let metric = metadata_change_metric();
        let batch = vec![metric];
        self.ds.insert_bulk(batch, timefail).await?;
        Self::metadata_updated(ctxt, key).await?;
        Ok(())
    }
}

#[allow(clippy::use_self)]
#[interface(name = "se.modio.logger.Logger1")]
impl Logger1 {
    /// Signal sent when new metadata is set
    #[zbus(signal)]
    async fn metadata_updated(ctxt: &SignalEmitter<'_>, key: &str) -> zbus::Result<()>;

    // Should emit "MetadataUpdated"
    #[instrument(skip(self, ctxt))]
    async fn set_metadata_name(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalEmitter<'_>,
        key: &str,
        name: String,
    ) -> Result<(), LogErr> {
        if self
            .ds
            .metadata_set_name(key, &name)
            .await
            .inspect_err(|e| error!("Failed to set metadata name  err={:?}", e))?
        {
            info!("Updated name of key={} to name='{}'", &key, &name);
            self.metadata_last_changed(&ctxt, key).await?;
        }
        Ok(())
    }

    #[instrument(skip(self, ctxt))]
    async fn set_metadata_description(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalEmitter<'_>,
        key: &str,
        description: String,
    ) -> Result<(), LogErr> {
        if self
            .ds
            .metadata_set_description(key, &description)
            .await
            .inspect_err(|e| error!("Failed to set metadata description err={:?}", e))?
        {
            info!(
                "Updated description of key={} to description='{}'",
                &key, &description
            );
            self.metadata_last_changed(&ctxt, key).await?;
        }
        Ok(())
    }

    // Set Metadata for row batches
    //
    // row_items is an array of (Name, Datatype) where both are (currently) strings.
    // The items in a batch array should not change (types or counts) while the names can be.
    // The names will be written to the Description field, and the data-types will be stored
    // elsewhere.
    //
    // Currently, it's all dumped as description to make hacking easier.
    //
    // This should only be used on keys that expect to store rows of correlated, time-stamped data.
    #[instrument(skip(self, ctxt))]
    async fn set_metadata_row(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalEmitter<'_>,
        key: &str,
        row_items: Vec<RowItem>,
    ) -> Result<(), LogErr> {
        // Create an Vec of the "datatypes" in the order they come in.
        let row_metadata: Vec<modio_logger_db::DataType> =
            row_items.iter().map(|x| x.datatype.0).collect();

        // And make a pair of row items into a description string
        let desc = serde_json::to_string(&row_items)?;

        // If we fail to set the row data, fex. the row already has a different row-definition, we
        // fail here.
        self.ds
            .metadata_set_row(key, row_metadata)
            .await
            .inspect_err(|e| error!("Failed to set metadata row err={:?}", e))?;
        info!("Updated row metadata of key='{}'", key);

        if self
            .ds
            .metadata_set_description(key, &desc)
            .await
            .inspect_err(|e| error!("Failed to set metadata description err={:?}", e))?
        {
            info!(
                "Updated description of key={} to auto generated description='{}'",
                &key, &desc
            );
            self.metadata_last_changed(&ctxt, key).await?;
        }
        Ok(())
    }

    #[instrument(skip(self, ctxt))]
    async fn set_metadata_mode(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalEmitter<'_>,
        key: &str,
        mode: SensorMode,
    ) -> Result<(), LogErr> {
        let new_mode: modio_logger_db::SensorMode = mode.into();
        if self
            .ds
            .metadata_set_mode(key, &new_mode)
            .await
            .inspect_err(|e| error!("Failed to set metadata mode err={:?}", e))?
        {
            info!("Updated mode of key={} to mode='{:?}'", &key, &new_mode);
            self.metadata_last_changed(&ctxt, key).await?;
        }
        Ok(())
    }

    #[instrument(skip(self, ctxt))]
    async fn set_metadata_value_map(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalEmitter<'_>,
        key: &str,
        value_map: ValueMap,
    ) -> Result<(), LogErr> {
        if self
            .ds
            .metadata_set_enum(key, &value_map)
            .await
            .inspect_err(|e| error!("Failed to set metadata enum err={:?}", e))?
        {
            info!(
                "Updated metadata of key={} to enum='{:?}'",
                &key, &value_map
            );
            self.metadata_last_changed(&ctxt, key).await?;
        }
        Ok(())
    }

    #[instrument(skip(self, ctxt))]
    async fn set_metadata_unit(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalEmitter<'_>,
        key: &str,
        unit: String,
    ) -> Result<(), LogErr> {
        use crate::types::UnitError;
        use std::convert::TryFrom;

        // First validate by turning it into our Unit, then return it back to a string format.
        let unit = Unit::try_from(unit)?.into_inner();

        let res = self.ds.metadata_set_unit(key, &unit).await;

        // Error handling here is more involved than normally.
        // We want to return a UnitError "Unique" when we get a Unique constraint failure from the
        // database, as we do not permit overwriting unit errors.
        //
        // This in turn means that we need to manually match the error type, rather than let
        // automatic conversion do it for us.
        match res {
            // Actually wrote the update to db
            Ok(true) => {
                info!("Updated unit of key={} to unit={}", &key, &unit);
                self.metadata_last_changed(&ctxt, key).await?;
                Ok(())
            }
            // Did not change the db
            Ok(false) => Ok(()),
            Err(modio_logger_db::Error::Unique { source }) => {
                debug!("Throwing away database error: {:?}", source);
                Err(UnitError::Unique.into())
            }
            Err(e) => {
                error!("Failed to set metadata unit. err={:?}", e);
                Err(e.into())
            }
        }
    }

    #[instrument(skip(self))]
    async fn get_metadata(&mut self, key: &str) -> Result<Metadata, LogErr> {
        // Get the metadata for the key.  Error if not found?
        //
        info!("Fetching metadata for key={}", &key);
        match self.ds.get_metadata(key).await {
            Ok(None) => Err(LogErr::NoMetadata),
            Ok(Some(data)) => Ok(Metadata::from(data)),
            // We don't log key not found errors as they are the callers problem.
            Err(e @ modio_logger_db::Error::NotFound { .. }) => {
                debug!(
                    "Requested metadata for non-existing key. key={}, err={:?}",
                    key, e
                );
                Err(e.into())
            }
            Err(e) => {
                error!("Failed to fetch metadata from db. err={:?}", e);
                Err(e.into())
            }
        }
    }

    /// Signal sent when values are stored
    #[zbus(signal)]
    async fn store_signal(
        ctxt: &SignalEmitter<'_>,
        batch: Vec<(String, zValue<'_>, i64)>,
    ) -> zbus::Result<()>;

    #[instrument(skip_all)]
    async fn store_batch(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalEmitter<'_>,
        mut batch: HashMap<String, zValue<'_>>,
    ) -> Result<(), LogErr> {
        let time = modio_logger_db::fxtime_ms();
        let timefail = self.timefail.is_timefail();

        // Check all the values and keys, print a debug-friendly log message on the terminal, and
        // then return an early error.
        for (key, value) in &batch {
            if let Err(e) = keys::valid_key(key) {
                warn!("Invalid key for key='{key}' value='{value:?}' err='{e}'");
                return Err(e.into());
            }
            if let Err(e) = valid_metric(value) {
                warn!("Invalid data for key='{key}' value='{value:?}' err='{e}'");
                return Err(e.into());
            }
        }

        // Clone the values and insert into a vec of Metrics to store
        // This converts any floats/ints/bools/strings to string, which is an implementation
        // detail that may change in the future.
        let db_batch: Vec<Metric> = batch
            .iter()
            // Apply "maybe_metric" on the (key, value, timestamp), which returns Option<Metric>.
            // And filter out anything that is None.
            .filter_map(|(key, value)| maybe_metric(key, value, time))
            .collect();

        // If this fails, bail early.
        self.ds
            .insert_bulk(db_batch, timefail)
            .await
            .inspect_err(|e| error!("Failed to insert batch of data to ds. err={:?}", e))?;
        #[allow(clippy::cast_possible_truncation)]
        let trunc_time = time as i64;

        let payload: Vec<_> = batch
            .drain()
            // Keep all keys that do not start with "modio."
            .filter(|(key, _)| !key.starts_with("modio."))
            // And turn them into a tuple
            .map(|(key, value)| (key, value, trunc_time))
            .collect();

        // Do not send signals if the payload is empty
        if !payload.is_empty() {
            Self::store_signal(&ctxt, payload).await?;
        }

        Ok(())
    }

    /// Store a single metric.
    ///
    /// In reality, it will just call the batch interface and is thus less
    /// efficient than just using a batch.
    ///
    /// However, it helps porting from the old interface.
    #[instrument(skip_all)]
    async fn store(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalEmitter<'_>,
        key: String,
        value: zValue<'_>,
    ) -> Result<(), LogErr> {
        let batch = HashMap::from([(key, value)]);
        self.store_batch(ctxt, batch).await?;
        Ok(())
    }

    /// This store a series of _rows_ of data for a single key.
    ///
    /// The interface is meant for things that store coherent rows of data in the same form:
    ///    timestamp,  [key1, key2, key3, key4, key5]
    ///    timestamp,  [key1, key2, key3, key4, key5]
    ///    timestamp,  [key1, key2, key3, key4, key5]
    ///
    /// The rows are expected to not change in size, but can come in high resolution compared to
    /// other data, ie. millisecond resolution of timestamps.
    ///
    /// A key is expected to tell the system _first_ that there will be a row and it's format, and
    /// the logger will fail data in case this does not match up with expectations.
    ///
    /// We also expect the data to always be the same size and shape.
    ///
    #[instrument(skip_all)]
    async fn store_rows(&mut self, key: &str, rows: Vec<TimestampRow<'_>>) -> Result<(), LogErr> {
        keys::valid_key(key)?;

        // Check if the key has a registered metadata row, if not, return an error.
        let row_datatypes = self
            .ds
            .metadata_get_row(key)
            .await
            // RowNotFound errors are common here and do not deserve to be logged as error
            .inspect_err(|e| debug!("Failed to get row metadata for key. err={:?}", e))?;

        // Pre-allocate enough data to fit the row we have
        let mut db_batch: Vec<Metric> = Vec::with_capacity(rows.len());

        for (time, row) in rows {
            // Do data validation for the row.
            check_data_row(&row_datatypes, &row)?;

            let value = zbus_row_to_json_string(row)?;
            let m = Metric {
                name: key.to_string(),
                value,
                time,
            };
            db_batch.push(m);
        }
        self.ds
            .insert_bulk(db_batch, false)
            .await
            .inspect_err(|e| error!("Failed to insert bulk data for key. err={:?}", e))?;
        // Row data does not cause signals to happen as DBus is not the proper place to get bulk
        // data every few milliseconds.
        Ok(())
    }

    /// Store a series of metrics with timestamp
    ///
    /// Each metric may cause a signal to be emitted.
    /// Do not use for bulk data.
    #[instrument(skip_all)]
    async fn store_ts(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalEmitter<'_>,
        batch: Vec<(String, zValue<'_>, f64)>,
    ) -> Result<(), LogErr> {
        // Check all the values and keys, print a debug-friendly log message on the terminal, and
        // then return an early error.
        for (key, value, _) in &batch {
            if let Err(e) = keys::valid_key(key) {
                warn!("Invalid key for key='{key}' value='{value:?}' err='{e}'");
                return Err(e.into());
            }
            if let Err(e) = valid_metric(value) {
                warn!("Invalid data for key='{key}' value='{value:?}' err='{e}'");
                return Err(e.into());
            }
        }

        // Clone the values and insert into a vec of Metrics to store
        // This converts any floats/ints/bools/strings to string, which is an implementation
        // detail that may change in the future.
        let db_batch: Vec<Metric> = batch
            .iter()
            // Apply "maybe_metric" on the (key, value, time) which returns Option<Metric>.
            // Then removing all None, leaving a vector of Metric values
            .filter_map(|(key, value, time)| maybe_metric(key, value, *time))
            .collect();

        // If this fails, bail early.
        self.ds
            .insert_bulk(db_batch, false)
            .await
            .inspect_err(|e| error!("Failed to insert data with ts. err={:?}", e))?;

        #[allow(clippy::cast_possible_truncation)]
        let payload: Vec<_> = batch
            .into_iter()
            // Keep all keys that do not start with "modio."
            .filter(|(key, _, _)| !key.starts_with("modio."))
            // And turn them into a tuple
            .map(|(key, value, time)| (key, value, time as i64))
            .collect();

        // Do not send signals if the payload is empty
        if !payload.is_empty() {
            Self::store_signal(&ctxt, payload).await?;
        }
        Ok(())
    }
}

#[cfg(test)]
mod test {
    use super::Logger1;
    use crate::conn::make_connection;
    use crate::testing::TestServer;
    use crate::types::Unit;
    use async_std::task::sleep;
    use fsipc::logger1::SensorMode;
    use futures_util::{FutureExt, StreamExt};
    use modio_logger_db::Datastore;
    use modio_logger_db::SqlitePoolBuilder;
    use std::collections::HashMap;
    use std::error::Error;
    use std::time::Duration;
    use tempfile;
    use test_log::test;
    use timeout_macro::timeouttest;
    use zbus::zvariant;

    #[test(timeouttest)]
    async fn set_metadata_works() -> Result<(), Box<dyn Error>> {
        const PATH: &str = "/se/modio/logger/metadata";
        let dbfile = tempfile::Builder::new()
            .prefix("set_metadata_works")
            .suffix(".sqlite")
            .tempfile()
            .expect("Error on tempfile");

        // Open the first pool
        let pool = SqlitePoolBuilder::new()
            .db_path(dbfile.path())
            .build()
            .await
            .expect("Error opening database");

        let ds = Datastore::new(pool).await?;
        {
            // As signals require the connection to be active, we set up the connection, object
            // server and logger here.
            let connection = make_connection(true).await?;
            let logger = Logger1::builder()
                .development(true)
                .datastore(ds)
                .build()
                .await?;

            // For signals to work, the logger needs to be registered in the object server of a
            // connection.
            connection.object_server().at(PATH, logger).await?;

            // Take a reference for the interface of our logger & path, so we can get a signal
            // context to use.
            let iface_ref = connection
                .object_server()
                .interface::<_, Logger1>(PATH)
                .await?;

            // Get the logger object back from the server again. Or well, a shimmed one.
            let mut logger = iface_ref.get_mut().await;
            // The signal context is held by the async function until the signal is done, after
            // which it is consumed.
            // So to test, we need a new context for each call.
            let ctx = iface_ref.signal_emitter();
            logger
                .set_metadata_name(
                    ctx.to_owned(),
                    "modio.key.key",
                    "Some internal name".to_string(),
                )
                .await?;
            logger
                .set_metadata_description(
                    ctx.to_owned(),
                    "modio.key.key",
                    "Some internal description".to_string(),
                )
                .await?;

            logger
                .set_metadata_name(
                    ctx.to_owned(),
                    "customer.key.key.key",
                    "Some customer name".to_string(),
                )
                .await?;
            logger
                .set_metadata_mode(
                    ctx.to_owned(),
                    "customer.key.key.key",
                    SensorMode::ReadWrite,
                )
                .await?;
            logger
                .set_metadata_description(
                    ctx.to_owned(),
                    "customer.key.key.key",
                    "Some customer description".to_string(),
                )
                .await?;
            logger.persist_data().await;
            drop(logger);
        }
        // Open the pool again
        let pool = SqlitePoolBuilder::new()
            .db_path(dbfile.path())
            .build()
            .await
            .expect("Error opening database");
        let ds = Datastore::new(pool).await?;
        let res = ds.metadata_get_names().await?;
        // Both customer and internal should be available when querying the datastore
        assert!(res.len() == 2);
        eprintln!("{res:?}");
        Ok(())
    }

    #[test(timeouttest)]
    async fn set_unit_override_fails() -> Result<(), Box<dyn Error>> {
        const PATH: &str = "/se/modio/logger/testcase";

        let ds = Datastore::temporary().await;
        {
            // As signals require the connection to be active, we set up the connection, object
            // server and logger here.
            let connection = make_connection(true).await?;
            let logger = Logger1::builder()
                .development(true)
                .datastore(ds)
                .build()
                .await?;
            // For signals to work, the logger needs to be registered in the object server of a
            // connection.
            connection.object_server().at(PATH, logger).await?;

            // Take a reference for the interface of our logger & path, so we can get a signal
            // context to use.
            let iface_ref = connection
                .object_server()
                .interface::<_, Logger1>(PATH)
                .await?;
            // Get the logger object back from the server again. Or well, a shimmed one.
            let mut logger = iface_ref.get_mut().await;
            // The signal context is held by the async function until the signal is done, after
            // which it is consumed.
            // So to test, we need a new context for each call.
            let ctx = iface_ref.signal_emitter();
            logger
                .set_metadata_name(
                    ctx.to_owned(),
                    "customer.key.key",
                    "Some customer name".to_string(),
                )
                .await?;

            logger
                .set_metadata_unit(ctx.to_owned(), "customer.key.key.key", Unit::string("Cel"))
                .await?;

            // Same unit once more. should work
            logger
                .set_metadata_unit(ctx.to_owned(), "customer.key.key.key", Unit::string("Cel"))
                .await?;

            // New unit. should fail
            let res = logger
                .set_metadata_unit(ctx.to_owned(), "customer.key.key.key", Unit::string("m"))
                .await;
            drop(logger);
            let e = res.expect_err("should not be able to set it twice");
            assert!(e.to_string().contains("May not replace unit"));
        }
        Ok(())
    }

    #[test(timeouttest)]
    async fn set_read_only() -> Result<(), Box<dyn Error>> {
        const PATH: &str = "/se/modio/logger/testcase/set_read_only";
        let ds = Datastore::temporary().await;
        {
            // As signals require the connection to be active, we set up the connection, object
            // server and logger here.
            let connection = make_connection(true).await?;
            let logger = Logger1::builder()
                .development(true)
                .datastore(ds)
                .build()
                .await?;

            // For signals to work, the logger needs to be registered in the object server of a
            // connection.
            connection.object_server().at(PATH, logger).await?;

            // Take a reference for the interface of our logger & path, so we can get a signal
            // context to use.
            let iface_ref = connection
                .object_server()
                .interface::<_, Logger1>(PATH)
                .await?;
            // Get the logger object back from the server again. Or well, a shimmed one.
            let mut logger = iface_ref.get_mut().await;

            let ctx = iface_ref.signal_emitter();
            logger
                .set_metadata_name(
                    ctx.to_owned(),
                    "customer.key.key",
                    "Some customer name".to_string(),
                )
                .await?;
            let res = logger.get_metadata("customer.key.key").await?;
            assert!(res.mode.is_none());
            logger
                .set_metadata_mode(ctx.to_owned(), "customer.key.key", SensorMode::ReadOnly)
                .await?;
            let res = logger.get_metadata("customer.key.key").await?;
            drop(logger);
            assert_eq!(res.mode, Some(SensorMode::ReadOnly));
        }
        Ok(())
    }

    #[test(timeouttest)]
    async fn no_modio_signals_in_batch() -> Result<(), Box<dyn Error>> {
        let server = TestServer::new(line!()).await?;
        let logger1 = server.logger1().await?;
        let mut stream = logger1.receive_store_signal().await?;

        // Send a transaction
        let mut batch = HashMap::<String, zvariant::Value<'_>>::new();
        batch.insert("test.test.string".into(), String::from("string").into());
        batch.insert("test.test.int".into(), (42_u64).into());
        batch.insert("test.test.float".into(), (0.3_f64).into());
        batch.insert("modio.test.bool.true".into(), (true).into());
        batch.insert("modio.test.bool.false".into(), (false).into());
        logger1.store_batch(batch).await?;

        // Read signal, it should only be one, with a batch of data.
        let sig = stream.next().await.unwrap();
        let payload = sig.args()?;
        assert!(payload.batch.len() == 3, "Should have three out of 4 keys");
        // The modio internal keys should not be stored
        for (key, _, _) in payload.batch {
            assert!(key.starts_with("test.test"));
        }
        // We should not have more signals waiting
        let last_signal = stream.next().now_or_never();
        assert!(last_signal.is_none());

        let ipc = server.proxy().await?;
        {
            let m = ipc.retrieve("test.test.string").await?;
            assert_eq!(m.key, "test.test.string");
            assert_eq!(m.value, "string");
        }
        {
            let m = ipc.retrieve("test.test.int").await?;
            assert_eq!(m.key, "test.test.int");
            assert_eq!(m.value, "42");
        }

        {
            let m = ipc.retrieve("test.test.float").await?;
            assert_eq!(m.key, "test.test.float");
            assert_eq!(m.value, "0.3");
        }

        {
            let m = ipc.retrieve("modio.test.bool.true").await?;
            assert_eq!(m.key, "modio.test.bool.true");
            assert_eq!(m.value, "1");
        }

        {
            let m = ipc.retrieve("modio.test.bool.false").await?;
            assert_eq!(m.key, "modio.test.bool.false");
            assert_eq!(m.value, "0");
        }

        Ok(())
    }

    #[test(timeouttest)]
    async fn modio_only_no_signals() -> Result<(), Box<dyn Error>> {
        let server = TestServer::new(line!()).await?;
        let logger1 = server.logger1().await?;
        let mut stream = logger1.receive_store_signal().await?;

        // Send a transaction
        let mut batch = HashMap::<String, zvariant::Value<'_>>::new();
        batch.insert("modio.test.bool.true".into(), (true).into());
        batch.insert("modio.test.bool.false".into(), (false).into());
        logger1.store_batch(batch).await?;

        // We should not have any signals waiting
        assert!(
            stream.next().now_or_never().is_none(),
            "Should have no signals pending"
        );
        Ok(())
    }

    #[test(timeouttest)]
    async fn store_singles_work_as_well() -> Result<(), Box<dyn Error>> {
        let server = TestServer::new(line!()).await?;
        let logger1 = server.logger1().await?;
        let mut stream = logger1.receive_store_signal().await?;
        logger1
            .store("test.test.string".into(), String::from("string").into())
            .await?;
        logger1
            .store("test.test.int".into(), (42_u64).into())
            .await?;
        logger1
            .store("test.test.float".into(), (0.3_f64).into())
            .await?;
        logger1
            .store("modio.test.bool.true".into(), (true).into())
            .await?;
        logger1
            .store("modio.test.bool.false".into(), (false).into())
            .await?;

        // Read signal, it should only be one, with a batch of data.
        for _ in 0..3 {
            let sig = stream.next().await.unwrap();
            let payload = sig.args()?;
            // Maybe we should allow some buffering in the store signals too?
            assert!(payload.batch.len() == 1, "May have 1 after single stores");
            // The modio internal keys should not be stored
            for (key, _, _) in payload.batch {
                assert!(key.starts_with("test.test"));
            }
        }
        // We should not have more signals waiting
        assert!(
            stream.next().now_or_never().is_none(),
            "Should have no signals pending"
        );
        Ok(())
    }

    #[test(async_std::test)]
    async fn store_misc_with_ts_works() -> Result<(), Box<dyn Error>> {
        let server = TestServer::new(line!()).await?;
        let logger1 = server.logger1().await?;
        let mut stream = logger1.receive_store_signal().await?;

        // Ugly helper to make a tuple of test-data that goes in the interface
        fn tst<'a>(
            key: &str,
            val: impl Into<zvariant::Value<'a>>,
            ts: f64,
        ) -> (String, zvariant::Value<'a>, f64) {
            (key.into(), val.into(), ts)
        }

        let values = vec![
            tst("test.test.string", "string", 1708017581.0),
            tst("test.test.int", 42_u64, 1708017581.1),
            tst("test.test.float", 0.3_f64, 1708017581.2),
            tst("modio.test.bool.true", true, 1708017581.3),
            tst("modio.test.bool.false", false, 1708017581.4),
        ];
        logger1.store_ts(values).await?;

        // Read signal, it should only be one, with a batch of data.
        let sig = stream.next().await.unwrap();
        let payload = sig.args()?;
        assert!(
            payload.batch.len() == 3,
            "Should have three sets in the payload"
        );
        // The modio internal keys should not be stored
        for (key, _, _) in payload.batch {
            assert!(key.starts_with("test.test"));
        }
        // We should not have more signals waiting
        assert!(
            stream.next().now_or_never().is_none(),
            "Should have no signals pending"
        );
        Ok(())
    }

    #[test(async_std::test)]
    async fn store_rows_works() -> Result<(), Box<dyn Error>> {
        use zbus::zvariant::Value;

        let server = TestServer::new(line!()).await?;
        let logger1 = server.logger1().await?;
        let mut stream = logger1.receive_store_signal().await?;

        fn cast(v: &[(f64, [f64; 7])]) -> Vec<(f64, Vec<Value<'static>>)> {
            v.iter()
                .map(|(ts, vals)| (*ts, vals.iter().map(|x| Value::from(*x)).collect()))
                .collect()
        }
        // Send a transaction
        let raw_rows = [
            (1707329936.0, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
            (1707329936.1, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
            (1707329936.2, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
            (1707329936.3, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
            (1707329936.4, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
            (1707329936.5, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
            (1707329936.6, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
            (1707329936.7, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
            (1707329936.8, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
            (1707329936.9, [0.0, 1.1, 2.2, 3.3, 4.4, 5.5, 6.6]),
        ];
        let rows = cast(&raw_rows);
        let rows2 = cast(&raw_rows);

        let res1 = logger1.store_rows("test.rows.key", rows2).await;
        res1.expect_err("Should be an error that test.rows.key is not prepared");
        let row_items = vec![
            ("nil", "f64"),
            ("First_name", "f64"),
            ("Second", "f64"),
            ("Thirdly", "f64"),
            ("other", "f64"),
            ("fifthly", "f64"),
            ("sixer", "f64"),
        ];
        logger1.set_metadata_row("test.rows.key", row_items).await?;

        let res1 = logger1.store_rows("test.rows.key", rows).await;
        res1.expect("Should be ok because test.rows.key is prepared");
        // We should not have more signals waiting
        assert!(
            stream.next().now_or_never().is_none(),
            "Should have no signals pending"
        );
        Ok(())
    }

    #[test(async_std::test)]
    async fn row_metadata_checks() -> Result<(), Box<dyn Error>> {
        let server = TestServer::new(line!()).await?;
        let logger1 = server.logger1().await?;

        let mut row_items = vec![
            ("nil", "f64"),
            ("First_name", "string"),
            ("Second", "u64"),
            ("liar", "bool"),
        ];

        logger1
            .set_metadata_row("test.rows.meta", row_items.clone())
            .await?;
        // Setting row items twice is expected to work
        logger1
            .set_metadata_row("test.rows.meta", row_items.clone())
            .await?;

        row_items.push(("this fails", "f64"));
        // Setting row items to a different thing should fail
        logger1
            .set_metadata_row("test.rows.meta", row_items)
            .await
            .expect_err("Expected a failure but got success");
        Ok(())
    }
    #[test(timeouttest)]
    async fn timestamp_on_update() -> Result<(), Box<dyn Error>> {
        let server = TestServer::new(line!()).await?;
        let logger1 = server.logger1().await?;
        let ipc = server.proxy().await?;
        let second = Duration::from_secs(1);
        ipc.retrieve("modio.logger.metadata.change")
            .await
            .expect_err("Should have no data");

        logger1
            .set_metadata_name("customer.key.key", "Some customer name")
            .await
            .expect("Should be able to set name");
        let res = logger1
            .get_metadata("customer.key.key")
            .await
            .expect("Should have metadata");
        assert!(res.name.is_some());
        let metric = ipc
            .retrieve("modio.logger.metadata.change")
            .await
            .expect("Should have data for metadata changed");

        // Need to have a delay as the Measures are truncated to second precision
        sleep(second).await;

        logger1
            .set_metadata_name("customer.key.key", "Some customer name")
            .await
            .expect("Should be able to set name to the same thing");
        let metric2 = ipc
            .retrieve("modio.logger.metadata.change")
            .await
            .expect("Should have data for metadata changed");
        assert_eq!(
            metric.timestamp, metric2.timestamp,
            "Timestamp should not change when no metadata updated."
        );
        // Need to have a delay as the Measures are truncated to second precision
        sleep(second).await;
        logger1
            .set_metadata_name("customer.key.key", "Some OTHER name")
            .await
            .expect("Should be able to change the name");
        let metric3 = ipc
            .retrieve("modio.logger.metadata.change")
            .await
            .expect("Should have data for metadata changed");
        assert!(
            metric.timestamp < metric3.timestamp,
            "Timestamp should have updated"
        );
        Ok(())
    }
}