modio_logger_db/datastore/

ds.rs

// Author: D.S. Ljungmark <spider@skuggor.se>, Modio AB
// SPDX-License-Identifier: AGPL-3.0-or-later
use super::{raw, TagType};
use crate::buffer::fxtime;
use crate::buffer::{TimeStatus, VecBuffer};
use crate::types::{Metric, Statistics, TXMetric};
use crate::Error;
use async_std::sync::Mutex;
use async_std::task;
use radix_trie::{Trie, TrieCommon};
use sqlx::SqlitePool;
use std::sync::Arc;
use tempfile::NamedTempFile;
use tracing::instrument;
use tracing::{debug, error, info, warn};

#[derive(Clone)]
pub struct Datastore {
    pool: SqlitePool,
    buffer: VecBuffer,
    names: Arc<Mutex<Trie<String, ()>>>,
    drop_event: Arc<tokio::sync::Notify>,
}

// To avoid printing out _all_ the spam in drop_event, we implement a custom Debug here.
impl std::fmt::Debug for Datastore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let opts = &self.pool.options();
        let mut d = f.debug_struct("Datastore");
        d.field("pool.size", &self.pool.size())
            .field("pool.num_idle", &self.pool.num_idle())
            .field("pool.is_closed", &self.pool.is_closed())
            .field("pool.acquire_timeout", &opts.get_acquire_timeout())
            .field("buffer", &self.buffer);
        match self.names.try_lock() {
            Some(guard) => {
                d.field("names", &guard.len());
            }
            _ => {
                d.field("names", &"<Locked>");
            }
        }
        d.finish_non_exhaustive()
    }
}

const MAX_PERSIST_AGE: f64 = 57.0;

impl Datastore {
    /// Construct a datastore
    /// Takes an pre-configured and migrated DB pool and opens a temp buffer.
    ///
    /// # Errors
    ///    Any `SQLx` Error is returned.
    pub async fn new(pool: SqlitePool) -> Result<Self, Error> {
        let buffer = VecBuffer::new();
        let drop_event = Self::drop_event(pool.clone(), buffer.clone(), None);
        let names = Arc::new(Mutex::new(Trie::new()));
        let res = Self {
            pool,
            buffer,
            names,
            drop_event,
        };
        Ok(res)
    }

    #[must_use]
    pub fn pool(&self) -> SqlitePool {
        self.pool.clone()
    }
    #[cfg(test)]
    async fn get_name(&self, key: &str) -> Result<String, Error> {
        raw::get_name(&self.pool, key).await
    }

    #[cfg(test)]
    async fn check_tempdata(&self) -> Result<(), Error> {
        if let Some(oldest) = self.buffer.oldest() {
            info!("Oldst value is {}", oldest);
        } else {
            info!("Buffer was locked or otherwise occupied.");
        }
        Ok(())
    }
    #[cfg(test)]
    async fn sensor_id(&self, key: &str) -> Result<i64, Error> {
        self.add_sensor(key).await?;
        raw::sensor_id(&self.pool, key).await
    }
    /// Helper, count the remaining transaction in the table.
    ///
    /// # Errors
    ///
    /// Will return Err in case of database errors (connections, not existing, etc)
    #[cfg(test)]
    pub(crate) async fn count_transactions(&self) -> Result<i32, Error> {
        let count = raw::count_transactions(&self.pool).await?;
        Ok(count)
    }

    /// Return statistics for the database, meant to be used for logging and similar.
    ///
    /// # Errors
    ///
    /// Will return Err in case of database errors (connections, not existing, etc)
    #[instrument]
    pub async fn get_statistics(&self) -> Result<Statistics, Error> {
        let mut res = raw::get_statistics(&self.pool).await?;
        // Replace the "buffered" value with a fresh count, mapping "None" (locked, etc) to -1, and
        // casting the usize result to i32
        res.buffered = self.buffer.count().map_or(-1, |v| v as i32);
        Ok(res)
    }
    /// Insert a single data-point into the database or buffer
    ///
    /// # Errors
    ///   Any SQL error.
    #[instrument(level = "debug", skip(value, time, timefail))]
    pub async fn insert(
        &self,
        key: &str,
        value: &str,
        time: f64,
        timefail: bool,
    ) -> Result<(), Error> {
        let arf = vec![Metric {
            name: key.into(),
            value: value.into(),
            time,
        }];
        self.insert_bulk(arf, timefail).await
    }

    /// Insert a batch of data-points into the buffer store
    ///
    /// # Errors
    ///   Any SQL error.
    async fn insert_bulk_buffer(&self, data: Vec<Metric>, status: TimeStatus) -> Result<(), Error> {
        for metric in data {
            self.buffer.add_metric(metric, status).await;
        }
        Ok(())
    }
    #[instrument(level = "debug")]
    async fn add_sensor(&self, key: &str) -> Result<(), Error> {
        {
            let mut guard = self.names.lock().await;
            if guard.get(key).is_none() {
                raw::add_sensor(&self.pool, key).await?;
                guard.insert(key.to_string(), ());
            }
        }
        Ok(())
    }

    /// Insert a batch of data-points into the datastore
    /// Stores all the keys to permanent storage, but may buffer actual log values.
    ///
    /// # Errors
    ///   Any SQL error.
    #[instrument(level = "debug", skip(data))]
    pub async fn insert_bulk(&self, data: Vec<Metric>, timefail: bool) -> Result<(), Error> {
        // Start by adding all the names.
        for metric in &data {
            self.add_sensor(&metric.name).await?;
        }
        let status = TimeStatus::from_bool(timefail);
        self.insert_bulk_buffer(data, status).await?;
        Ok(())
    }
    /// Check if we should persist data to disk according to heurestic
    ///
    /// # Errors
    /// Any SQL errors are returned.
    #[instrument]
    pub async fn should_persist(&self) -> Result<bool, Error> {
        const CUTOFF: usize = 100;
        #[allow(clippy::option_if_let_else)]
        if let Some(num) = self.buffer.count() {
            Ok(num > CUTOFF)
        } else {
            warn!("Failed to count data in buffer, assuming full.");
            Ok(true)
        }
    }

    /// Check if we should persist data to disk according to age heurestic
    ///
    /// # Errors
    /// Any SQL errors are returned.
    #[instrument]
    pub async fn should_persist_age(&self) -> Result<bool, Error> {
        let now = fxtime();
        #[allow(clippy::option_if_let_else)]
        match self.buffer.oldest() {
            Some(eldest) => {
                let age = now - eldest;
                debug!("Oldest value in buffer is {} ( age: {}s )", &eldest, &age);
                Ok(age > MAX_PERSIST_AGE)
            }
            None => {
                error!("Buffer locked, assuming we should persist");
                Ok(true)
            }
        }
    }
    // Persist data to disk
    #[instrument(level = "info", skip(self))]
    pub async fn persist_data(&self) -> Result<(), Error> {
        let delete_buffer = self.buffer.consume_metrics().await;
        debug!("Persisting data, count={}", delete_buffer.len());
        raw::persist_data(&self.pool, &delete_buffer)
            .await
            .inspect_err(|e| error!("Failed to persist, err={e:?}, pool={:?}", &self.pool))?;
        Ok(())
    }

    // Maybe perist data to disk based on age
    #[instrument(level = "info", skip(self))]
    pub async fn maybe_persist_data(&self) -> Result<(), Error> {
        if self.should_persist_age().await? {
            let delete_buffer = self.buffer.consume_metrics().await;
            if let Err(e) = raw::persist_data(&self.pool, &delete_buffer).await {
                // Re-insert the data into the buffer.
                self.buffer.add_metrics(delete_buffer).await;
                warn!("Failed to persist data, continuing. err={:?}", e);
            };
        }
        Ok(())
    }

    // When dropping the datastore, Drop is called in a Sync context, which may not be using
    // async features, giving us a head-ache.
    //
    // This function spawns a task that waits for the notifier that we should shut down and save
    // buffer, and only exits after that.
    //
    // By signalling the drop event during Drop, we can then fake a callback style of async code
    // to write data to disk when the Datastore is deallocated.
    fn drop_event(
        pool: SqlitePool,
        buffer: VecBuffer,
        dbfile: Option<Arc<NamedTempFile>>,
    ) -> Arc<tokio::sync::Notify> {
        let notify = Arc::new(tokio::sync::Notify::new());
        let waiting = notify.clone();
        task::spawn(async move {
            waiting.notified().await;
            info!(
                "Persisting Buffered data to disk due to deallocation, file={:?}",
                dbfile
            );
            let delete_buffer = buffer.consume_metrics().await;
            if !delete_buffer.is_empty() {
                debug!("Persisting count={} buffered values", delete_buffer.len());
                if let Err(e) = raw::persist_data(&pool, &delete_buffer).await {
                    error!(
                        "Failed to write to database {:?} {:?} file={:?}",
                        &pool, e, &dbfile
                    );
                    if let Some(fil) = dbfile {
                        error!("File: exists: {:?}", fil.path().try_exists());
                        error!("File: metadata: {:?}", fil.path().metadata());
                    }
                    panic!("Failed to persist data {e}");
                }
            };
            info!("Closing pool due to permanence.");
            pool.close().await;
        });
        notify
    }
}

impl Drop for Datastore {
    fn drop(&mut self) {
        let size = self.pool.size();
        let closed = self.pool.is_closed();
        debug!("Dropping datastore size={size}, closed={closed}");
        // As drop is always called in a SYNC context, and it is hairy to recurse and find an async
        // executor that would allow us to write buffered data to disk, instead we have a task
        // waiting when creating the Datastore, which we notify here. That task will then run the
        // job of writing data to disk and closing files properly.
        self.drop_event.notify_waiters();
    }
}

// Batches of data
impl Datastore {
    #[instrument]
    pub async fn get_batch(&self, size: u32) -> Result<Vec<TXMetric>, Error> {
        let res = raw::get_batch(&self.pool, size).await?;
        if res.len() >= (size as usize) {
            return Ok(res);
        }
        // If we have < size metrics in our datastore,
        // we _may_ persist some data to disk.
        //
        // As we do not wish to transmit data without storing it,
        // and we do not want a loop of:
        //    buffer: 1 data, to send: 0 data
        //    submit calls "get batch"
        //    write buffer to disk,  give to submit
        //    submit stores result of that work (1 data)
        //    submit sees "we had data", asks for data again
        //    write buffer to disk, ...
        //
        // As that would ruin our disk performance, we only consider writing the buffer
        // in case we do not have enough metrics to give, and the buffer should be synced.
        // (time or fill-rate)
        self.maybe_persist_data().await?;
        let res = raw::get_batch(&self.pool, size).await?;
        Ok(res)
    }

    #[instrument]
    pub async fn get_internal_batch(&self, size: u32) -> Result<Vec<TXMetric>, Error> {
        let res = raw::get_internal_batch(&self.pool, size).await?;
        // See the comment in get_batch for why we check the count like this.
        if res.len() >= (size as usize) {
            return Ok(res);
        }
        self.maybe_persist_data().await?;
        let res = raw::get_internal_batch(&self.pool, size).await?;
        Ok(res)
    }

    #[instrument(level = "debug", skip(ids))]
    pub async fn drop_batch(&self, ids: &[i64]) -> Result<(), Error> {
        raw::drop_batch(&self.pool, ids).await?;
        Ok(())
    }

    // Handling timefail
    #[instrument(level = "debug")]
    pub async fn fix_timefail(&self, adjust: f32) -> Result<u64, Error> {
        self.persist_data().await?;
        let count = raw::fix_timefail(&self.pool, adjust).await?;
        info!(
            "Updated TIMEFAIL status for count={} items with offset={}",
            &count, adjust
        );
        Ok(count)
    }
    #[instrument(level = "debug")]
    pub async fn get_last_datapoint(&self, key: &str) -> Result<Metric, Error> {
        // If the metric is in our buffer, return it early and do not cause extra write to disk.
        if let Some(metric) = self.buffer.get_metric(key).await {
            return Ok(metric);
        }
        let res = raw::get_last_datapoint(&self.pool, key).await?;
        Ok(res)
    }
    // Return the last point for all keys
    #[instrument(level = "debug", skip_all, fields(self))]
    pub async fn get_latest_logdata(&self) -> Result<Vec<Metric>, Error> {
        // Even if we are asked to get latest logdata, we do not retrieve the ones from our
        // internal buffer.
        //
        // Except if the data is about to expire.
        self.maybe_persist_data().await?;
        let res = raw::get_latest_logdata(&self.pool).await?;
        Ok(res)
    }
}

// Transaction (changes)
mod changes {
    use super::raw;
    use super::Datastore;
    use crate::buffer::TimeStatus;
    use crate::types::Transaction;
    use crate::Error;
    use tracing::instrument;
    use tracing::{debug, info};

    impl Datastore {
        #[instrument(level = "debug")]
        pub async fn has_transaction(&self, token: &str) -> Result<bool, Error> {
            let res = raw::has_transaction(&self.pool, token).await?;
            Ok(res)
        }

        #[instrument(level = "info", skip(expected, target, token))]
        pub async fn transaction_add(
            &self,
            key: &str,
            expected: &str,
            target: &str,
            token: &str,
        ) -> Result<(), Error> {
            let internal_key = format!("mytemp.internal.change.{key}");
            self.add_sensor(key).await?;
            self.add_sensor(&internal_key).await?;
            info!(
                "Storing transaction {}, '{}' => '{}'",
                &key, &expected, &target
            );
            raw::transaction_add(&self.pool, key, expected, target, token).await?;
            Ok(())
        }
        #[instrument(level = "info")]
        pub async fn transaction_get(&self, prefix: &str) -> Result<Vec<Transaction>, Error> {
            let res = raw::transaction_get(&self.pool, prefix).await?;
            Ok(res)
        }
        #[instrument]
        pub async fn transaction_fail(
            &self,
            transaction_id: i64,
            timefail: bool,
        ) -> Result<u64, Error> {
            use crate::datastore::TransactionStatus::Failed;
            debug!("Failing transaction with id={}", transaction_id);
            let timefail = TimeStatus::from_bool(timefail);
            let count = raw::transaction_mark(&self.pool, transaction_id, Failed, timefail).await?;
            Ok(count)
        }

        #[instrument]
        pub async fn transaction_pass(
            &self,
            transaction_id: i64,
            timefail: bool,
        ) -> Result<u64, Error> {
            use crate::datastore::TransactionStatus::Success;
            debug!("Passing transaction with id={}", transaction_id);
            let timefail = TimeStatus::from_bool(timefail);
            let count =
                raw::transaction_mark(&self.pool, transaction_id, Success, timefail).await?;
            Ok(count)
        }

        /// Fail all pending transactions
        #[instrument(level = "info")]
        pub async fn transaction_fail_pending(&self) -> Result<u64, Error> {
            let count = raw::transaction_fail_pending(&self.pool).await?;
            info!("Marked {} pending transactions as failed", count);
            Ok(count)
        }
    }
}

mod clean {
    use super::raw;
    use super::Datastore;
    use crate::CleanResult;
    use crate::Error;
    use sqlx::{Connection, SqlitePool};
    use tracing::instrument;
    use tracing::{debug, error};

    impl Datastore {
        #[instrument(level = "info", skip(pool))]
        pub async fn clean_maintenance(pool: SqlitePool) -> Result<CleanResult, Error> {
            let mut conn = pool
                .acquire()
                .await
                .inspect_err(|e| error!(err=?e, "Failed to get connection"))?;
            conn.ping().await?;
            // These functions take a Sqliteconnection as one cannot perform "VACUUM" and "PRAGMA" on
            // the pool, but need to use a raw connection (and no transaction)
            let trans_failed = raw::fail_queued_transactions(&mut conn)
                .await
                .inspect_err(|e| error!("Failed to mark queued transactions. err={:?}", e))
                .unwrap_or(-1);
            let trans_old = raw::delete_old_transactions(&mut conn)
                .await
                .inspect_err(|e| error!("Failed to delete old transactions. err={:?}", e))
                .unwrap_or(-1);
            let data_deleted = raw::delete_old_logdata(&mut conn)
                .await
                .inspect_err(|e| error!("Failed to delete old logdata. err={:?}", e))
                .unwrap_or(-1);
            raw::need_vacuum_or_shrink(&mut conn)
                .await
                .inspect_err(|e| error!("Failed to vacuum / shrink. err={:?}", e))?;
            conn.close().await?;
            let res = CleanResult {
                trans_failed,
                trans_old,
                data_deleted,
            };
            Ok(res)
        }

        #[cfg(test)]
        /// Wrapper of delete_old_transactions_raw, for use with tests.
        pub(crate) async fn delete_old_transactions(&self) -> Result<i32, Error> {
            debug!("delete_old_transactions, grabbing pool connection");
            let mut conn = self
                .pool
                .acquire()
                .await
                .inspect_err(|e| error!(err=?e, "Failed to acquire connection"))?;
            let res = raw::delete_old_transactions(&mut conn).await?;
            conn.close().await?;
            Ok(res)
        }

        /// Wrapper of fail_queued_transactions_raw
        #[instrument]
        pub async fn fail_queued_transactions(&self) -> Result<i32, Error> {
            debug!("fail_queued_transactions, grabbing pool connection");
            let mut conn = self
                .pool
                .acquire()
                .await
                .inspect_err(|e| error!(err=?e, "Failed to acquire connection"))?;
            let res = raw::fail_queued_transactions(&mut conn).await?;
            conn.close().await?;
            Ok(res)
        }

        /// Wrapper of the private delete_old_logdata_raw for use with tests.
        #[cfg(test)]
        pub(crate) async fn delete_old_logdata(&self) -> Result<i32, Error> {
            debug!("delete_old_logdata, grabbing pool connection");
            let mut conn = self
                .pool
                .acquire()
                .await
                .inspect_err(|e| error!(err=?e, "Failed to acquire connection"))?;
            let res = raw::delete_old_logdata(&mut conn).await?;
            conn.close().await?;
            Ok(res)
        }

        /// Delete random data
        #[cfg(test)]
        pub async fn delete_random_data(&self) -> Result<i32, Error> {
            debug!("delete_random_data, grabbing pool connection");
            let mut conn = self
                .pool
                .acquire()
                .await
                .inspect_err(|e| error!(err=?e, "Failed to acquire connection"))?;
            let res = raw::delete_random_data(&mut conn).await?;
            conn.close().await?;
            Ok(res)
        }
    }
}

impl Datastore {
    /// Create a new datastore with a new temporary file
    ///
    /// Panic:
    ///   will panic if anything fails
    pub async fn temporary() -> Self {
        let dbfile = tempfile::Builder::new()
            .prefix("database")
            .suffix(".sqlite")
            .tempfile()
            .expect("Error on tempfile");
        Self::new_tempfile(dbfile.into()).await
    }

    /// Create a new datastore from an existing temporary file
    ///
    /// Panic:
    ///   will panic if anything fails
    pub async fn new_tempfile(dbfile: Arc<NamedTempFile>) -> Self {
        use crate::db::SqlitePoolBuilder;
        let pool = SqlitePoolBuilder::new()
            .db_path(dbfile.path())
            .migrate(true)
            .build()
            .await
            .expect("Failed to build pool");

        let buffer = VecBuffer::new();
        let drop_event = Self::drop_event(pool.clone(), buffer.clone(), Some(dbfile));
        let names = Arc::new(Mutex::new(Trie::new()));
        Self {
            pool,
            buffer,
            names,
            drop_event,
        }
    }
}

#[cfg(test)]
fn metrc(name: &str, value: &str, time: f64) -> crate::Metric {
    crate::Metric {
        name: name.into(),
        value: value.into(),
        time,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::SqlitePoolBuilder;
    use std::error::Error;
    use test_log::test;
    use timeout_macro::timeouttest;
    type TestResult = Result<(), Box<dyn Error>>;

    #[test(timeouttest)]
    async fn has_file_database() -> TestResult {
        let tempfile = tempfile::Builder::new()
            .prefix("loggerdb_has_file_database")
            .suffix(".sqlite")
            .tempfile()?;
        let named_path = tempfile.path();
        let pool = SqlitePoolBuilder::new()
            .db_path(named_path)
            .migrate(true)
            .build()
            .await
            .unwrap();
        sqlx::query("SELECT * FROM sensor where sensor.name = 'mytemp.internal.sensors'")
            .fetch_one(&pool)
            .await?;
        // Close the pool so we do not leave sqlite's .shm and .wal files around
        pool.close().await;
        drop(pool);
        Ok(())
    }

    #[test(timeouttest)]
    async fn datastore_tempdata() {
        let ds = Datastore::temporary().await;
        ds.check_tempdata()
            .await
            .expect("Should be able to get data from temp table");
    }

    #[test(timeouttest)]
    async fn datastore_names() -> TestResult {
        let ds = Datastore::temporary().await;
        let res = ds.get_name("mytemp.internal.sensors").await?;
        assert_eq!(res, "mytemp.internal.sensors");
        Ok(())
    }

    #[test(timeouttest)]
    async fn add_one_name() -> TestResult {
        let ds = Datastore::temporary().await;
        let res = ds.sensor_id("test.test.test").await?;
        assert_eq!(res, 2);
        Ok(())
    }

    #[test(timeouttest)]
    async fn add_two_names() -> TestResult {
        let ds = Datastore::temporary().await;
        let res = ds.sensor_id("test.test.test").await?;
        assert_eq!(res, 2);
        let res = ds.sensor_id("test.test.test").await?;
        assert_eq!(res, 2);
        let res = ds.sensor_id("test.test.test").await?;
        assert_eq!(res, 2);
        let res = ds.get_name("test.test.test").await?;
        assert_eq!(res, "test.test.test");
        Ok(())
    }

    #[test(timeouttest)]
    async fn insert_timefail() -> TestResult {
        let ds = Datastore::temporary().await;
        ds.insert("test.test.ok", "value", 1_620_850_252.0, false)
            .await?;
        ds.insert("test.test.ok", "value1", 1_620_850_253.0, false)
            .await?;
        ds.insert("test.test.ok", "value2", 1_620_850_255.0, false)
            .await?;
        ds.insert("test.test.fimefail", "value", 1_620_850_252.0, true)
            .await?;
        Ok(())
    }

    #[test(timeouttest)]
    async fn retrieve_last_empty() -> TestResult {
        let ds = Datastore::temporary().await;
        let pool = ds.pool();
        ds.insert("test.test.ok", "value", 1_620_850_252.0, false)
            .await?;
        let before_sync = ds.get_last_datapoint("test.test.ok").await;
        assert!(before_sync.is_ok(), "Should have a value");

        ds.persist_data().await?;

        let after_sync = ds.get_last_datapoint("test.test.ok").await;
        assert!(after_sync.is_ok(), "Should have a value");

        let delete_all = sqlx::query("DELETE FROM logdata");
        delete_all.execute(&pool).await?;

        let after = ds.get_last_datapoint("test.test.ok").await;
        assert!(after.is_err(), "Should fail");
        Ok(())
    }

    #[test(timeouttest)]
    async fn retrieve_sorting() -> TestResult {
        let ds = Datastore::temporary().await;
        ds.insert("test.test.one", "value0", 1_620_850_000.0, false)
            .await?;
        ds.insert("test.test.one", "value1", 1_620_850_111.0, true)
            .await?;
        ds.insert("test.test.one", "value3", 1_620_850_222.0, false)
            .await?;
        ds.insert("test.test.two", "value3", 1_620_850_333.0, true)
            .await?;
        ds.insert("test.test.two", "value1", 1_620_850_222.0, false)
            .await?;
        ds.insert("test.test.two", "value0", 1_620_850_111.0, true)
            .await?;
        let res = ds.get_last_datapoint("test.test.two").await?;
        assert_eq!(res.name, "test.test.two");
        assert_eq!(res.value, "value3");
        assert_eq!(res.time, 1_620_850_333.0);

        let res = ds.get_last_datapoint("test.test.one").await?;
        assert_eq!(res.name, "test.test.one");
        assert_eq!(res.value, "value3");
        assert_eq!(res.time, 1_620_850_222.0);
        Ok(())
    }

    #[test(timeouttest)]
    async fn insert_bulk() -> TestResult {
        let ds = Datastore::temporary().await;

        let vals = vec![
            metrc("test.test.one", "etta", 16_208_501_111.0),
            metrc("test.test.two", "etta", 16_208_501_111.0),
            metrc("test.test.three", "etta", 16_208_501_112.0),
            metrc("test.test.one", "tvåa", 16_208_502_222.0),
            metrc("test.test.two", "tvåa", 16_208_502_223.0),
            metrc("test.test.three", "tvåa", 16_208_502_222.0),
            metrc("test.test.one", "trea", 16_208_503_333.0),
            metrc("test.test.two", "trea", 16_208_503_331.0),
            metrc("test.test.three", "trea", 16_208_503_333.0),
            metrc("test.test.one", "fyra", 16_208_504_444.0),
        ];
        ds.insert_bulk(vals, true).await?;
        let res = ds.get_last_datapoint("test.test.one").await?;
        assert_eq!(res.value, "fyra");
        let res = ds.get_last_datapoint("test.test.three").await?;
        assert_eq!(res.value, "trea");
        let res = ds.get_last_datapoint("test.test.two").await?;
        assert_eq!(res.value, "trea");
        Ok(())
    }

    #[test(timeouttest)]
    async fn insert_persist() -> TestResult {
        let ds = Datastore::temporary().await;

        let vals = vec![
            metrc("test.test.one", "etta", 16_208_501_111.0),
            metrc("test.test.two", "etta", 16_208_501_111.0),
            metrc("test.test.three", "etta", 16_208_501_112.0),
        ];
        // Generate a bulk of 1000 keys
        let seq = 0..1000;
        let more: Vec<Metric> = seq
            .map(|x| metrc("test.test.three", &format!("{x}"), fxtime()))
            .collect();

        ds.insert_bulk(vals, true).await?;
        let first = ds.should_persist().await?;
        assert!(!first, "Should not need persist with only 3 values");

        let should = ds.buffer.get_metric("test.test.one").await.is_some();
        assert!(
            should,
            "Should need persist because test.test.one is in buffer"
        );

        // Insert the bulk of the data
        ds.insert_bulk(more, true).await?;
        let second = ds.should_persist().await?;
        assert!(second, "Should need persist with more values");

        // Check that persist works
        let res = ds.persist_data().await;
        assert!(res.is_ok(), "We should have succesfully persisted data");

        // We should no longer need persist
        let third = ds.should_persist().await?;
        assert!(!third, "We should not need persist again");

        let should = ds.buffer.get_metric("test.test.one").await.is_some();
        assert!(!should, "This key should no longer be in the buffer");

        // Calling persist on an empty set should be ok.
        ds.persist_data().await?;
        Ok(())
    }

    #[test(timeouttest)]
    async fn persist_delete_and_purge() -> TestResult {
        let ds = Datastore::temporary().await;
        let vals: Vec<Metric> = (0..1000)
            .map(|x| {
                metrc(
                    &format!("test.test.bulk.{}", x % 7),
                    &x.to_string(),
                    fxtime(),
                )
            })
            .collect();
        ds.insert_bulk(vals, true).await?;
        let should_count = ds.should_persist().await?;
        let should_age = ds.should_persist_age().await?;
        assert!(should_count, "Should persist based on count");
        assert!(!should_age, "Oldest should not be that big");
        ds.persist_data().await?;
        let count = ds.delete_random_data().await?;
        assert!(count > 0, "Should have deleted some data");
        Ok(())
    }

    #[test(timeouttest)]
    async fn delete_old_logged_values() -> TestResult {
        let ds = Datastore::temporary().await;
        ds.insert("test.test.ok", "one", 1_620_850_000.0, false)
            .await?;
        ds.insert("test.test.ok", "two", 1_999_950_251.0, false)
            .await?;

        // We should have a value
        let metric = ds.get_last_datapoint("test.test.ok").await?;
        assert_eq!(metric.value, "two");

        // Submitter comes and does it's stuff here
        let to_xmit = ds.get_batch(50).await?;
        let mut to_remove = Vec::<i64>::with_capacity(5);
        for i in to_xmit {
            to_remove.push(i.id);
        }
        ds.drop_batch(&to_remove).await?;
        //  End submitter part

        // We should still have a value after this
        let metric = ds.get_last_datapoint("test.test.ok").await?;
        assert_eq!(metric.value, "two");

        // Cleanout routine should run on the device
        let count = ds.delete_old_logdata().await?;
        assert_eq!(count, 1);

        // After dropping data, we should still be able to get the last value out.
        let metric = ds.get_last_datapoint("test.test.ok").await?;
        assert_eq!(metric.value, "two");

        // Cleanout routine should run on the device
        let count = ds.delete_old_logdata().await?;
        // We should not have removed any rows now.
        assert_eq!(count, 0);

        Ok(())
    }

    /// Inserting multiple items and then returning "latest" should return one point for each value
    /// and nothing more.
    #[test(timeouttest)]
    async fn get_latest_datapoints() -> TestResult {
        let ds = Datastore::temporary().await;

        let vals = vec![
            metrc("test.test.one", "etta", 1_620_850_111.0),
            metrc("test.test.two", "etta", 1_620_850_111.0),
            metrc("test.test.three", "etta", 1_620_850_111.0),
            metrc("test.test.one", "tvåa", 1_620_850_222.0),
            metrc("test.test.two", "tvåa", 1_620_850_222.0),
            metrc("test.test.three", "tvåa", 1_620_850_222.0),
            metrc("test.test.one", "trea", 1_620_850_333.0),
            metrc("test.test.three", "trea", 1_620_850_333.0),
            metrc("test.test.one", "fyra", 1_620_850_444.0),
        ];
        ds.insert_bulk(vals, true).await?;
        ds.persist_data().await?;
        let res = ds.get_latest_logdata().await?;
        assert_eq!(res[0].name, "test.test.two");
        assert_eq!(res[0].value, "tvåa");
        assert_eq!(res[1].name, "test.test.three");
        assert_eq!(res[1].value, "trea");
        assert_eq!(res[2].name, "test.test.one");
        assert_eq!(res[2].value, "fyra");
        assert_eq!(res.len(), 3);
        Ok(())
    }

    /// Multiple values for a key should persist if the oldest is too old.
    #[test(timeouttest)]
    async fn get_latest_datapoints_persists() -> TestResult {
        let ds = Datastore::temporary().await;

        let vals = vec![
            metrc("test.test.one", "etta", fxtime() - MAX_PERSIST_AGE),
            metrc("test.test.one", "tvåa", fxtime() - MAX_PERSIST_AGE / 2.0),
            metrc("test.test.one", "trea", fxtime() - 1.0),
        ];
        ds.insert_bulk(vals, true).await?;
        let res = ds.get_latest_logdata().await?;
        assert_eq!(res[0].name, "test.test.one");
        assert_eq!(res[0].value, "trea");
        assert_eq!(res.len(), 1);

        // If the data in the buffer is "too fresh" it should not be persisted by the call to
        // get_latest_logdata() and we should thus get the previous value ("trea")
        let vals = vec![metrc("test.test.one", "fyra", fxtime())];
        ds.insert_bulk(vals, true).await?;
        let res = ds.get_latest_logdata().await?;
        assert_eq!(res[0].name, "test.test.one");
        assert_eq!(res[0].value, "trea");
        assert_eq!(res.len(), 1);

        // After a persist, we should get the latest value
        ds.persist_data().await?;
        let res = ds.get_latest_logdata().await?;
        assert_eq!(res[0].name, "test.test.one");
        assert_eq!(res[0].value, "fyra");
        assert_eq!(res.len(), 1);

        Ok(())
    }

    #[test(timeouttest)]
    async fn get_transactions() -> TestResult {
        let ds = Datastore::temporary().await;

        let vals = vec![
            metrc("test.test.one", "etta", 16_208_501_111.0),
            metrc("test.test.two", "tvåa", 16_208_504_444.0),
        ];
        ds.insert_bulk(vals, false).await?;
        ds.transaction_add("test.test.one", "etta", "ettatvåa", "xxxXXxx")
            .await?;
        let res = ds.transaction_get("test.test.one").await?;
        assert_eq!(res.len(), 1, "Expecting only one result");
        assert_eq!(
            res[0].name, "test.test.one",
            "Expecting key to match added transaction"
        );
        assert_eq!(res[0].expected, "etta", "Epected value mismatch");
        assert_eq!(res[0].target, "ettatvåa", "Target value mismatch");
        assert_eq!(res[0].t_id, 1, "Transaction ID mismatch");
        Ok(())
    }

    async fn with_keys() -> Datastore {
        let ds = Datastore::temporary().await;
        let vals = vec![
            metrc("test.test.one", "etta", 16_208_501_111.0),
            metrc("test.test.two", "etta", 16_208_501_111.0),
            metrc("test.test.three", "etta", 16_208_501_112.0),
            metrc("test.test.two", "tvåa", 16_208_502_223.0),
            metrc("test.test.three", "tvåa", 16_208_502_222.0),
            metrc("test.test.three", "trea", 16_208_503_333.0),
        ];

        ds.insert_bulk(vals, false).await.unwrap();
        ds
    }

    #[test(timeouttest)]
    async fn transmit_drop() -> TestResult {
        let ds = with_keys().await;
        ds.insert("modio.test.one", "modio-ett", 16_208_502_222.0, false)
            .await?;
        ds.insert("modio.test.one", "modio-ett", 16_208_502_222.0, true)
            .await?;
        ds.persist_data().await?;
        let to_xmit = ds.get_batch(50).await?;
        let mut to_remove = Vec::<i64>::with_capacity(50);
        for i in to_xmit {
            to_remove.push(i.id);
        }
        ds.drop_batch(&to_remove).await?;
        let second = ds.get_batch(10).await?;
        assert_eq!(second.len(), 0, "No elements should remain");
        let third = ds.get_internal_batch(5).await?;
        assert_eq!(
            third.len(),
            1,
            "Expecting an internal value, because one is timefailed."
        );
        Ok(())
    }
    #[test(timeouttest)]
    async fn timefail_handling_internal() -> TestResult {
        let ds = with_keys().await;
        ds.insert("modio.test.one", "modio-ett", 16_208_502_222.0, true)
            .await?;
        ds.insert("modio.test.one", "modio-ett", 16_208_502_221.0, true)
            .await?;
        let res = ds.get_internal_batch(10).await?;
        assert_eq!(res.len(), 0);
        ds.fix_timefail(10.0).await?;
        let res = ds.get_internal_batch(10).await?;
        assert_eq!(res.len(), 2);
        Ok(())
    }

    #[test(timeouttest)]
    async fn timefail_handling() -> TestResult {
        let ds = with_keys().await;
        ds.insert("test.test.one", "modio-ett", 1_620_850_222.0, true)
            .await?;
        ds.insert("test.test.two", "modio-två", 1_620_850_222.0, true)
            .await?;
        let res = ds.get_batch(10).await?;
        assert_eq!(res.len(), 6);
        ds.fix_timefail(10.0).await?;
        let res = ds.get_batch(10).await?;
        assert_eq!(res.len(), 8);
        Ok(())
    }

    #[test(timeouttest)]
    async fn fail_transactions() -> TestResult {
        use serde::Deserialize;
        #[derive(Deserialize)]
        struct TransJson {
            id: String,
            clock: i64,
            status: String,
        }
        let ds = with_keys().await;

        ds.transaction_add("test.test.one", "etta", "ettatvåa", "xxxXXxx")
            .await?;
        ds.transaction_add("test.test.two", "etta", "ettatvåa", "YYyyyyYYY")
            .await?;
        let first = ds.transaction_get("test.test.one").await?;
        assert_eq!(first.len(), 1, "Expecting only one result");
        assert_eq!(
            first[0].name, "test.test.one",
            "Expecting key to match added transaction"
        );
        ds.transaction_fail(first[0].t_id, false).await?;
        let logrow = ds
            .get_last_datapoint("mytemp.internal.change.test.test.one")
            .await?;
        let v1: TransJson = serde_json::from_str(&logrow.value)?;
        assert_eq!(v1.status, "FAILED");
        assert_eq!(v1.id, "xxxXXxx");
        assert!(v1.clock > 0);

        let second = ds.transaction_get("test.test.one").await?;
        assert_eq!(second.len(), 0, "Should not have pending transactions");

        let third = ds.transaction_get("test.test").await?;
        assert_eq!(third.len(), 1, "Should have pending for test.test.two");
        assert_eq!(
            third[0].name, "test.test.two",
            "Expecting key to match transaction two"
        );
        Ok(())
    }
    #[test(timeouttest)]
    async fn empty_fetch() {
        let ds = with_keys().await;
        let res = ds.get_last_datapoint("abc.def.ghi").await;
        assert!(res.is_err(), "Should have an error from absent keys");
    }

    #[test(timeouttest)]
    async fn pass_transactions() -> TestResult {
        use serde::Deserialize;
        let ds = with_keys().await;

        #[derive(Deserialize)]
        struct TransJson {
            id: String,
            clock: i64,
            status: String,
        }

        ds.transaction_add("test.test.one", "etta", "ettatvåa", "xxxXXxx")
            .await?;
        ds.transaction_add("test.test.two", "etta", "ettatvåa", "YYyyyyYYY")
            .await?;
        let first = ds.transaction_get("test.test.one").await?;

        assert_eq!(first.len(), 1, "Expecting only one result");
        assert_eq!(
            first[0].name, "test.test.one",
            "Expecting key to match added transaction"
        );
        ds.transaction_pass(first[0].t_id, false).await?;
        let logrow = ds
            .get_last_datapoint("mytemp.internal.change.test.test.one")
            .await?;
        let v1: TransJson = serde_json::from_str(&logrow.value)?;
        assert_eq!(v1.status, "SUCCESS");
        assert_eq!(v1.id, "xxxXXxx");
        assert!(v1.clock > 0);

        let second = ds.transaction_get("test.test.one").await?;
        assert_eq!(second.len(), 0, "Should not have pending transactions");

        let third = ds.transaction_get("test.test").await?;
        assert_eq!(third.len(), 1, "Should have pending for test.test.two");
        assert_eq!(
            third[0].name, "test.test.two",
            "Expecting key to match transaction two"
        );
        // this row should not exist in the database.
        let third_row = ds
            .get_last_datapoint("mytemp.internal.change.test.test.two")
            .await;
        assert!(third_row.is_err());
        Ok(())
    }

    #[test(timeouttest)]
    async fn delete_old_transactions() -> TestResult {
        let ds = with_keys().await;

        ds.transaction_add("test.test.one", "etta", "ettatvåa", "xxxXXxx")
            .await?;
        ds.transaction_add("test.test.one", "tvåa", "ettatvåa", "zzZZZzzz")
            .await?;
        ds.transaction_add("test.test.two", "etta", "ettatvåa", "YYyyyyYYY")
            .await?;

        assert_eq!(ds.count_transactions().await?, 3);
        // Get and pass the transaction
        let trans = ds.transaction_get("test.test.one").await?;
        ds.transaction_pass(trans[0].t_id, false).await?;
        ds.transaction_fail(trans[1].t_id, false).await?;

        assert_eq!(ds.count_transactions().await?, 3);

        // Submitter comes and does it's stuff here
        let to_xmit = ds.get_batch(50).await?;
        let mut to_remove = Vec::<i64>::with_capacity(5);
        for i in to_xmit {
            to_remove.push(i.id);
        }
        ds.drop_batch(&to_remove).await?;
        //  End submitter part
        assert_eq!(ds.count_transactions().await?, 3);
        // Cleanout routine should run on the device
        let count = ds.delete_old_transactions().await?;
        assert_eq!(count, 1);

        // We expect to have 2 remaining transactions still. One "PENDING" for test.test.two, and
        // one for "test.test.one" that is in "SUCCESS" or "FAIL" state.
        assert_eq!(ds.count_transactions().await?, 2);

        // Cleanout routine should run on the device
        let count = ds.delete_old_transactions().await?;
        // We should not have removed any rows now.
        assert_eq!(count, 0);
        Ok(())
    }

    #[test(timeouttest)]
    async fn fail_queued_transactions() -> TestResult {
        use uuid::Uuid;

        let ds = with_keys().await;

        for x in 0..18 {
            let tok = Uuid::new_v4().hyphenated().to_string();
            let val = format!("newval{x}");
            ds.transaction_add("test.test.one", "etta", &val, &tok)
                .await?;
        }
        ds.transaction_add("test.test.two", "etta", "ettatvåa", "xxxXXxx")
            .await?;
        ds.transaction_add("test.test.three", "tvåa", "ettatvåa", "zzZZZzzz")
            .await?;
        ds.transaction_add("test.test.four", "etta", "ettatvåa", "YYyyyyYYY")
            .await?;
        ds.transaction_add("test.test.one", "etta", "lasttarget", "xxXXxxxXXxxx")
            .await?;
        assert_eq!(ds.count_transactions().await?, 22);

        // Now run the routine to mark deep queues of transactions to failed
        let count = ds.fail_queued_transactions().await?;
        assert_eq!(count, 3, "3 should be marked as failed");

        ds.delete_old_transactions().await?;
        assert_eq!(ds.count_transactions().await?, 19, "19 should exist");

        let res = ds.transaction_get("test.test.one").await?;
        let lastval = res.last().unwrap();
        assert_eq!(
            lastval.target, "lasttarget",
            "Expecting last transaction to be the last added"
        );
        Ok(())
    }
}

mod metadata {
    use super::{raw, TagType};
    use super::{Datastore, Error};
    use crate::types::DataType;
    use crate::types::Metadata;
    use crate::types::SensorMode;
    use crate::types::ValueMap;
    use std::collections::{BTreeMap, HashMap};
    use tracing::info;
    use tracing::instrument;

    #[derive(Debug)]
    enum KeyType {
        Modio,
        Customer,
    }
    impl Datastore {
        #[instrument]
        pub async fn metadata_get_names(&self) -> Result<Vec<Metadata>, Error> {
            info!("Requested all names");
            let vals = self.metadata_get_tag(TagType::Name).await?;
            let res = vals
                .into_iter()
                .map(|(key, name)| Metadata::builder(key).name(name).build())
                .collect();
            Ok(res)
        }

        #[instrument]
        pub async fn metadata_get_units(&self) -> Result<Vec<Metadata>, Error> {
            info!("Requested all units");
            let vals = self.metadata_get_tag(TagType::Unit).await?;
            let res = vals
                .into_iter()
                .map(|(key, unit)| Metadata::builder(key).unit(unit).build())
                .collect();
            Ok(res)
        }

        #[instrument]
        pub async fn metadata_get_descriptions(&self) -> Result<Vec<Metadata>, Error> {
            info!("Requested all units");
            let vals = self.metadata_get_tag(TagType::Description).await?;
            let res = vals
                .into_iter()
                .map(|(key, description)| Metadata::builder(key).description(description).build())
                .collect();
            Ok(res)
        }

        #[instrument]
        pub async fn metadata_get_enum(&self) -> Result<Vec<Metadata>, Error> {
            info!("Requested all value maps");
            let vals = self.metadata_get_tag(TagType::Enum).await?;
            let res = vals
                .into_iter()
                .map(|(key, stringy)| Metadata::builder(key).value_map_string(stringy).build())
                .collect();
            Ok(res)
        }
        #[instrument(level = "info")]
        pub async fn get_metadata(&self, key: &str) -> Result<Option<Metadata>, Error> {
            let pairs = raw::get_metadata(&self.pool, key).await?;
            // Early exit so we don't return an empty Metadata for
            // a key that has no metadata
            if pairs.is_empty() {
                return Ok(None);
            }
            let mut builder = Metadata::builder(key);
            for (tag, value) in pairs {
                builder = builder.pair(&tag, value);
            }
            let res = builder.build();
            Ok(Some(res))
        }
        // Grab all metadata with only a pool, returns a hashmap of Key=>Metadata::Builder()
        #[instrument(level = "info")]
        async fn get_all_metadata_inner(&self, keytype: KeyType) -> Result<Vec<Metadata>, Error> {
            let mut map = HashMap::new();
            let mut offset: u32 = 0;
            info!("Requested all metadata from DB");
            loop {
                let pairs = match keytype {
                    KeyType::Modio => raw::get_all_metadata_internal(&self.pool, offset).await?,
                    KeyType::Customer => raw::get_all_metadata_customer(&self.pool, offset).await?,
                };
                if pairs.is_empty() {
                    // No more data
                    break;
                }
                offset += pairs.len() as u32;
                for val in pairs {
                    // If there is a matching builder in the hashmap, use it
                    let mut builder = map
                        .remove(&val.key)
                        // Or create if absent.
                        .unwrap_or_else(|| Metadata::builder(&val.key));
                    builder = builder.pair(&val.tag, val.value);
                    map.insert(val.key, builder);
                }
            }
            let res: Vec<Metadata> = map.drain().map(|(_, builder)| builder.build()).collect();
            Ok(res)
        }

        #[instrument(level = "debug")]
        pub async fn get_all_metadata(&self) -> Result<Vec<Metadata>, Error> {
            let res = self.get_all_metadata_inner(KeyType::Customer).await?;
            Ok(res)
        }

        #[instrument(level = "debug")]
        pub async fn get_all_internal_metadata(&self) -> Result<Vec<Metadata>, Error> {
            let res = self.get_all_metadata_inner(KeyType::Modio).await?;
            Ok(res)
        }
        /// Set a tag. Replace it if it was already set
        ///
        /// Returns bool, truth means it was set,
        /// false means it was already set to that value and nothing changed
        #[instrument(level = "info")]
        async fn metadata_replace_tag(
            &self,
            key: &str,
            tag: TagType,
            value: &str,
        ) -> Result<bool, Error> {
            // Check if we can just skip the entire thing.
            if self.metadata_tag_equals(key, tag, value).await {
                return Ok(false);
            }
            // Ensure the name is stored first.
            self.add_sensor(key).await?;
            raw::metadata_replace_tag(&self.pool, key, tag, value).await?;
            Ok(true)
        }
        /// Set a tag. fail if it is already set.
        #[instrument(level = "info")]
        async fn metadata_set_tag(
            &self,
            key: &str,
            tag: TagType,
            value: &str,
        ) -> Result<(), Error> {
            self.add_sensor(key).await?;
            raw::metadata_set_tag(&self.pool, key, tag, value).await?;
            Ok(())
        }
        #[instrument(level = "info")]
        async fn metadata_get_tag(&self, tag: TagType) -> Result<Vec<(String, String)>, Error> {
            let res = raw::metadata_get_tag(&self.pool, tag).await?;
            Ok(res)
        }
        // Returns an error if the key doesnt exist.
        #[instrument(level = "info")]
        async fn metadata_get_single_tag(&self, key: &str, tag: TagType) -> Result<String, Error> {
            let res = raw::metadata_get_single_tag(&self.pool, key, tag).await?;
            Ok(res)
        }

        /// Sometimes we just want to check if the tags are already in place and look the same.
        /// This function treats errors as "not equal" and hopes you deal with it.
        async fn metadata_tag_equals(&self, key: &str, tag: TagType, value: &str) -> bool {
            self.metadata_get_single_tag(key, tag)
                .await
                .is_ok_and(|old_val| old_val == value)
        }

        /// Set the name for a sensor
        /// If the name was already set, replaces it.
        pub async fn metadata_set_name(&self, key: &str, name: &str) -> Result<bool, Error> {
            self.metadata_replace_tag(key, TagType::Name, name).await
        }

        /// Set the unit for an key.
        /// If the unit is already set, will return some kind of Error.
        /// Returns true if it wrote to database, false if not
        pub async fn metadata_set_unit(&self, key: &str, unit: &str) -> Result<bool, Error> {
            // Take a peek if we already have a unit for this key first
            if let Some(meta) = self.get_metadata(key).await? {
                if let Some(u) = meta.u {
                    if u == unit {
                        // The two are equal, return early rather than cause a Unique error.
                        return Ok(false);
                    }
                }
            }
            self.metadata_set_tag(key, TagType::Unit, unit).await?;
            Ok(true)
        }

        /// Set the description for a sensor
        /// If the description was already set, replaces it.
        #[instrument]
        pub async fn metadata_set_description(
            &self,
            key: &str,
            description: &str,
        ) -> Result<bool, Error> {
            self.metadata_replace_tag(key, TagType::Description, description)
                .await
        }

        /// Set the mode for a sensor. (`ReadOnly`, `ReadWrite`, `WriteOnly`)
        /// If the mode was already set, replaces it.
        #[instrument]
        pub async fn metadata_set_mode(&self, key: &str, mode: &SensorMode) -> Result<bool, Error> {
            self.metadata_replace_tag(key, TagType::Mode, mode.as_str())
                .await
        }

        /// Set the enum (lookup table)  for a sensor
        /// If the enum was already set, replaces it.
        #[instrument]
        pub async fn metadata_set_enum(
            &self,
            key: &str,
            value_map: &ValueMap,
        ) -> Result<bool, Error> {
            let into = {
                // Convert to a BTreeMap to guarantee order of values
                // This will create a map of <&u32, &String>, which is fine for use with serde_json,
                // but is in general a data-structure with a nightmare of a lifetime
                let sorted_map: BTreeMap<_, _> = value_map.iter().collect();
                serde_json::to_string(&sorted_map)?
            };
            self.metadata_replace_tag(key, TagType::Enum, &into).await
        }

        /// Set the key as a row
        /// Row data is expected to be an Array of Datatype that will then be stored and parsed
        /// from strings (json)
        /// Length of the array must match the incoming data for this key.
        /// Returns true if it wrote to database, false if not
        #[instrument]
        pub async fn metadata_set_row(
            &self,
            key: &str,
            row_types: Vec<DataType>,
        ) -> Result<bool, Error> {
            let rowdata = serde_json::to_string(&row_types)?;
            // If we have row metadata, and it's the same, we return equality.
            // The logic is similar as for Unit,  we want to allow _set_ and set again with the
            // same value, but we forbid changing it.
            if let Ok(old_row) = self.metadata_get_single_tag(key, TagType::Row).await {
                if old_row == rowdata {
                    return Ok(false);
                }
            }
            self.metadata_set_tag(key, TagType::Row, &rowdata).await?;
            Ok(true)
        }

        // Check if the name exists in the sensor table and is a bulk-looking key
        #[instrument]
        pub async fn metadata_get_row(&self, key: &str) -> Result<Vec<DataType>, Error> {
            let value = self.metadata_get_single_tag(key, TagType::Row).await?;
            let res = DataType::vec_from_str(&value)?;
            Ok(res)
        }
    }

    #[cfg(test)]
    mod tests {
        use super::*;
        use std::error::Error as StdError;
        use test_log::test;
        use timeout_macro::timeouttest;
        type TestResult = Result<(), Box<dyn StdError>>;

        #[test(timeouttest)]
        async fn get_empty_metadata() -> TestResult {
            let ds = Datastore::temporary().await;
            ds.insert("modio.test.key", "one", 1_620_850_000.0, false)
                .await?;
            ds.insert("public.test.key", "two", 1_620_850_000.0, false)
                .await?;

            let res = ds.metadata_get_names().await?;
            assert!(res.is_empty(), "should be empty");

            let res = ds.get_metadata("modio.test.key").await?;
            assert!(res.is_none(), "should not exist");
            Ok(())
        }

        #[test(timeouttest)]
        async fn get_metadata_name() -> TestResult {
            let ds = Datastore::temporary().await;

            // First time should be Ok(true)
            let res = ds
                .metadata_set_name("modio.test.key", "Modio Test Key")
                .await?;
            assert!(res);

            // Second time should be Ok(false)
            let res = ds
                .metadata_set_name("modio.test.key", "Modio Test Key")
                .await?;
            assert!(!res);

            let res = ds.metadata_get_names().await?;
            assert_eq!(res.len(), 1, "should have one key");
            assert_eq!(res[0].name, Some("Modio Test Key".into()), "Should match");
            assert_eq!(res[0].n, "modio.test.key", "Should be our key");

            ds.metadata_set_name("modio.test.key", "Modio Test Key Two")
                .await?;
            let res = ds.metadata_get_names().await?;
            assert_eq!(res.len(), 1, "should have one key");
            assert_eq!(
                res[0].name,
                Some("Modio Test Key Two".into()),
                "Should match"
            );
            Ok(())
        }
        #[test(timeouttest)]
        async fn get_metadata_description() -> TestResult {
            let ds = Datastore::temporary().await;

            ds.metadata_set_description("modio.test.key", "Modio Test description")
                .await?;
            let res = ds.metadata_get_descriptions().await?;
            assert_eq!(res.len(), 1, "should have one key");
            assert_eq!(
                res[0].description,
                Some("Modio Test description".into()),
                "Should match"
            );
            assert_eq!(res[0].n, "modio.test.key", "Should be our key");

            ds.metadata_set_description(
                "modio.test.key",
                "The second  update is to change the description",
            )
            .await?;
            let res = ds.metadata_get_descriptions().await?;
            assert_eq!(res.len(), 1, "should have one key");
            Ok(())
        }

        #[test(timeouttest)]
        async fn get_metadata_unit() -> TestResult {
            let ds = Datastore::temporary().await;

            ds.metadata_set_unit("modio.test.key", "Cel").await?;

            let res = ds.metadata_get_units().await?;
            assert_eq!(res.len(), 1, "should be one item");
            assert_eq!(res[0].u, Some("Cel".into()), "Should be Celsius");
            assert_eq!(res[0].n, "modio.test.key", "Should be our key");

            let status = ds.metadata_set_unit("modio.test.key", "m").await;
            assert!(status.is_err(), "Should not be able to replace unit");
            let res = ds.metadata_get_units().await?;
            assert_eq!(res[0].u, Some("Cel".into()), "Should still be Celsius");
            Ok(())
        }

        #[test(timeouttest)]
        async fn set_unit_unique() -> TestResult {
            let ds = Datastore::temporary().await;
            ds.metadata_set_unit("modio.test.key", "Cel").await?;
            // Calling it again with the same unit should work
            ds.metadata_set_unit("modio.test.key", "Cel").await?;

            let err = ds
                .metadata_set_unit("modio.test.key", "m")
                .await
                .expect_err("Should get unique constraint failed");
            assert_eq!(err.to_string(), "Unique constraint failed");
            Ok(())
        }

        #[test(timeouttest)]
        async fn get_metadata_enum() -> TestResult {
            let ds = Datastore::temporary().await;

            let value_map = ValueMap::from([
                (0, "error".to_string()),
                (1, "enabled".to_string()),
                (2, "disabled".to_string()),
            ]);
            ds.metadata_set_enum("modio.test.key", &value_map).await?;
            let mut res = ds.metadata_get_enum().await?;
            assert_eq!(res.len(), 1, "Should have one value");
            assert_eq!(res[0].n, "modio.test.key");
            assert!(res[0].value_map.is_some());
            let entry = res.pop().unwrap();
            let vmap = entry.value_map.unwrap();
            assert_eq!(vmap.get(&0).unwrap(), &"error".to_string());
            assert_eq!(vmap.get(&1).unwrap(), &"enabled".to_string());
            assert_eq!(vmap.get(&2).unwrap(), &"disabled".to_string());

            let humm = ds.get_metadata("modio.test.key").await?;
            assert!(humm.is_some());
            let humm = humm.unwrap();
            assert_eq!(humm.n, "modio.test.key");
            assert!(humm.value_map.is_some());
            // Unpacking it again should also look the same as above.
            let vmap = humm.value_map.unwrap();
            assert_eq!(vmap.get(&0).unwrap(), &"error".to_string());
            assert_eq!(vmap.get(&1).unwrap(), &"enabled".to_string());
            assert_eq!(vmap.get(&2).unwrap(), &"disabled".to_string());

            Ok(())
        }

        #[test(timeouttest)]
        async fn test_get_all_internal_metadata() -> TestResult {
            let ds = Datastore::temporary().await;
            let value_map = ValueMap::from([
                (0, "error".to_string()),
                (1, "enabled".to_string()),
                (2, "disabled".to_string()),
            ]);
            ds.metadata_set_name("customer.data", "Customer name")
                .await?;
            ds.metadata_set_name("modio.test.dupe", "Modio Test Another Key")
                .await?;
            ds.metadata_set_enum("modio.test.key", &value_map).await?;
            ds.metadata_set_name("modio.test.key", "Modio Test Key")
                .await?;
            ds.metadata_set_unit("modio.test.key", "Cel").await?;
            ds.metadata_set_description("modio.test.key", "Our Description")
                .await?;
            let res = ds.get_all_internal_metadata().await?;
            assert_eq!(res.len(), 2, "should have one key");

            let mut filt: Vec<Metadata> = res
                .into_iter()
                .filter(|x| x.n == "modio.test.key")
                .collect();
            let obj = filt.pop().unwrap();
            assert_eq!(obj.n, "modio.test.key");
            assert_eq!(obj.name, Some("Modio Test Key".into()));
            assert_eq!(obj.description, Some("Our Description".into()));
            assert_eq!(obj.u, Some("Cel".into()));
            assert_eq!(obj.value_map.unwrap().get(&0), Some(&"error".to_string()));

            let mut cust = ds.get_all_metadata().await?;
            assert_eq!(cust.len(), 1, "should have one key");
            let c_item = cust.pop().unwrap();
            assert_eq!(c_item.n, "customer.data", "Should be a customer key");
            assert_eq!(
                c_item.name,
                Some("Customer name".into()),
                "Should have a customer name"
            );
            Ok(())
        }
    }
}